package deliberi

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"slices"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/posthog/posthog-go"
	"tangled.org/core/deliberi/analytics"
	deldb "tangled.org/core/deliberi/db"
	"tangled.org/core/deliberi/kv"
	models "tangled.org/core/deliberi/models"
	"tangled.org/core/idresolver"
	"tangled.org/core/jetstream"
	"tangled.org/core/sets"
)

const (
	maxMentions     = 8
	idleCursorEvery = time.Minute
	// rewinding replays events, the rev check in plan drops the repeats
	cursorRewind = 5 * time.Minute
)

type Ingester struct {
	db         *deldb.DB
	bobbin     bobbinLookup
	idResolver *idresolver.Resolver
	endpoint   string
	logger     *slog.Logger
	// nil when analytics is off
	analytics *analytics.Client
	// nil when the KV cache is off
	emailKV kv.EmailKV

	idleCursorUs int64
}

var ingestCollections = []string{
	repoManifestNSID,
	ticketNSID,
	commentNSID,
	patchNSID,
	collaboratorNSID,
	subscriptionNSID,
	starNSID,
	followNSID,
}

// star and follow stay out, their handlers only notify on live creates
var revGuarded = []string{
	repoManifestNSID,
	ticketNSID,
	commentNSID,
	patchNSID,
	collaboratorNSID,
	subscriptionNSID,
}

func NewIngester(database *deldb.DB, bobbin bobbinLookup, idRes *idresolver.Resolver, endpoint string, logger *slog.Logger) *Ingester {
	return &Ingester{
		db:         database,
		bobbin:     bobbin,
		idResolver: idRes,
		endpoint:   endpoint,
		logger:     logger,
	}
}

func (i *Ingester) Run(ctx context.Context) error {
	stream, err := jetstream.NewStream(jetstream.StreamConfig{
		Endpoint:    i.endpoint,
		Collections: ingestCollections,
		// hydrant only sends backfilled commits when historical is asked for
		EventTypes: []string{"live", "historical"},
		Cursor:     func() (int64, bool, error) { return deldb.GetCursor(i.db) },
		Rewind:     cursorRewind,
		UserAgent:  "deliberi",
	}, i.logger)
	if err != nil {
		return err
	}
	return stream.Run(ctx, i.process)
}

// runs in the event tx, so network calls belong in plan
type writer func(x deldb.Execer) error

func (i *Ingester) process(ctx context.Context, e *jetstream.Event) error {
	if e.Kind == jetstream.EventKindAccount && e.Account != nil {
		return i.processAccount(e)
	}
	var write writer
	if e.Kind == jetstream.EventKindCommit && e.Commit != nil {
		write = i.plan(ctx, e)
	}
	if write == nil {
		return i.advance(e.TimeUS)
	}

	tx, err := i.db.Begin()
	if err != nil {
		return fmt.Errorf("starting event tx: %w", err)
	}
	defer tx.Rollback()

	if err := write(tx); err != nil {
		return err
	}
	if err := deldb.SaveCursor(tx, e.TimeUS); err != nil {
		return fmt.Errorf("saving cursor: %w", err)
	}
	return tx.Commit()
}

