package spindle

import (
	"context"
	"errors"
	"slices"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/tapc"
)

const (
	defaultRepoJobWorkers = 256
	// unserved repos can only have a refresh pending, which keeps the cap low
	recordMarksPerRepo         = 1000
	recordMarksPerUnservedRepo = 20
)

type jobState uint8

const (
	jobQueued jobState = iota + 1
	jobRunning
	jobAgain
)

type queuedJob struct {
	did  syntax.DID
	knot string
}

// one job per repo, a kick during a run folds into one more run
// knots share the pool, and a reserve stays back so new knots can start
type repoJobs struct {
	mu    sync.Mutex
	state map[syntax.DID]jobState
	queue []queuedJob
	// two channels so one kind of worker can't eat the other's wakeup
	wake         chan struct{}
	wakePriority chan struct{}

	// zero until configure runs, and zero means no limits
	pool     int
	knotOf   func(syntax.DID) string
	priority map[string]bool

	queued  map[string]int
	running map[string]int
	// the slot goes back under the knot it was taken from, not whatever the repo says now
	runningKnot map[syntax.DID]runningJob
}

type runningJob struct {
	knot   string
	shared bool
}

// priority is keyed by knot, the same string knotOf returns
func (j *repoJobs) configure(pool int, knotOf func(syntax.DID) string, priority map[string]bool) {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.pool, j.knotOf, j.priority = pool, knotOf, priority
}

func (j *repoJobs) wakeLocked() chan struct{} {
	if j.wake == nil {
		j.wake = make(chan struct{}, 1)
		j.wakePriority = make(chan struct{}, 1)
		j.state = map[syntax.DID]jobState{}
		j.queued = map[string]int{}
		j.running = map[string]int{}
		j.runningKnot = map[syntax.DID]runningJob{}
	}
	return j.wake
}

func nudge(wake chan struct{}) {
	select {
	case wake <- struct{}{}:
	default:
	}
}

func (j *repoJobs) nudgeAllLocked() {
	nudge(j.wake)
	nudge(j.wakePriority)
}

func (j *repoJobs) enqueueLocked(did syntax.DID, knot string) {
	j.state[did] = jobQueued
	j.queue = append(j.queue, queuedJob{did, knot})
	j.queued[knot]++
}

func (j *repoJobs) kick(did syntax.DID) {
	j.mu.Lock()
	j.wakeLocked()
	switch j.state[did] {
	case jobRunning:
		j.state[did] = jobAgain
		fallthrough
	case jobQueued, jobAgain:
		j.mu.Unlock()
		return
	}
	knotOf := j.knotOf
	j.mu.Unlock()

	// knotOf hits the db, call it outside the lock
	knot := ""
	if knotOf != nil {
		knot = knotOf(did)
	}
	j.mu.Lock()
	defer j.mu.Unlock()
	switch j.state[did] {
	case 0:
		j.enqueueLocked(did, knot)
		j.nudgeAllLocked()
	case jobRunning:
		j.state[did] = jobAgain
	}
}

func (j *repoJobs) limitLocked(knot string) int {
	limit := j.pool
	if others := len(j.queued) - 1; others > 0 {
		limit = max(2, j.pool/(others+1))
	}
	if knot == "" {
		limit = max(2, limit/2)
	}
	return min(limit, max(1, j.pool-max(2, j.pool/8)))
}

func (j *repoJobs) pickLocked(dedicated bool) int {
	for i, q := range j.queue {
		switch {
		case dedicated:
			if j.priority[q.knot] {
				return i
			}
		case j.pool == 0 || j.running[q.knot] < j.limitLocked(q.knot):
			return i
		}
	}
	return -1
}

func (j *repoJobs) next(ctx context.Context) (syntax.DID, bool) {
	return j.take(ctx, false)
}

func (j *repoJobs) nextPriority(ctx context.Context) (syntax.DID, bool) {
	return j.take(ctx, true)
}

func (j *repoJobs) take(ctx context.Context, dedicated bool) (syntax.DID, bool) {
	for {
		j.mu.Lock()
		j.wakeLocked()
		wake := j.wake
		if dedicated {
			wake = j.wakePriority
		}
		if i := j.pickLocked(dedicated); i >= 0 {
			q := j.queue[i]
			j.queue = slices.Delete(j.queue, i, i+1)
			if j.queued[q.knot]--; j.queued[q.knot] <= 0 {
				delete(j.queued, q.knot)
			}
			j.state[q.did] = jobRunning
			if !dedicated {
				j.running[q.knot]++
			}
			j.runningKnot[q.did] = runningJob{q.knot, !dedicated}
			if len(j.queue) > 0 {
				j.nudgeAllLocked()
			}
			j.mu.Unlock()
			return q.did, true
		}
		j.mu.Unlock()
		select {
		case <-wake:
		case <-ctx.Done():
			return "", false
		}
	}
}

