package knotstream

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"net/url"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/bluesky-social/indigo/util/ssrf"
	"github.com/carlmjohnson/versioninfo"
	"github.com/gorilla/websocket"
	"tangled.org/core/api/tangled"
	"tangled.org/core/eventconsumer"
	"tangled.org/core/eventstream"
	"tangled.org/core/knotmirror/config"
	"tangled.org/core/knotmirror/db"
	"tangled.org/core/knotmirror/models"
	"tangled.org/core/log"
)

type KnotSlurper struct {
	logger *slog.Logger
	db     *sql.DB
	ssrf   bool

	hostsMu sync.RWMutex
	hosts   map[string]models.Host

	consumer *eventconsumer.Consumer

	ctxMu   sync.Mutex
	rootCtx context.Context
}

func NewKnotSlurper(l *slog.Logger, sqlDB *sql.DB, cfg *config.Config) *KnotSlurper {
	logger := log.SubLogger(l, "slurper")
	s := &KnotSlurper{
		logger: logger,
		db:     sqlDB,
		ssrf:   cfg.KnotSSRF,
		hosts:  make(map[string]models.Host),
	}

	dialer := &websocket.Dialer{
		HandshakeTimeout: 5 * time.Second,
		NetDialContext:   s.dialContext,
	}

	store := &hostsCursorStore{db: sqlDB, logger: logger}

	headers := http.Header{}
	headers.Set("User-Agent", userAgent())

	s.consumer = eventconsumer.NewConsumer(eventconsumer.ConsumerConfig{
		Logger:            logger,
		ProcessFunc:       s.processEvent,
		Dialer:            dialer,
		URLFunc:           s.urlFor,
		RequestHeader:     headers,
		CursorStore:       store,
		MaxRetryAttempts:  30,
		OnConnectExceeded: s.banHost,
		RetryInterval:     time.Second,
		MaxRetryInterval:  30 * time.Second,
		ConnectionTimeout: 5 * time.Second,
		WorkerCount:       cfg.Slurper.ConcurrencyPerHost,
		QueueSize:         1024,
	})
	return s
}

func (s *KnotSlurper) Start(ctx context.Context) {
	s.ctxMu.Lock()
	s.rootCtx = ctx
	s.ctxMu.Unlock()
	s.consumer.Start(ctx)
}

func (s *KnotSlurper) Shutdown(ctx context.Context) error {
	s.consumer.Stop()
	s.logger.Info("slurper shutdown complete")
	return nil
}

func (s *KnotSlurper) CheckIfSubscribed(hostname string) bool {
	s.hostsMu.RLock()
	defer s.hostsMu.RUnlock()
	_, ok := s.hosts[hostname]
	return ok
}

func (s *KnotSlurper) Subscribe(host models.Host) error {
	s.hostsMu.Lock()
	if _, ok := s.hosts[host.Hostname]; ok {
		s.hostsMu.Unlock()
		return fmt.Errorf("already subscribed: %s", host.Hostname)
	}
	s.hosts[host.Hostname] = host
	s.hostsMu.Unlock()

	connectedInbound.Inc()
	s.consumer.AddSource(s.subscribeCtx(), eventconsumer.NewKnotSource(host.Hostname))
	return nil
}

func (s *KnotSlurper) subscribeCtx() context.Context {
	s.ctxMu.Lock()
	defer s.ctxMu.Unlock()
	if s.rootCtx != nil {
		return s.rootCtx
	}
	return context.Background()
}

func (s *KnotSlurper) banHost(src eventconsumer.Source, err error) {
	s.logger.Warn("host does not appear to be online, disabling", "host", src.Host, "err", err)
	connectedInbound.Dec()

	s.hostsMu.Lock()
	h, ok := s.hosts[src.Host]
	if ok {
		delete(s.hosts, src.Host)
	}
	s.hostsMu.Unlock()
	if !ok {
		return
	}
	h.Status = models.HostStatusOffline
	if err := db.UpsertHost(context.Background(), s.db, &h); err != nil {
		s.logger.Error("failed to update host status", "err", err)
	}
	s.consumer.RemoveSource(src)
}