// only "deleted" is final, deactivated and takendown accounts can come back
func (i *Ingester) processAccount(e *jetstream.Event) error {
	if e.Account.Active || e.Account.Status == nil || *e.Account.Status != "deleted" {
		return i.advance(e.TimeUS)
	}
	tx, err := i.db.Begin()
	if err != nil {
		return fmt.Errorf("starting account tx: %w", err)
	}
	defer tx.Rollback()
	emails, err := deldb.DeleteAccountEmails(tx, e.Did)
	if err != nil {
		return fmt.Errorf("deleting emails for %s: %w", e.Did, err)
	}
	if err := deldb.SaveCursor(tx, e.TimeUS); err != nil {
		return fmt.Errorf("saving cursor: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	if len(emails) > 0 {
		i.logger.Info("removed emails of deleted account", "did", e.Did, "count", len(emails))
	}
	if i.emailKV != nil {
		for _, email := range emails {
			i.emailKV.DeleteEmailKey(email)
		}
		i.emailKV.DeletePrimaryEmail(e.Did)
	}
	return nil
}

func (i *Ingester) advance(timeUs int64) error {
	if timeUs-i.idleCursorUs < idleCursorEvery.Microseconds() {
		return nil
	}
	i.idleCursorUs = timeUs
	return deldb.SaveCursor(i.db, timeUs)
}

func (i *Ingester) plan(ctx context.Context, e *jetstream.Event) writer {
	c := e.Commit
	uri := fmt.Sprintf("at://%s/%s/%s", e.Did, c.Collection, c.RKey)

	guarded := c.Rev != "" && slices.Contains(revGuarded, c.Collection)
	if guarded && !deldb.RevIsNew(i.db, uri, c.Rev) {
		return nil
	}

	notify := i.notifying(e)
	var write writer
	switch c.Collection {
	case repoManifestNSID:
		write = i.planManifest(e, uri)
	case ticketNSID:
		write = i.planTicket(ctx, e, uri, notify)
	case commentNSID:
		write = i.planComment(ctx, e, uri, notify)
	case patchNSID:
		write = i.planPatch(e, uri)
	case collaboratorNSID:
		write = i.planCollaborator(e)
	case subscriptionNSID:
		write = i.planSubscription(e, uri)
	case starNSID:
		write = i.planStar(ctx, e, uri, notify)
	case followNSID:
		write = i.planFollow(e, uri, notify)
	}
	if write == nil || !guarded {
		return write
	}
	return func(x deldb.Execer) error {
		if err := deldb.PutRev(x, uri, c.Rev); err != nil {
			return err
		}
		return write(x)
	}
}

// bootstrap time is the cutoff, anything older only builds state
func (i *Ingester) notifying(e *jetstream.Event) bool {
	return e.Commit.IsLive() && e.TimeUS >= deldb.GetBootstrapUs(i.db)
}

func (i *Ingester) planManifest(e *jetstream.Event, uri string) writer {
	if e.Commit.Operation == jetstream.CommitOperationDelete || len(e.Commit.Record) == 0 {
		return nil
	}
	var rec repoManifest
	if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
		i.logger.Warn("decoding manifest record", "err", err, "uri", uri)
		return nil
	}
	if rec.Declaration == nil || rec.Declaration.Owner == "" {
		return nil
	}
	return func(x deldb.Execer) error {
		return deldb.PutRepoName(x, e.Did, rec.Declaration.Owner, rec.Declaration.Name)
	}
}

func (i *Ingester) planTicket(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
	c := e.Commit
	if c.Operation == jetstream.CommitOperationDelete {
		return func(x deldb.Execer) error {
			if err := deldb.DeleteEntity(x, uri); err != nil {
				return err
			}
			return deldb.DeleteNotificationsAbout(x, uri)
		}
	}
	if len(c.Record) == 0 {
		return nil
	}
	var rec ticketRecord
	if err := json.Unmarshal(c.Record, &rec); err != nil {
		i.logger.Warn("decoding ticket record", "err", err, "uri", uri)
		return nil
	}

	actor := editorOr(rec.Editor, e.Did)
	known := deldb.GetEntity(i.db, uri)
	next := deldb.Entity{
		Title:   rec.title(),
		RepoDid: e.Did,
		Kind:    rec.kind(),
		State:   rec.state(),
		Cid:     c.CID,
		Author:  known.Author,
	}
	// a backfill sends the latest version as a create, and a maintainer may have
	// written that one, so only a live create names who opened the ticket
	opened := c.IsLive() && c.Operation == jetstream.CommitOperationCreate && known.Kind == ""
	if opened {
		next.Author = actor
	}
	store := func(x deldb.Execer) error { return deldb.PutEntity(x, uri, next) }
	if !notify {
		return store
	}
	i.fillOpener(ctx, uri, &next)

	var kind models.NotificationType
	switch {
	case opened:
		kind = models.NotificationTypeIssueCreated
		if next.Kind == models.EntityKindPull {
			kind = models.NotificationTypePullCreated
		}
	case known.State != "":
		kind = stateNotification(next.Kind, known.State, next.State)
	}
	// todo: assigned and unassigned, once the knot takes assignees
	if kind == "" {
		return store
	}

	owner := i.repoOwner(ctx, e.Did)
	var mentions []string
	if opened {
		mentions = i.mentionDids(ctx, actor, bodyText(rec.Body))
	}
	return func(x deldb.Execer) error {
		if err := store(x); err != nil {
			return err
		}
		collaborators, err := deldb.ListCollaborators(x, e.Did)
		if err != nil {
			return err
		}
		recipients := append([]string{owner}, collaborators...)
		n := models.Notification{
			AtUri:       uri,
			Type:        kind,
			ActorDid:    actor,
			RepoDid:     e.Did,
			EntityAt:    uri,
			EntityKind:  next.Kind,
			EntityTitle: next.Title,
			Created:     time.UnixMicro(e.TimeUS).UTC(),
		}
		if !opened {
			// a ticket can close, reopen and close again, so each change gets its own uri
			n.AtUri = uri + "#" + c.Rev
			participants, err := deldb.TicketParticipants(x, uri)
			if err != nil {
				return err
			}
			recipients = append(recipients, participants...)
		}
		if err := i.fanOut(x, n, recipients, mentions); err != nil {
			return err
		}
		if event := stateEvent(kind); event != "" {
			// the ticket's author is the subject, as the appview's events had it
			i.analytics.Capture(event, editorOr(next.Author, actor), uri+"#"+c.Rev, posthog.NewProperties().
				Set("repo_did", e.Did).
				Set("uri", uri).
				Set("actor", actor))
		}
		return nil
	}
}

