package spindle

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/hashicorp/golang-lru/v2/expirable"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/gitutil"
	"tangled.org/core/knotfeed"
	"tangled.org/core/log"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/xrpc"
	"tangled.org/core/tid"
	"tangled.org/core/workflow"
)

const (
	notifyWorkBudget   = 5 * time.Minute
	defaultNotifySlots = 64
	notifyMinShare     = 2
	notifyWaitWindow   = 2 * time.Minute

	// unknown repos cost two reads each and get a pool of their own
	notifyMaxUnknown        = 2
	notifyMaxUnknownPerKnot = 1
	unknownRepoMemory       = time.Minute
	unknownRepoMemoryMax    = 4096

	// keep pushes an hour longer because the knot's outbox gives up after 3 days
	pushMemory = 3*24*time.Hour + time.Hour

	changedFileOverheadBytes = 48
	changedFilesMaxBytes     = 524288
	changedFilesMaxCount     = 8192
)

// xrpc gets the status through HTTPStatus since it can't import spindle
type notifyStatus struct {
	status int
	err    error
}

func (e notifyStatus) Error() string   { return e.err.Error() }
func (e notifyStatus) Unwrap() error   { return e.err }
func (e notifyStatus) HTTPStatus() int { return e.status }

// every cap keys on this form because the repos table and did docs spell hosts differently
func canonicalKnotHost(host string) string {
	name, port, hasPort := strings.Cut(strings.ToLower(host), ":")
	name = strings.TrimSuffix(name, ".")
	if hasPort {
		return name + ":" + port
	}
	return name
}

type slotLimits struct {
	pool int
	// a new knot always finds a slot because one knot can't use up the reserve
	reserve   int
	dedicated int
	priority  map[string]bool
}

func slotReserve(pool int) int { return max(2, pool/8) }

// only knots with work get entries, which keeps this from growing with every knot seen
type notifySlots struct {
	mu        sync.Mutex
	held      map[string]int
	total     int
	prioHeld  map[string]int
	prioTotal int
	waiting   map[string]time.Time
	swept     time.Time
	now       func() time.Time
}

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

// a refused knot still goes in waiting, which shrinks everyone else's share
func (n *notifySlots) acquire(host string, lim slotLimits) bool {
	host = canonicalKnotHost(host)
	n.mu.Lock()
	defer n.mu.Unlock()
	now := n.clock()
	n.forgetStale(now)

	if lim.priority[host] && n.prioTotal < lim.dedicated {
		if n.prioHeld == nil {
			n.prioHeld = map[string]int{}
		}
		n.prioHeld[host]++
		n.prioTotal++
		return true
	}

	limit := max(1, lim.pool-lim.reserve)
	if others := n.othersWaiting(host); others > 0 {
		limit = max(notifyMinShare, lim.pool/(others+1))
	}
	if n.total >= lim.pool || n.held[host] >= limit {
		if n.waiting == nil {
			n.waiting = map[string]time.Time{}
		}
		n.waiting[host] = now
		return false
	}
	if n.held == nil {
		n.held = map[string]int{}
	}
	n.held[host]++
	n.total++
	delete(n.waiting, host)
	return true
}

func (n *notifySlots) release(host string) {
	host = canonicalKnotHost(host)
	n.mu.Lock()
	defer n.mu.Unlock()
	if n.held[host] == 0 {
		n.prioTotal--
		if n.prioHeld[host]--; n.prioHeld[host] == 0 {
			delete(n.prioHeld, host)
		}
		return
	}
	n.total--
	if n.held[host]--; n.held[host] == 0 {
		delete(n.held, host)
	}
}

func (n *notifySlots) inflight() int {
	n.mu.Lock()
	defer n.mu.Unlock()
	return n.total + n.prioTotal
}

func (n *notifySlots) othersWaiting(host string) int {
	count := len(n.waiting)
	if _, ok := n.waiting[host]; ok {
		count--
	}
	return count
}

func (n *notifySlots) forgetStale(now time.Time) {
	if now.Sub(n.swept) < notifyWaitWindow {
		return
	}
	n.swept = now
	for host, at := range n.waiting {
		if now.Sub(at) >= notifyWaitWindow {
			delete(n.waiting, host)
		}
	}
}

