package spindle

import (
	"cmp"
	"context"
	"errors"
	"math/rand/v2"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/spindle/db"
)

const (
	repoRetryBase     = 30 * time.Second
	repoRetryMax      = time.Hour
	repoRetryAttempts = 12
	repoRetryTick     = 10 * time.Second
	repoRetryCap      = 5000
)

var errRetryTableFull = errors.New("repo retry table is full")

// the jetstream cursor moves past a failed event, so this has to live in the db
type repoRetries struct {
	mu   sync.Mutex
	now  func() time.Time
	tick time.Duration
	jobs repoJobs
}

func (r *repoRetries) clock() time.Time {
	if r.now != nil {
		return r.now()
	}
	return time.Now()
}

func (r *repoRetries) fail(d *db.DB, did syntax.DID) (bool, error) {
	return r.failWith(
		func() (int, error) {
			attempts, exists, err := d.RepoRetryRow(did)
			if err != nil || exists {
				return attempts, err
			}
			// anyone can send a fake did, and new rows only fit while there's room
			rows, err := d.CountRepoRetries()
			if err == nil && rows >= repoRetryCap {
				err = errRetryTableFull
			}
			return 0, err
		},
		func(attempts int, due time.Time) error { return d.PutRepoRetry(did, attempts, due) },
		func() error { return d.DeleteRepoRetry(did) },
	)
}

// counts one more failure in whichever store the callbacks reach, false once it's out of tries
func (r *repoRetries) failWith(prior func() (int, error), put func(attempts int, due time.Time) error, drop func() error) (bool, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	n, err := prior()
	if err != nil {
		return false, err
	}
	attempts, due, retained := r.next(n)
	if !retained {
		return false, drop()
	}
	return true, put(attempts, due)
}

func (r *repoRetries) next(prior int) (attempts int, due time.Time, retained bool) {
	attempts = prior + 1
	if attempts > repoRetryAttempts {
		return attempts, time.Time{}, false
	}
	backoff := min(repoRetryBase<<(attempts-1), repoRetryMax)
	// jitter stops everything that failed in one outage from coming due together
	backoff -= rand.N(backoff / 4)
	return attempts, r.clock().Add(backoff), true
}

func (r *repoRetries) clear(d *db.DB, did syntax.DID) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	return d.DeleteRepoRetry(did)
}

func (s *Spindle) repoJobWorkerCount() int {
	if n := s.cfg.Server.RepoJobWorkers; n > 0 {
		return n
	}
	return defaultRepoJobWorkers
}

func (s *Spindle) priorityJobWorkerCount() int {
	if len(s.priorityKnots()) == 0 {
		return 0
	}
	if n := s.cfg.Server.PriorityRepoJobWorkers; n > 0 {
		return n
	}
	return max(1, s.repoJobWorkerCount()/4)
}

// the config can list a did or a host, and both get reduced to a host here
func (s *Spindle) priorityKnots() map[string]bool {
	return priorityKnotHosts(s.cfg.Server.PriorityKnots)
}

// refresh jobs run here too, and returning waits for them
func (s *Spindle) retryFailedRepos(ctx context.Context) {
	workers := s.repoJobWorkerCount()
	s.retries.jobs.configure(workers, s.jobKnot, s.priorityKnots())
	var pool sync.WaitGroup
	for range workers {
		pool.Go(func() { s.retries.jobs.work(ctx, s.runRepoJob) })
	}
	for range s.priorityJobWorkerCount() {
		pool.Go(func() { s.retries.jobs.workPriority(ctx, s.runRepoJob) })
	}
	defer pool.Wait()

	ticker := time.NewTicker(cmp.Or(s.retries.tick, repoRetryTick))
	defer ticker.Stop()
	var lastPrune time.Time
	// after a crash there can be untried marks, kick now instead of waiting a tick
	s.kickDueRepos()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		s.kickDueRepos()
		if time.Since(lastPrune) > time.Hour {
			if _, err := s.db.PruneRepoTurns(time.Now().Add(-time.Hour)); err != nil {
				s.l.Error("pruning the repo turns", "err", err)
			}
			if _, err := s.db.PruneRepoPushes(time.Now().Add(-pushMemory)); err != nil {
				s.l.Error("pruning the recorded pushes", "err", err)
			}
			lastPrune = time.Now()
		}
	}
}

func (r *repoRetries) clearAt(d *db.DB, did syntax.DID, epoch int64) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	return d.ClearRepoRetryAt(did, epoch)
}

func (s *Spindle) failRepoRefresh(did syntax.DID) {
	retained, err := s.retries.fail(s.db, did)
	switch {
	case errors.Is(err, errRetryTableFull):
		s.l.Warn("repo retry table is full, not retrying this refresh", "repo", did)
	case err != nil:
		s.l.Error("saving a repo retry", "repo", did, "err", err)
	case !retained:
		s.l.Error("giving up on refreshing a repo", "repo", did)
	}
}