// the appview's names, so existing dashboards keep counting
func stateEvent(kind models.NotificationType) string {
	switch kind {
	case models.NotificationTypeIssueClosed:
		return "issue_closed"
	case models.NotificationTypeIssueReopen:
		return "issue_reopen"
	case models.NotificationTypePullClosed:
		return "pull_closed"
	case models.NotificationTypePullReopen:
		return "pull_reopen"
	case models.NotificationTypePullMerged:
		return "pull_merged"
	}
	return ""
}

func stateNotification(kind models.EntityKind, from, to string) models.NotificationType {
	if from == to {
		return ""
	}
	pull := kind == models.EntityKindPull
	pick := func(issueType, pullType models.NotificationType) models.NotificationType {
		if pull {
			return pullType
		}
		return issueType
	}
	switch to {
	case ticketStateClosed:
		return pick(models.NotificationTypeIssueClosed, models.NotificationTypePullClosed)
	case ticketStateCompleted:
		// completed on a pull means it merged
		return pick(models.NotificationTypeIssueClosed, models.NotificationTypePullMerged)
	case ticketStateOpen:
		return pick(models.NotificationTypeIssueReopen, models.NotificationTypePullReopen)
	}
	return ""
}

func (i *Ingester) planComment(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
	c := e.Commit
	if c.Operation == jetstream.CommitOperationDelete {
		return func(x deldb.Execer) error {
			if err := deldb.DeleteComment(x, uri); err != nil {
				return err
			}
			return deldb.DeleteNotificationsFrom(x, uri)
		}
	}
	// an edit can't change the author or reply target, so only creates count
	if c.Operation != jetstream.CommitOperationCreate || len(c.Record) == 0 {
		return nil
	}
	var rec commentRecord
	if err := json.Unmarshal(c.Record, &rec); err != nil {
		i.logger.Warn("decoding comment record", "err", err, "uri", uri)
		return nil
	}
	if rec.Subject == nil {
		return nil
	}
	subject, err := syntax.ParseATURI(rec.Subject.Uri)
	if err != nil || subject.Authority().String() != e.Did {
		return nil
	}
	ticketUri := subject.String()
	switch subject.Collection().String() {
	case ticketNSID:
	case patchNSID:
		// a backfill sends a repo's comments before its patches, so an unknown patch
		// stays the key and TicketParticipants finds it once the patch arrives
		if t := deldb.GetPatchTicket(i.db, ticketUri); t != "" {
			ticketUri = t
		}
	default:
		return nil
	}

	actor := editorOr(rec.Editor, e.Did)
	comment := deldb.Comment{Uri: uri, TicketUri: ticketUri, Author: actor}
	if rec.ReplyTo != nil {
		comment.ReplyTo = rec.ReplyTo.Uri
	}
	store := func(x deldb.Execer) error { return deldb.PutComment(x, comment) }
	if !notify {
		return store
	}
	// no ticket yet, or a deleted one, so there's nothing to name in a notice
	ent := deldb.GetEntity(i.db, ticketUri)
	if ent.Kind == "" {
		return store
	}
	learned := i.fillOpener(ctx, ticketUri, &ent)
	owner := i.repoOwner(ctx, e.Did)
	mentions := i.mentionDids(ctx, actor, bodyText(rec.Body))

	return func(x deldb.Execer) error {
		if err := store(x); err != nil {
			return err
		}
		if learned {
			if err := deldb.PutEntity(x, ticketUri, ent); err != nil {
				return err
			}
		}

		var participants []string
		var err error
		switch {
		case ent.Kind == models.EntityKindPull:
			participants, err = deldb.TicketParticipants(x, ticketUri)
		case comment.ReplyTo != "":
			participants, err = deldb.ThreadParticipants(x, comment.ReplyTo)
		default:
			participants = []string{ent.Author}
		}
		if err != nil {
			return err
		}

		n := models.Notification{
			AtUri:       uri,
			Type:        models.NotificationTypeIssueCommented,
			ActorDid:    actor,
			RepoDid:     e.Did,
			EntityAt:    ticketUri,
			EntityKind:  ent.Kind,
			EntityTitle: ent.Title,
			Created:     time.UnixMicro(e.TimeUS).UTC(),
		}
		if ent.Kind == models.EntityKindPull {
			n.Type = models.NotificationTypePullCommented
		}
		return i.fanOut(x, n, append([]string{owner}, participants...), mentions)
	}
}