func priorityKnotHosts(knots []string) map[string]bool {
	hosts := map[string]bool{}
	for _, knot := range knots {
		host := strings.TrimSpace(knot)
		if rest, ok := strings.CutPrefix(strings.ToLower(host), "did:web:"); ok {
			host = strings.ReplaceAll(rest, "%3a", ":")
		}
		if host != "" {
			hosts[canonicalKnotHost(host)] = true
		}
	}
	return hosts
}

type notifyAdmission struct {
	mu    sync.Mutex
	held  map[string]int
	total int

	once   sync.Once
	missed *expirable.LRU[syntax.DID, struct{}]
}

func (a *notifyAdmission) acquire(host string) bool {
	host = canonicalKnotHost(host)
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.total >= notifyMaxUnknown || a.held[host] >= notifyMaxUnknownPerKnot {
		return false
	}
	if a.held == nil {
		a.held = map[string]int{}
	}
	a.held[host]++
	a.total++
	return true
}

func (a *notifyAdmission) release(host string) {
	host = canonicalKnotHost(host)
	a.mu.Lock()
	defer a.mu.Unlock()
	a.total--
	if a.held[host]--; a.held[host] == 0 {
		delete(a.held, host)
	}
}

func (a *notifyAdmission) memory() *expirable.LRU[syntax.DID, struct{}] {
	a.once.Do(func() {
		a.missed = expirable.NewLRU[syntax.DID, struct{}](unknownRepoMemoryMax, nil, unknownRepoMemory)
	})
	return a.missed
}

func (a *notifyAdmission) missedRecently(did syntax.DID) bool {
	_, ok := a.memory().Get(did)
	return ok
}

func (a *notifyAdmission) miss(did syntax.DID) {
	a.memory().Add(did, struct{}{})
}

