package main

import (
	"context"
	"encoding/json"
	"log"
	"log/slog"
	"net/http"
	"sync"

	"github.com/bluesky-social/indigo/api/bsky"
	"github.com/bluesky-social/indigo/xrpc"
	"github.com/bluesky-social/jetstream/pkg/client"
	"github.com/bluesky-social/jetstream/pkg/models"
	"github.com/gorilla/mux"
	"github.com/gorilla/websocket"
)

type Set[T comparable] map[T]struct{}

// Data structures
type SubscriberData struct {
	DID      string
	Conn     *websocket.Conn
	ListenTo Set[string]
	Reposts  Set[string]
}

type NotificationMessage struct {
	Liked     bool   `json:"liked"`
	ByDid     string `json:"did"`
	RepostURI string `json:"repost_uri"`
}

// Global state
var (
	subscribers    = make(map[string]*SubscriberData)
	subscribersMux sync.RWMutex

	likeStream       *client.Client
	subscriberStream *client.Client

	upgrader = websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}

	logger *slog.Logger
)

func getFollowsDids() []string {
	subscribersMux.RLock()
	defer subscribersMux.RUnlock()

	var dids []string
	for _, subscriber := range subscribers {
		for follow, _ := range subscriber.ListenTo {
			dids = append(dids, follow)
		}
	}

	return dids
}

func getSubscriberDids() []string {
	subscribersMux.RLock()
	defer subscribersMux.RUnlock()

	var dids []string
	for did := range subscribers {
		dids = append(dids, did)
	}

	return dids
}

func main() {
	logger = slog.Default()

	go likeStreamLoop(logger)
	go subscriberStreamLoop(logger)

	r := mux.NewRouter()
	r.HandleFunc("/subscribe/{did}", handleSubscribe).Methods("GET")

	log.Println("Server starting on :8080")
	if err := http.ListenAndServe(":8080", r); err != nil {
		log.Fatalf("error while serving: %s", err)
	}
}

func handleSubscribe(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	did := vars["did"]

	logger = logger.With("did", did)

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		logger.Error("WebSocket upgrade failed", "error", err)
		return
	}
	defer conn.Close()

	logger.Info("new subscriber")

	pdsURI, err := findUserPDS(r.Context(), did)
	if err != nil {
		logger.Error("cant resolve user pds", "error", err)
		return
	}
	logger = logger.With("pds", pdsURI)

	xrpcClient := &xrpc.Client{
		Host: pdsURI,
	}
	// todo: implement skipping fetching follows and allow specifying users to listen to via websocket
	follows, err := fetchFollows(r.Context(), xrpcClient, did)
	if err != nil {
		logger.Error("error fetching follows", "error", err)
		return
	}
	logger.Info("fetched follows")
	reposts, err := fetchReposts(r.Context(), xrpcClient, did)
	if err != nil {
		logger.Error("error fetching reposts", "error", err)
		return
	}
	logger.Info("fetched reposts")

	subscriber := &SubscriberData{
		DID:  did,
		Conn: conn,
		// use user follows as default listen to
		ListenTo: follows,
		Reposts:  reposts,
	}

	subscribersMux.Lock()
	subscribers[did] = subscriber
	subscribersMux.Unlock()
	updateSubscriberStreamOpts()
	updateLikeStreamOpts()
	// delete subscriber after we are done
	defer func() {
		subscribersMux.Lock()
		delete(subscribers, did)
		subscribersMux.Unlock()
		updateSubscriberStreamOpts()
		updateLikeStreamOpts()
	}()

	logger.Info("serving subscriber")

	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			logger.Info("WebSocket connection closed", "error", err)
			break
		}
	}
}

func getLikeStreamOpts() models.SubscriberOptionsUpdatePayload {
	return models.SubscriberOptionsUpdatePayload{
		WantedCollections: []string{"app.bsky.feed.like"},
		WantedDIDs:        getFollowsDids(),
	}
}