func (s *KnotSlurper) dialContext(ctx context.Context, network, addr string) (net.Conn, error) {
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		host = addr
	}
	s.hostsMu.RLock()
	h, ok := s.hosts[host]
	s.hostsMu.RUnlock()
	useSSRF := s.ssrf || !ok || !h.NoSSL
	if useSSRF {
		return ssrf.PublicOnlyDialer().DialContext(ctx, network, addr)
	}
	return (&net.Dialer{}).DialContext(ctx, network, addr)
}

func (s *KnotSlurper) urlFor(src eventconsumer.Source, cursor string) (*url.URL, error) {
	s.hostsMu.RLock()
	h, ok := s.hosts[src.Host]
	s.hostsMu.RUnlock()
	scheme := "wss"
	if ok && h.NoSSL {
		scheme = "ws"
	}
	u, err := url.Parse(scheme + "://" + src.Host + "/events")
	if err != nil {
		return nil, err
	}
	if cursor != "" {
		q := url.Values{}
		q.Add("cursor", cursor)
		u.RawQuery = q.Encode()
	}
	return u, nil
}

type legacyGitRefUpdate struct {
	OwnerDid      *string `json:"ownerDid,omitempty"`
	Repo          *string `json:"repo,omitempty"`
	LegacyRepoDid *string `json:"repoDid,omitempty"`
}

func (s *KnotSlurper) processEvent(ctx context.Context, src eventconsumer.Source, ev eventstream.Event) error {
	if ev.Nsid != tangled.GitRefUpdateNSID {
		knotstreamEventsSkipped.Inc()
		return nil
	}
	return s.processLegacyGitRefUpdate(ctx, src.Host, &ev)
}

// lookupRepoForRefUpdate resolves the local repo row for an incoming refUpdate
// via the stable RepoDid join. Returns (nil, "", nil) when the event has no
// repoDid (unjoinable) and (nil, key, nil) on a clean miss.
func (s *KnotSlurper) lookupRepoForRefUpdate(ctx context.Context, payload legacyGitRefUpdate) (*models.Repo, string, error) {
	raw := payload.Repo
	if raw == nil || *raw == "" {
		raw = payload.LegacyRepoDid
	}
	if raw == nil || *raw == "" {
		return nil, "", nil
	}
	repoDid := syntax.DID(*raw)
	curr, err := db.GetRepoByRepoDid(ctx, s.db, repoDid)
	return curr, repoDid.String(), err
}

func (s *KnotSlurper) processLegacyGitRefUpdate(ctx context.Context, source string, evt *eventstream.Event) error {
	knotstreamEventsReceived.Inc()

	l := s.logger.With("src", source)

	var payload legacyGitRefUpdate
	if err := json.Unmarshal(evt.EventJson, &payload); err != nil {
		return fmt.Errorf("decoding gitRefUpdate payload: %w", err)
	}

	curr, lookupKey, err := s.lookupRepoForRefUpdate(ctx, payload)
	if err != nil {
		return fmt.Errorf("failed to get repo '%s': %w", lookupKey, err)
	}
	if curr == nil {
		if lookupKey == "" {
			l.Warn("skipping gitRefUpdate: event has no fields to join on",
				"repo", payload.Repo, "legacy_repo_did", payload.LegacyRepoDid)
		} else {
			// if repo doesn't exist in DB, just ignore the event. That repo is unknown.
			// Hopefully crawler/tap will sync it later.
			l.Warn("skipping event from unknown repo", "key", lookupKey)
		}
		knotstreamEventsSkipped.Inc()
		return nil
	}
	l = l.With("repoAt", curr.AtUri())

	// TODO: should plan resync to resyncBuffer on RepoStateResyncing
	if curr.State != models.RepoStateActive {
		l.Debug("skipping non-active repo")
		knotstreamEventsSkipped.Inc()
		return nil
	}

	if curr.GitRev != "" && evt.Rkey <= curr.GitRev.String() {
		l.Debug("skipping replayed event", "event.Rkey", evt.Rkey, "currentRev", curr.GitRev)
		knotstreamEventsSkipped.Inc()
		return nil
	}

	if err := db.UpdateRepoState(ctx, s.db, curr.RepoDid, models.RepoStateDesynchronized); err != nil {
		return err
	}

	l.Info("event processed", "eventRev", evt.Rkey)

	knotstreamEventsProcessed.Inc()
	return nil
}

func userAgent() string {
	return fmt.Sprintf("knotmirror/%s", versioninfo.Short())
}