func (i *Ingester) planPatch(e *jetstream.Event, uri string) writer {
	if e.Commit.Operation == jetstream.CommitOperationDelete {
		return func(x deldb.Execer) error { return deldb.DeletePatch(x, uri) }
	}
	var rec patchRecord
	if err := json.Unmarshal(e.Commit.Record, &rec); err != nil || rec.Ticket == "" {
		i.logger.Warn("decoding patch record", "err", err, "uri", uri)
		return nil
	}
	ticketUri := fmt.Sprintf("at://%s/%s/%s", e.Did, ticketNSID, rec.Ticket)
	return func(x deldb.Execer) error { return deldb.PutPatch(x, uri, ticketUri) }
}

func (i *Ingester) planCollaborator(e *jetstream.Event) writer {
	did, err := syntax.ParseDID(e.Commit.RKey)
	if err != nil {
		return nil
	}
	if e.Commit.Operation == jetstream.CommitOperationDelete {
		return func(x deldb.Execer) error { return deldb.DeleteCollaborator(x, e.Did, did.String()) }
	}
	return func(x deldb.Execer) error { return deldb.PutCollaborator(x, e.Did, did.String()) }
}

func (i *Ingester) planSubscription(e *jetstream.Event, uri string) writer {
	c := e.Commit
	if c.Operation == jetstream.CommitOperationDelete {
		return func(x deldb.Execer) error { return deldb.DeleteSubscription(x, e.Did, c.RKey) }
	}
	var rec subscriptionRecord
	if err := json.Unmarshal(c.Record, &rec); err != nil || rec.subject() == "" {
		i.logger.Warn("decoding subscription record", "err", err, "uri", uri)
		return nil
	}
	return func(x deldb.Execer) error {
		return deldb.PutSubscription(x, e.Did, c.RKey, rec.subject(), rec.Ignore)
	}
}

// undoing a star or a follow takes its notice back, the same as erasing a comment
func takeBack(uri string) writer {
	return func(x deldb.Execer) error { return deldb.DeleteNotificationsFrom(x, uri) }
}

func (i *Ingester) planStar(ctx context.Context, e *jetstream.Event, uri string, notify bool) writer {
	if e.Commit.Operation == jetstream.CommitOperationDelete {
		return takeBack(uri)
	}
	if e.Commit.Operation != jetstream.CommitOperationCreate || !notify {
		return nil
	}
	var rec starRecord
	if err := json.Unmarshal(e.Commit.Record, &rec); err != nil {
		i.logger.Warn("decoding star record", "err", err, "uri", uri)
		return nil
	}
	if rec.Subject == nil || rec.Subject.Type != starRepoType || rec.Subject.Identity == "" {
		return nil
	}
	repoDid := rec.Subject.Identity
	ownerDid := i.repoOwner(ctx, repoDid)
	if ownerDid == "" {
		i.logger.Warn("star: could not resolve repo owner, skipping notification", "repoDid", repoDid)
		return nil
	}
	return func(x deldb.Execer) error {
		return i.deliver(x, ownerDid, models.Notification{
			AtUri:    uri,
			Type:     models.NotificationTypeRepoStarred,
			ActorDid: e.Did,
			RepoDid:  repoDid,
			Created:  time.UnixMicro(e.TimeUS).UTC(),
		})
	}
}