func (j *repoJobs) finish(did syntax.DID) {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.wakeLocked()
	ran := j.runningKnot[did]
	delete(j.runningKnot, did)
	if ran.shared {
		if j.running[ran.knot]--; j.running[ran.knot] <= 0 {
			delete(j.running, ran.knot)
		}
	}
	defer j.nudgeAllLocked()
	if j.state[did] != jobAgain {
		delete(j.state, did)
		return
	}
	j.enqueueLocked(did, ran.knot)
}

func (j *repoJobs) work(ctx context.Context, run func(context.Context, syntax.DID)) {
	j.workWith(ctx, run, j.next)
}

func (j *repoJobs) workPriority(ctx context.Context, run func(context.Context, syntax.DID)) {
	j.workWith(ctx, run, j.nextPriority)
}

func (j *repoJobs) workWith(ctx context.Context, run func(context.Context, syntax.DID), next func(context.Context) (syntax.DID, bool)) {
	for {
		did, ok := next(ctx)
		if !ok {
			return
		}
		// a panic can't kill the worker, and the marks are still there for the next run
		_ = recovered("repo job", func() error { run(ctx, did); return nil })
		j.finish(did)
	}
}

type markOnlyKey struct{}

func withMarkOnly(ctx context.Context) context.Context {
	return context.WithValue(ctx, markOnlyKey{}, true)
}

func markOnly(ctx context.Context) bool {
	return ctx.Value(markOnlyKey{}) != nil
}

// a mark is a db row saying the repo needs a refresh, the repo's job picks it up
func (s *Spindle) markRepo(repoDid syntax.DID) error {
	known, err := s.knowsRepo(repoDid)
	if err != nil {
		return err
	}
	marked, err := s.db.MarkRepoDirty(repoDid, s.retries.clock(), known, repoRetryCap)
	if err != nil {
		return err
	}
	if !marked {
		s.l.Warn("repo retry table is full, dropping the mark of an unknown repo", "repo", repoDid)
		return nil
	}
	s.retries.jobs.kick(repoDid)
	return nil
}

// same as markRepo but for one record, applied by the repo's job
func (s *Spindle) markRecord(rec *tapc.RecordEventData) error {
	repo, err := s.servedRepo(rec.Did)
	if err != nil {
		return err
	}
	perRepo := recordMarksPerRepo
	if repo == nil {
		if _, pending, err := s.db.RepoRetryRow(rec.Did); err != nil || !pending {
			return err
		}
		perRepo = recordMarksPerUnservedRepo
	}
	r := db.RecordRetry{RepoDid: rec.Did, Collection: rec.Collection.String(), Rkey: rec.Rkey.String()}
	if err := s.db.MarkRecordDirty(r, s.retries.clock(), perRepo); err != nil {
		return err
	}
	s.retries.jobs.kick(rec.Did)
	return nil
}

func (s *Spindle) runDueWipe(ctx context.Context, did syntax.DID) {
	w, err := s.db.GetWipe(did)
	if err != nil || w.Due.After(s.retries.clock()) {
		return
	}
	if err := s.runWipe(ctx, did); err != nil && ctx.Err() == nil {
		s.l.Warn("finishing a wipe failed, it will be retried", "repo", did, "err", err)
	}
}

func (s *Spindle) kickDueRepos() {
	now := s.retries.clock()
	for _, due := range []struct {
		what string
		read func() ([]syntax.DID, error)
	}{
		{"wipes", func() ([]syntax.DID, error) { return s.db.DueWipes(now) }},
		{"repo retries", func() ([]syntax.DID, error) { return s.db.DueRepoRetries(now) }},
		{"record marks", func() ([]syntax.DID, error) { return s.db.DueRecordMarkRepos(now) }},
		{"pull events", func() ([]syntax.DID, error) { return s.db.DuePullEventRepos(now) }},
	} {
		dids, err := due.read()
		if err != nil {
			s.l.Error("reading what is due", "what", due.what, "err", err)
			continue
		}
		for _, did := range dids {
			s.retries.jobs.kick(did)
		}
	}
}

// a slow knot would pin a worker forever, and the pool is shared
var repoJobBudget = 5 * time.Minute

// past this many marks the job requeues itself, which stops one repo from hogging a worker
const recordMarksPerJob = 200

var errRepoJobBudget = errors.New("the repo job ran out of time")

func (s *Spindle) runRepoJob(ctx context.Context, did syntax.DID) {
	// pull events run in this job too, and the one-job-per-repo rule keeps them in order
	defer s.runDuePullEvents(ctx, did)
	if wiping, err := s.db.WipeExists(did); err != nil {
		s.l.Error("reading a wipe", "repo", did, "err", err)
		return
	} else if wiping {
		s.runDueWipe(ctx, did)
		return
	}
	jobCtx, cancel := context.WithTimeoutCause(ctx, repoJobBudget, errRepoJobBudget)
	defer cancel()
	s.refreshAndRunMarks(jobCtx, did)
	// count a timeout as a failure, or the repo comes back every tick and eats a worker
	if ctx.Err() == nil && errors.Is(context.Cause(jobCtx), errRepoJobBudget) {
		s.l.Warn("a repo job ran out of time, backing the repo off", "repo", did)
		s.failRepoRefresh(did)
	}
}