func getSubscriberStreamOpts() models.SubscriberOptionsUpdatePayload {
	return models.SubscriberOptionsUpdatePayload{
		WantedCollections: []string{"app.bsky.feed.repost", "app.bsky.graph.follow"},
		WantedDIDs:        getSubscriberDids(),
	}
}

func updateLikeStreamOpts() {
	opts := getLikeStreamOpts()
	err := likeStream.SendOptionsUpdate(opts)
	if err != nil {
		logger.Error("couldnt update like stream opts", "error", err)
		return
	}
	logger.Info("updated like stream opts", "requestedDids", len(opts.WantedDIDs))
}

func updateSubscriberStreamOpts() {
	opts := getSubscriberStreamOpts()
	err := subscriberStream.SendOptionsUpdate(opts)
	if err != nil {
		logger.Error("couldnt update subscriber stream opts", "error", err)
		return
	}
	logger.Info("updated subscriber stream opts", "userCount", len(opts.WantedDIDs))
}

func likeStreamLoop(logger *slog.Logger) {
	startJetstreamLoop(logger, &likeStream, "like_tracker", HandleLikeEvent, getLikeStreamOpts)
}

func subscriberStreamLoop(logger *slog.Logger) {
	startJetstreamLoop(logger, &subscriberStream, "subscriber", HandleSubscriberEvent, getSubscriberStreamOpts)
}

func HandleLikeEvent(ctx context.Context, event *models.Event) error {
	if event == nil || event.Commit == nil || len(event.Commit.Record) == 0 {
		return nil
	}

	var like bsky.FeedLike
	if err := json.Unmarshal(event.Commit.Record, &like); err != nil {
		logger.Error("Failed to unmarshal like", "error", err)
		return nil
	}

	subscribersMux.RLock()
	defer subscribersMux.RUnlock()

	for _, subscriber := range subscribers {
		for repostURI, _ := range subscriber.Reposts {
			// (un)liked a post the subscriber reposted
			if like.Subject.Uri == repostURI {
				notification := NotificationMessage{
					Liked:     event.Commit.Operation != models.CommitOperationDelete,
					ByDid:     event.Did,
					RepostURI: repostURI,
				}

				if err := subscriber.Conn.WriteJSON(notification); err != nil {
					logger.Error("Failed to send notification", "subscriber", subscriber.DID, "error", err)
				}
			}
		}
	}

	return nil
}

func HandleSubscriberEvent(ctx context.Context, event *models.Event) error {
	if event == nil || event.Commit == nil {
		return nil
	}

	switch event.Commit.Collection {
	case "app.bsky.feed.repost":
		modifySubscribersWithEvent(
			event,
			func(s *SubscriberData, r bsky.FeedRepost) { delete(s.Reposts, r.Subject.Uri) },
			func(s *SubscriberData, r bsky.FeedRepost) {
				s.Reposts[r.Subject.Uri] = struct{}{}
			},
		)
	case "app.bsky.graph.follow":
		modifySubscribersWithEvent(
			event,
			func(s *SubscriberData, r bsky.GraphFollow) { delete(s.ListenTo, r.Subject) },
			func(s *SubscriberData, r bsky.GraphFollow) {
				s.ListenTo[r.Subject] = struct{}{}
			},
		)
	}

	return nil
}

type ModifyFunc[v any] func(*SubscriberData, v)

func modifySubscribersWithEvent[v any](event *models.Event, onDelete ModifyFunc[v], onUpdate ModifyFunc[v]) error {
	if len(event.Commit.Record) == 0 {
		return nil
	}

	var data v
	if err := json.Unmarshal(event.Commit.Record, &data); err != nil {
		logger.Error("Failed to unmarshal repost", "error", err, "raw", event.Commit.Record)
		return nil
	}

	subscribersMux.Lock()
	defer subscribersMux.Unlock()

	if subscriber, exists := subscribers[event.Did]; exists {
		if event.Commit.Operation == models.CommitOperationDelete {
			onDelete(subscriber, data)
		} else {
			onUpdate(subscriber, data)
		}
	}

	return nil
}