func (i *Ingester) planFollow(e *jetstream.Event, uri string, notify bool) writer {
	if e.Commit.Operation == jetstream.CommitOperationDelete {
		return takeBack(uri)
	}
	if e.Commit.Operation != jetstream.CommitOperationCreate || !notify {
		return nil
	}
	subject, err := syntax.ParseDID(e.Commit.RKey)
	if err != nil {
		return nil
	}
	return func(x deldb.Execer) error {
		return i.deliver(x, subject.String(), models.Notification{
			AtUri:    uri,
			Type:     models.NotificationTypeFollowed,
			ActorDid: e.Did,
			Created:  time.UnixMicro(e.TimeUS).UTC(),
		})
	}
}

func (i *Ingester) fanOut(x deldb.Execer, n models.Notification, implicit, mentions []string) error {
	// over the cap is spam, so drop them all instead of trimming to the cap
	if len(mentions) > maxMentions {
		mentions = nil
	}

	recipients := sets.New[string]()
	for _, did := range implicit {
		if did != "" {
			recipients.Insert(did)
		}
	}
	subscribers, ignorers, err := deldb.ListSubscribers(x, n.EntityAt, n.RepoDid)
	if err != nil {
		return fmt.Errorf("listing subscribers: %w", err)
	}
	for _, did := range subscribers {
		recipients.Insert(did)
	}
	for _, did := range mentions {
		recipients.Remove(did)
	}
	// a mute beats everything, the owner too
	for _, did := range ignorers {
		recipients.Remove(did)
	}

	for did := range recipients.All() {
		if err := i.deliver(x, did, n); err != nil {
			return err
		}
	}

	// mentions ignore mutes, so a muted thread still pings you by name
	mention := n
	mention.Type = models.NotificationTypeUserMentioned
	for _, did := range mentions {
		if err := i.deliver(x, did, mention); err != nil {
			return err
		}
	}
	return nil
}

func (i *Ingester) repoOwner(ctx context.Context, repoDid string) string {
	if owner := deldb.GetRepoOwner(i.db, repoDid); owner != "" {
		return owner
	}
	if i.bobbin == nil {
		return ""
	}

	owner, name, err := i.bobbin.RepoOwner(ctx, repoDid)
	if err != nil {
		i.logger.Warn("resolving repo owner", "err", err, "repoDid", repoDid)
		return ""
	}
	if owner == "" {
		return ""
	}

	// a nameless record must not blank out a name we already cached
	if name == "" {
		name = deldb.GetRepoName(i.db, repoDid)
	}
	if err := deldb.PutRepoName(i.db, repoDid, owner, name); err != nil {
		i.logger.Warn("caching repo owner", "err", err, "repoDid", repoDid)
	}
	return owner
}

// a backfilled ticket only names its last editor, so bobbin says who opened it. true when it found out
func (i *Ingester) fillOpener(ctx context.Context, ticketUri string, ent *deldb.Entity) bool {
	if ent.Author != "" || i.bobbin == nil {
		return false
	}
	author, err := i.bobbin.TicketAuthor(ctx, ticketUri)
	if err != nil {
		i.logger.Warn("looking up who opened a ticket", "err", err, "uri", ticketUri)
		return false
	}
	ent.Author = author
	return author != ""
}

func (i *Ingester) deliver(x deldb.Execer, recipientDid string, n models.Notification) error {
	if recipientDid == "" || recipientDid == n.ActorDid {
		return nil
	}

	prefs, err := deldb.GetNotificationPreference(x, recipientDid)
	if err != nil {
		return fmt.Errorf("loading prefs for %s: %w", recipientDid, err)
	}
	if !prefs.ShouldNotify(n.Type) {
		return nil
	}

	n.RecipientDid = recipientDid
	return deldb.CreateNotification(x, &n)
}