func (s *Spindle) jobKnot(did syntax.DID) string {
	repo, err := s.servedRepo(did)
	if err != nil || repo == nil {
		return ""
	}
	return repo.Knot
}

func (s *Spindle) refreshAndRunMarks(ctx context.Context, did syntax.DID) {
	marked, due, err := s.db.RepoRetryDue(did, s.retries.clock())
	if err != nil {
		s.l.Error("reading a repo retry", "repo", did, "err", err)
		return
	}
	// a kick doesn't skip the backoff, a failed refresh still waits it out
	if marked && !due {
		return
	}
	if marked {
		covered := s.collaboratorMarks(did)
		switch s.refreshForJob(ctx, did) {
		case refreshFailed:
			return
		case refreshSynced:
			// a full sync already read every collaborator, their marks can go
			for _, m := range covered {
				if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil {
					s.l.Error("clearing a record mark", "repo", did, "rkey", m.Rkey, "err", err)
				}
			}
		}
	}
	s.runRecordMarks(ctx, did)
}

type refreshResult int

const (
	refreshSynced refreshResult = iota
	refreshTooLong
	refreshNoKnot
	refreshFailed
)

func (s *Spindle) collaboratorMarks(did syntax.DID) []db.RecordMark {
	marks, err := s.db.DueRecordMarksOf(did, s.retries.clock())
	if err != nil {
		s.l.Error("reading the record marks", "repo", did, "err", err)
		return nil
	}
	return slices.DeleteFunc(marks, func(m db.RecordMark) bool { return m.Collection != org_tangled.GraphCollaboratorNSID })
}

func (s *Spindle) refreshForJob(ctx context.Context, did syntax.DID) refreshResult {
	epoch, _ := s.db.RepoRetryEpoch(did)
	err := s.RefreshRepo(ctx, did)
	switch {
	case err == nil:
		return refreshSynced
	case ctx.Err() != nil:
		return refreshFailed
	case errors.Is(err, knotclient.ErrNoKnot):
		// waiting won't give it a knot, the retry gets dropped
		if err := s.retries.clear(s.db, did); err != nil {
			s.l.Error("clearing a repo retry", "repo", did, "err", err)
		}
		return refreshNoKnot
	case errors.Is(err, knotclient.ErrTooManyRecords):
		// retrying won't shorten the list, the collaborator marks work through it one at a time
		if err := s.retries.clearAt(s.db, did, epoch); err != nil {
			s.l.Error("clearing a repo retry", "repo", did, "err", err)
		}
		return refreshTooLong
	}
	s.l.Warn("refreshing a repo failed", "repo", did, "err", err)
	s.failRepoRefresh(did)
	return refreshFailed
}

func (s *Spindle) runRecordMarks(ctx context.Context, did syntax.DID) {
	for worked := 0; ctx.Err() == nil; {
		marks, err := s.db.DueRecordMarksOf(did, s.retries.clock())
		if err != nil {
			s.l.Error("reading the record marks", "repo", did, "err", err)
			return
		}
		if len(marks) == 0 {
			return
		}
		failed := false
		for _, m := range marks {
			if ctx.Err() != nil {
				return
			}
			if worked++; worked > recordMarksPerJob {
				s.retries.jobs.kick(did)
				return
			}
			if !s.runRecordMark(ctx, m) {
				failed = true
			}
		}
		// a failed mark is in backoff, looping again would just spin
		if failed {
			return
		}
	}
}

func (s *Spindle) runRecordMark(ctx context.Context, m db.RecordMark) bool {
	if m.Collection == org_tangled.GraphCollaboratorNSID {
		return s.applyCollaboratorMark(ctx, m)
	}
	rec := &tapc.RecordEventData{Did: m.RepoDid, Collection: syntax.NSID(m.Collection), Rkey: syntax.RecordKey(m.Rkey), Action: tapc.RecordUpdateAction, Live: true}
	if err := s.processEvent(ctx, tapc.Event{Type: tapc.EvtRecord, Record: rec}); err != nil {
		if ctx.Err() == nil {
			s.l.Warn("applying a pull request record failed", "repo", m.RepoDid, "collection", m.Collection, "rkey", m.Rkey, "err", err)
		}
		return false
	}
	// clear it here, handlers drop unserved repos without clearing
	if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil {
		s.l.Error("clearing a record mark", "repo", m.RepoDid, "rkey", m.Rkey, "err", err)
		return false
	}
	return true
}