package spindle

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/observability"
	"tangled.org/core/tapc"
)

type Ingester func(ctx context.Context, e *jetstream.Event) error

func (s *Spindle) ingest() Ingester {
	var shards *ingestShards
	var start sync.Once
	wipes := make(chan struct{}, accountJobSlots)
	return func(ctx context.Context, e *jetstream.Event) error {
		if e.Kind != jetstream.EventKindAccount && e.Kind != jetstream.EventKindCommit {
			return nil
		}
		ticket := jetstream.Hold(ctx)
		if ticket == nil {
			err := s.ingestEvent(ctx, e)
			// ingestEvent already logged it and nothing retries out here
			if err != nil && e.Kind == jetstream.EventKindCommit && ctx.Err() == nil {
				return nil
			}
			return err
		}
		// wipes do io, they skip the shards and hold the ticket until done
		if e.Kind == jetstream.EventKindAccount {
			return s.startAccountJob(ctx, e, ticket, wipes)
		}
		start.Do(func() {
			shards = newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error {
				return s.ingestEvent(withMarkOnly(ctx), e)
			})
		})
		return shards.dispatch(ctx, e, ticket)
	}
}

const accountJobSlots = 4

// wipes that run past this budget are left to the repo jobs
var accountWipeBudget = 10 * time.Minute

func (s *Spindle) startAccountJob(ctx context.Context, e *jetstream.Event, ticket ingestTicket, slots chan struct{}) error {
	select {
	case slots <- struct{}{}:
	case <-ctx.Done():
		ticket.Drop()
		return ctx.Err()
	}
	go func() {
		defer func() { <-slots }()
		s.ingestAccountJob(ctx, e, ticket)
	}()
	return nil
}

func (s *Spindle) ingestAccountJob(ctx context.Context, e *jetstream.Event, ticket ingestTicket) {
	jobCtx, cancel := context.WithTimeout(ctx, accountWipeBudget)
	defer cancel()
	err := s.ingestEvent(jobCtx, e)
	// a failed wipe already left a wipes row for the repo jobs, and the ticket finishes anyway
	if err != nil && ctx.Err() == nil {
		s.l.Error("account event wipe failed, its wipes are retried", "did", e.Did, "err", err)
	}
	finishTicket(ctx, ticket, err)
}

func (s *Spindle) ingestEvent(ctx context.Context, e *jetstream.Event) error {
	if e.Kind == jetstream.EventKindAccount {
		return s.ingestAccount(ctx, e)
	}
	if e.Kind != jetstream.EventKindCommit {
		return nil
	}

	ctx, span := observability.Tracer().Start(ctx, "jetstream.ingest")
	defer span.End()

	if span.IsRecording() {
		var attrs []attribute.KeyValue
		if e.Did != "" {
			attrs = append(attrs, attribute.String(observability.UserDIDKey, e.Did))
		}
		if e.Commit != nil {
			if e.Commit.Collection != "" {
				attrs = append(attrs, attribute.String(observability.CollectionKey, e.Commit.Collection))
			}
			if e.Commit.RKey != "" {
				attrs = append(attrs, attribute.String(observability.RKeyKey, e.Commit.RKey))
			}
		}
		span.SetAttributes(attrs...)
	}
	var err error
	if evt, ok := jetstreamToTapEvent(e); ok {
		err = s.processEvent(ctx, evt)
	}

	// a cancelled event stays unconsumed, or the cursor moves past it
	if err != nil && ctx.Err() != nil {
		span.SetStatus(codes.Error, "cancelled while processing a jetstream event")
		return err
	}
	if err != nil {
		s.l.WarnContext(ctx, "failed to process message", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err)
		s.metrics.RecordEventIngestion("jetstream", "error")
		span.SetStatus(codes.Error, "failed to process jetstream event")
		return err
	}
	s.metrics.RecordEventIngestion("jetstream", "success")
	span.SetStatus(codes.Ok, "success")
	return nil
}

// the firehose sees every repo, so each handler drops what spindle doesn't track
func (s *Spindle) processEvent(ctx context.Context, evt tapc.Event) error {
	if evt.Type != tapc.EvtRecord || evt.Record == nil {
		return nil
	}
	rec := evt.Record
	var handle func(context.Context, *tapc.RecordEventData) error
	switch rec.Collection.String() {
	case org_tangled.RepoManifestNSID:
		handle = s.onManifest
	case org_tangled.GraphCollaboratorNSID:
		handle = s.onCollaborator
	case org_tangled.RepoDeclarationNSID:
		handle = s.onDeclaration
	case org_tangled.TrackTicketNSID:
		handle = s.onTicket
	case org_tangled.ReviewPatchNSID:
		handle = s.onPatch
	default:
		return nil
	}
	source := "backfill"
	if rec.Live {
		source = "live"
	}
	if err := handle(ctx, rec); err != nil {
		s.metrics.RecordRepoRecord(source, "failed")
		return err
	}
	s.metrics.RecordRepoRecord(source, "applied")
	return nil
}