// an error makes the knot retry, an empty id means nothing ran
func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) {
	push := event.EventPush
	if push == nil {
		return "", errors.New("only push events come from a knot")
	}

	key := fmt.Sprintf("%s %s %s %s", repoDid, push.Ref, push.OldSha, push.NewSha)
	// the work outlives a knot hanging up, so its retry joins it instead of redoing it
	res := s.pushGroup.DoChan(key, func() (any, error) {
		s.notifyWork.Add(1)
		defer s.notifyWork.Done()
		work, cancel := context.WithTimeout(context.WithoutCancel(ctx), notifyWorkBudget)
		defer cancel()
		// shutdown has to cancel this, or git keeps writing after the db closes
		if s.rootCtx != nil {
			defer context.AfterFunc(s.rootCtx, cancel)()
		}
		return s.runPush(work, k, repoDid, push)
	})
	select {
	case r := <-res:
		pipeline, _ := r.Val.(models.PipelineId)
		return pipeline, r.Err
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

func (s *Spindle) acquireNotifySlot(host string) (release func(), ok bool) {
	if !s.notifySlots.acquire(host, s.notifyLimits()) {
		return nil, false
	}
	return func() { s.notifySlots.release(host) }, true
}

func (s *Spindle) notifySlotCount() int {
	if n := s.cfg.Server.NotifySlots; n > 0 {
		return n
	}
	return defaultNotifySlots
}

func (s *Spindle) notifyLimits() slotLimits {
	pool := s.notifySlotCount()
	dedicated := s.cfg.Server.PriorityNotifySlots
	if dedicated <= 0 {
		dedicated = pool / 4
	}
	s.notifyPriorityOnce.Do(func() { s.notifyPriorityHosts = s.priorityKnots() })
	return slotLimits{pool: pool, reserve: slotReserve(pool), dedicated: dedicated, priority: s.notifyPriorityHosts}
}

func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, push *org_tangled.EventPush) (pipeline models.PipelineId, err error) {
	defer func() {
		if err != nil && !errors.Is(err, xrpc.ErrNotifyBusy) {
			log.FromContext(ctx).Error("notify event failed", "repo", repoDid, "err", err)
		}
	}()
	defer func() { err = refusedAsStatus(err) }()

	repo, err := s.repoForPush(ctx, k, repoDid)
	if err != nil || repo == nil {
		return "", err
	}
	release, ok := s.acquireNotifySlot(k.Host)
	if !ok {
		return "", xrpc.ErrNotifyBusy
	}
	defer release()

	return s.handlePush(ctx, k, repo, push)
}

// a miss asks the knot because db only holds repos that passed ensureRepo
func (s *Spindle) repoForPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (*db.Repo, error) {
	repo, err := s.db.GetRepoByDid(repoDid)
	if err == nil {
		return repo, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf("looking up repo %s: %w", repoDid, err)
	}
	if s.notifyAdmission.missedRecently(repoDid) {
		return nil, nil
	}
	if !s.notifyAdmission.acquire(k.Host) {
		return nil, xrpc.ErrNotifyBusy
	}
	defer s.notifyAdmission.release(k.Host)

	repo, err = s.ensureRepo(ctx, k, repoDid)
	if err != nil {
		return nil, fmt.Errorf("looking up repo %s: %w", repoDid, err)
	}
	if repo == nil {
		s.notifyAdmission.miss(repoDid)
		log.FromContext(ctx).Info("ignoring event for a repo this spindle does not serve", "repo", repoDid)
	}
	return repo, nil
}

// the knot drops an event on any 4xx, keep those for refusals a retry can't fix
func refusedAsStatus(err error) error {
	if errors.Is(err, errRevRefused) || errors.Is(err, errTooManyWorkflowFiles) {
		return notifyStatus{http.StatusUnprocessableEntity, err}
	}
	return err
}

func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush) (models.PipelineId, error) {
	l := log.FromContext(ctx).With("repo", repo.RepoDid, "ref", push.Ref)

	if !isMaterializedRef(push.Ref) {
		l.Info("skipping push outside heads and tags")
		return "", nil
	}
	oldSha, oldOk := knotfeed.ParseObjectID(push.OldSha)
	newSha, newOk := knotfeed.ParseObjectID(push.NewSha)
	if !oldOk || !newOk {
		// this 422 is final since a retry sends the same shas
		return "", notifyStatus{http.StatusUnprocessableEntity, fmt.Errorf("push shas %q and %q are not object ids", push.OldSha, push.NewSha)}
	}
	if newSha.IsZero() {
		l.Info("ref deleted, nothing to trigger")
		return "", nil
	}
	// bans apply even when the repo isn't banned, because the knot names the pusher
	if ban, err := s.db.GetBan(syntax.DID(*push.Actor)); err != nil {
		return "", fmt.Errorf("checking bans: %w", err)
	} else if ban != nil {
		l.Warn("push by a banned user, running nothing", "actor", *push.Actor)
		return "", nil
	}
	// the ban check runs before the fetch, so a banned repo costs no git work
	if gate, err := s.db.Gate(&s.cfg.Server, repo.RepoDid); err != nil {
		return "", fmt.Errorf("checking admission: %w", err)
	} else if gate.Banned || gate.Paused {
		l.Info("repo is banned or its owner isn't admitted, running nothing for the push")
		return "", nil
	}

	fire := func() {
		s.wh.FirePush(context.WithoutCancel(ctx), repo, *push.Actor, push.Ref, oldSha.String(), newSha.String())
	}
	if id, done, err := s.admitPush(l, repo, push, oldSha, newSha, fire); done || err != nil {
		return id, err
	}

	repoCloneUri := s.knots.CloneURL(k, repo.RepoDid)
	repoPath := s.newRepoPath(repo.RepoDid)
	if err := s.syncRepo(ctx, repoCloneUri, repoPath, newSha.String()); err != nil {
		return "", err
	}
	l.Info("synced git repo")

	changedFiles, _ := changedFilesUnderBudget(ctx, l, repoPath, oldSha, newSha)

	triggerRepo, err := s.buildTriggerRepo(ctx, repo)
	if err != nil {
		return "", fmt.Errorf("building trigger repo: %w", err)
	}
	if s.scheduler != nil && push.Ref == "refs/heads/"+triggerRepo.DefaultBranch {
		if err := s.scheduler.RefreshRepo(ctx, *repo); err != nil {
			l.Warn("failed to refresh schedules after default-branch push", "err", err)
		}
	}

	trigger := pushTrigger(triggerRepo, push, oldSha, newSha)
	pipelineId, err := s.runPipeline(ctx, models.PipelineId(tid.TID()), repo.RepoDid, trigger, changedFiles, repoCloneUri, repoPath, newSha.String(), nil, triggerRepo, false)
	if errors.Is(err, db.ErrBannedSubject) {
		l.Warn("dropping push for banned repo")
		return "", nil
	}
	if err != nil {
		return "", err
	}
	if pipelineId == "" {
		l.Info("no workflow matched 'push' trigger, skipping the event")
		return "", nil
	}
	l.Info("pipeline triggered", "pipeline", pipelineId)
	return pipelineId, nil
}

// the lock covers only db reads and the webhook since a fetch can hang on a knot
func (s *Spindle) admitPush(l *slog.Logger, repo *db.Repo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID, fire func()) (id models.PipelineId, done bool, err error) {
	defer s.lockRepo(repo.RepoDid).Unlock()

	if served, err := s.servedRepo(repo.RepoDid); err != nil {
		return "", true, fmt.Errorf("looking up repo %s: %w", repo.RepoDid, err)
	} else if served == nil {
		l.Info("repo went away before its push ran, dropping the push")
		return "", true, nil
	}
	// webhook fires on the first record, even if the push fails later
	if first, err := s.db.RecordPush(repo.RepoDid, push.Ref, oldSha.String(), newSha.String(), time.Now()); err != nil {
		return "", true, fmt.Errorf("recording the push: %w", err)
	} else if first {
		fire()
	}
	if existing, found, err := s.db.PushPipeline(repo.RepoDid.String(), push.Ref, oldSha.String(), newSha.String()); err != nil {
		return "", true, fmt.Errorf("looking up the push: %w", err)
	} else if found {
		l.Info("push already has a pipeline, skipping the replay", "pipeline", existing)
		return existing, true, nil
	}
	// skip ci only skips the pipeline, the webhook already fired
	if hasSkipCIPushOption(push.PushOptions) {
		l.Info("push requested ci skip, skipping the event")
		return "", true, nil
	}
	return "", false, nil
}

func hasSkipCIPushOption(pushOptions []string) bool {
	for _, opt := range pushOptions {
		switch opt {
		case "skip-ci", "ci-skip":
			return true
		}
	}
	return false
}

func pushTrigger(repo *workflow.TriggerRepo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID) workflow.TriggerMetadata {
	return workflow.TriggerMetadata{
		Kind: string(workflow.TriggerKindPush),
		Push: &workflow.PushTriggerData{
			Ref:    push.Ref,
			OldSha: oldSha.String(),
			NewSha: newSha.String(),
		},
		Repo:  repo,
		Actor: push.Actor,
	}
}

func isMaterializedRef(refname string) bool {
	return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/")
}

// unknown changes run the path-gated workflows too, since missing ci is worse
func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) (files []string, known bool) {
	if oldSha.IsZero() {
		return nil, false
	}
	paths, err := changedFiles(ctx, repoPath, oldSha, newSha)
	if err != nil {
		l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err)
		return nil, false
	}
	admitted, ok := admitChangedFiles(paths)
	if !ok {
		l.Info("too many changed files to filter by path, running the workflows that have a paths filter", "oldSha", oldSha, "newSha", newSha, "files", len(paths))
		return nil, false
	}
	return admitted, true
}

// sync is shallow so oldSha may be missing, and go-git can't read sha256 repos
func changedFiles(ctx context.Context, repoPath string, oldSha, newSha knotfeed.ObjectID) ([]string, error) {
	diff := func() ([]byte, error) {
		return gitutil.Output(ctx, "-C", repoPath, "diff", "--name-only", "--no-renames", "-z", oldSha.String(), newSha.String())
	}
	out, err := diff()
	if err != nil {
		if fetchErr := gitutil.Run(ctx, "-C", repoPath, "fetch", "--depth=1", "--filter=tree:0", "origin", oldSha.String()); fetchErr != nil {
			return nil, fmt.Errorf("fetching the old commit: %w", fetchErr)
		}
		if out, err = diff(); err != nil {
			return nil, err
		}
	}
	return strings.FieldsFunc(string(out), func(r rune) bool { return r == 0 }), nil
}

func admitChangedFiles(paths []string) (admitted []string, ok bool) {
	spent := 0
	admitted = make([]string, 0, min(len(paths), changedFilesMaxCount))
	for _, path := range paths {
		spent += changedFileOverheadBytes + len(path)
		if spent > changedFilesMaxBytes || len(admitted) == changedFilesMaxCount {
			return nil, false
		}
		admitted = append(admitted, path)
	}
	return admitted, true
}