func (s *Spindle) onManifest(ctx context.Context, rec *tapc.RecordEventData) error {
	if rec.Rkey != knotclient.ManifestRkey {
		return nil
	}
	known, err := s.knowsRepo(rec.Did)
	if err != nil {
		return err
	}
	if !known {
		if rec.Action == tapc.RecordDeleteAction {
			return nil
		}
		var manifest org_tangled.RepoManifest
		if err := json.Unmarshal(rec.Record, &manifest); err != nil {
			s.l.Warn("skipping invalid manifest record", "repo", rec.Did, "err", err)
			return nil
		}
		if manifest.Spindle == nil || *manifest.Spindle != s.cfg.Server.Did().String() {
			return nil
		}
		// only enrolled owners get knot and pds reads, since anyone can point a manifest at this spindle
		if manifest.Declaration == nil {
			return nil
		}
		owner, err := syntax.ParseDID(manifest.Declaration.Owner)
		if err != nil {
			return nil
		}
		if enrolled, err := db.Enrolls(s.db, &s.cfg.Server, owner); err != nil || !enrolled {
			return err
		}
	}
	return s.refreshForEvent(ctx, rec, rec.Did)
}

func (s *Spindle) onCollaborator(ctx context.Context, rec *tapc.RecordEventData) error {
	known, err := s.knowsRepo(rec.Did)
	if err != nil || !known {
		return err
	}
	if markOnly(ctx) {
		if err := s.markRecord(rec); err != nil {
			return err
		}
	}
	return s.refreshForEvent(ctx, rec, rec.Did)
}

// delete events carry no record, so the repo has to be found by owner and rkey
func (s *Spindle) onDeclaration(ctx context.Context, rec *tapc.RecordEventData) error {
	if rec.Action == tapc.RecordDeleteAction {
		repo, err := s.db.GetRepoByOwnerRkey(rec.Did, rec.Rkey)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		// old rows can have an empty repo did, so there's nothing to refresh
		if repo.RepoDid == "" {
			return nil
		}
		return s.refreshForEvent(ctx, rec, repo.RepoDid)
	}
	var declaration org_tangled.RepoDeclaration
	if err := json.Unmarshal(rec.Record, &declaration); err != nil {
		s.l.Warn("skipping invalid declaration record", "owner", rec.Did, "err", err)
		return nil
	}
	repoDid, err := syntax.ParseDID(declaration.Repo)
	if err != nil {
		s.l.Warn("skipping declaration with a malformed repo did", "owner", rec.Did, "repo", declaration.Repo)
		return nil
	}
	rows, err := s.db.ReposByDid(repoDid)
	if err != nil {
		return err
	}
	if len(rows) > 0 {
		// anyone can write a declaration naming a served repo, and every refresh costs round trips
		if !slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Owner == rec.Did }) {
			return nil
		}
	} else if enrolled, err := db.Enrolls(s.db, &s.cfg.Server, rec.Did); err != nil || !enrolled {
		return err
	}
	return s.refreshForEvent(ctx, rec, repoDid)
}

func jetstreamToTapEvent(e *jetstream.Event) (tapc.Event, bool) {
	if e.Commit == nil {
		return tapc.Event{}, false
	}
	did, err := syntax.ParseDID(e.Did)
	if err != nil {
		return tapc.Event{}, false
	}
	var action tapc.RecordAction
	switch e.Commit.Operation {
	case jetstream.CommitOperationCreate:
		action = tapc.RecordCreateAction
	case jetstream.CommitOperationUpdate:
		action = tapc.RecordUpdateAction
	case jetstream.CommitOperationDelete:
		action = tapc.RecordDeleteAction
	default:
		return tapc.Event{}, false
	}
	return tapc.Event{
		Type: tapc.EvtRecord,
		Record: &tapc.RecordEventData{
			Did:        did,
			Rkey:       syntax.RecordKey(e.Commit.RKey),
			Collection: syntax.NSID(e.Commit.Collection),
			Rev:        e.Commit.Rev,
			Action:     action,
			Record:     e.Commit.Record,
			// jetstream is only used for live
			Live: true,
		},
	}, true
}

func (s *Spindle) ingestAccount(ctx context.Context, e *jetstream.Event) error {
	if e.Account == nil || e.Account.Active {
		return nil
	}
	status := "unknown"
	if e.Account.Status != nil {
		status = *e.Account.Status
	}
	did, err := syntax.ParseDID(e.Account.Did)
	if err != nil {
		return fmt.Errorf("parsing account did: %w", err)
	}
	s.l.Warn("account went inactive, purging its repos", "did", did, "status", status)
	return s.WipeOwner(ctx, did, "account "+status)
}