package spindle

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/api/tangled"
	"tangled.org/core/gitutil"
	"tangled.org/core/hostutil"
	knotfeed "tangled.org/core/knotfeed"
	kgit "tangled.org/core/knotserver/git"
	"tangled.org/core/log"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/db"
	"tangled.org/core/workflow"
)

const collaboratorInviteNSID syntax.NSID = tangled.RepoCollaboratorInviteNSID

const (
	reconcileInterval = 10 * time.Minute
	reconcileJitter   = time.Minute
	maxListPages      = 256

	feedWipeAttempts  = 3
	feedWipeRetryWait = 100 * time.Millisecond
)

const (
	changedFileOverheadBytes = 48
	changedFilesMaxBytes     = 524288
	changedFilesMaxCount     = 8192
)

var knotHTTPClient = &http.Client{Timeout: 30 * time.Second}

type refRecord struct {
	Rkey syntax.RecordKey
	Sha  knotfeed.ObjectID
}

func (s *Spindle) handleKnotFeed(ctx context.Context, knot string, msg knotfeed.Message) error {
	if msg.Type != knotfeed.TypeCommit || msg.Commit == nil {
		return nil
	}
	var errs []error
	var invites []syntax.DID
	seen := make(map[syntax.DID]struct{})
	for _, op := range msg.Commit.Records {
		repoDid := msg.Commit.Repo
		switch op.Collection {
		case knotfeed.GitRefCollection:
			if err := s.handleRefOp(ctx, knot, repoDid, op); err != nil {
				errs = append(errs, err)
			}
		case collaboratorInviteNSID:
			if _, ok := seen[repoDid]; !ok {
				seen[repoDid] = struct{}{}
				invites = append(invites, repoDid)
			}
		}
	}
	for _, repoDid := range invites {
		if err := s.reconcileCollaborators(ctx, knot, repoDid); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

func (s *Spindle) owningRepo(l *slog.Logger, knot string, repoDid syntax.DID, event string) (*db.Repo, bool, error) {
	repo, err := s.db.GetRepoByDid(repoDid)
	switch {
	case errors.Is(err, sql.ErrNoRows):
		l.Info(fmt.Sprintf("skipping %s for unknown repo", event))
		return nil, false, nil
	case err != nil:
		return nil, false, fmt.Errorf("lookup repo %s: %w", repoDid, err)
	case repo.Knot != knot:
		l.Info(fmt.Sprintf("dropping %s from non-owning knot", event), "repoKnot", repo.Knot)
		return nil, false, nil
	}
	return repo, true, nil
}

func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.DID, op knotfeed.RecordOp) error {
	l := log.FromContext(ctx).With("repo", repoDid, "rkey", op.Rkey)

	repo, proceed, err := s.owningRepo(l, knot, repoDid, "ref update")
	if !proceed || err != nil {
		return err
	}

	refname, ok := knotfeed.UnescapeRkey(op.Rkey)
	if !ok {
		l.Info("skipping ref update with undecodable rkey")
		return nil
	}
	if !isMaterializedRef(refname) {
		l.Info("skipping ref update outside heads and tags", "ref", refname)
		return nil
	}

	if op.Deleted() {
		if err := s.db.DeleteFeedRef(repoDid, op.Rkey); err != nil {
			return fmt.Errorf("forgetting ref state: %w", err)
		}
		l.Info("ref deleted, nothing to trigger", "ref", refname)
		return nil
	}

	record, err := knotfeed.DecodeRefRecord(op.Bytes)
	if err != nil {
		return fmt.Errorf("decoding ref record: %w", err)
	}

	oldSha, err := s.priorSha(repoDid, op)
	if err != nil {
		return err
	}
	// deliver push webhooks independently of CI; a skip-ci push option
	// should not suppress webhook notifications
	s.wh.FirePush(ctx, repo, string(record.Editor), refname, oldSha.String(), record.Sha.String())

	if kgit.HasSkipCIPushOption(record.PushOptions) {
		if err := s.db.PutFeedRef(repoDid, op.Rkey, record.Sha); err != nil {
			return fmt.Errorf("recording ref state: %w", err)
		}
		l.Info("push requested ci skip, skipping the event", "ref", refname)
		return nil
	}

	repoCloneUri := s.newRepoCloneUrl(knot, repoDid)
	repoPath := s.newRepoPath(repoDid)
	if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, record.Sha.String(), sparseWorkflowDir); err != nil {
		return fmt.Errorf("sync git repo: %w", err)
	}
	l.Info("synced git repo")

	changedFiles := changedFilesUnderBudget(l, repoPath, oldSha, record.Sha)

	triggerRepo, err := s.buildTriggerRepo(ctx, repo)
	if err != nil {
		return fmt.Errorf("building trigger repo: %w", err)
	}
	if s.scheduler != nil && refname == "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 := tangled.Pipeline_TriggerMetadata{
		Kind: string(workflow.TriggerKindPush),
		Push: &tangled.Pipeline_PushTriggerData{
			Ref:    refname,
			OldSha: oldSha.String(),
			NewSha: record.Sha.String(),
		},
		Repo: triggerRepo,
	}

	pipelineId, err := s.runPipeline(ctx, repoDid, trigger, changedFiles, repoCloneUri, repoPath, record.Sha.String(), nil, triggerRepo)
	if err != nil {
		return err
	}
	if err := s.db.PutFeedRef(repoDid, op.Rkey, record.Sha); err != nil {
		return fmt.Errorf("recording ref state: %w", err)
	}
	if pipelineId == "" {
		l.Info("no workflow matched 'push' trigger, skipping the event")
		return nil
	}
	l.Info("pipeline triggered", "pipeline", pipelineId)
	return nil
}

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

// priorSha is where the ref pointed before this op: reported by the feed, recorded
// from the previous op, or zero when we have never seen the ref, which is not an
// error.
func (s *Spindle) priorSha(repoDid syntax.DID, op knotfeed.RecordOp) (knotfeed.ObjectID, error) {
	if sha, ok := op.Prior.Sha(); ok {
		return sha, nil
	}
	sha, found, err := s.db.FeedRefSha(repoDid, op.Rkey)
	if err != nil {
		return knotfeed.ObjectID{}, fmt.Errorf("reading ref state: %w", err)
	}
	if !found {
		return knotfeed.ObjectID{}, nil
	}
	return sha, nil
}

func changedFilesUnderBudget(l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) []string {
	if oldSha.IsZero() {
		return nil
	}
	gr, err := kgit.Open(repoPath, newSha.String())
	if err != nil {
		l.Warn("cannot open synced repo for changed files", "err", err)
		return nil
	}
	paths, err := gr.ChangedFilesBetween(oldSha.String(), newSha.String())
	if err != nil {
		l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err)
		return nil
	}
	return admitChangedFiles(paths)
}

func admitChangedFiles(paths []string) []string {
	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
		}
		admitted = append(admitted, path)
	}
	return admitted
}

func (s *Spindle) reconcileCollaborators(ctx context.Context, knot string, repoDid syntax.DID) error {
	l := log.FromContext(ctx).With("repo", repoDid)

	_, proceed, err := s.owningRepo(l, knot, repoDid, "collaborator reconcile")
	if !proceed || err != nil {
		return err
	}

	desired, err := s.collaboratorList(ctx, knot, repoDid)
	if err != nil {
		return fmt.Errorf("listing collaborators: %w", err)
	}
	tracked, err := s.db.ListKnotCollaboratorsByRepoDid(repoDid)
	if err != nil {
		return fmt.Errorf("listing tracked collaborators: %w", err)
	}

	desiredSet := make(map[syntax.DID]struct{}, len(desired))
	for _, subject := range desired {
		desiredSet[subject] = struct{}{}
	}
	trackedSet := make(map[syntax.DID]struct{}, len(tracked))
	for _, c := range tracked {
		trackedSet[c.Subject] = struct{}{}
	}

	for subject := range desiredSet {
		if _, ok := trackedSet[subject]; ok {
			continue
		}
		if err := s.e.AddCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil {
			l.Error("failed to add collaborator policy", "subject", subject, "err", err)
			continue
		}
		if err := s.db.AddKnotCollaborator(repoDid, subject); err != nil {
			l.Error("failed to track collaborator", "subject", subject, "err", err)
			continue
		}
		l.Info("added knot-managed collaborator", "subject", subject)
	}
	for subject := range trackedSet {
		if _, ok := desiredSet[subject]; ok {
			continue
		}
		if err := s.e.RemoveCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil {
			l.Error("failed to remove collaborator policy", "subject", subject, "err", err)
			continue
		}
		if err := s.db.DeleteRepoCollaboratorBySubjectRepo(subject, repoDid); err != nil {
			l.Error("failed to delete collaborator row", "subject", subject, "err", err)
			continue
		}
		l.Info("removed knot-managed collaborator", "subject", subject)
	}
	return nil
}

func (s *Spindle) reconcileCollaboratorsLoop(ctx context.Context) {
	s.reconcileAllCollaborators(ctx)
	for {
		delay := reconcileInterval + time.Duration(rand.Int64N(int64(reconcileJitter)))
		select {
		case <-ctx.Done():
			return
		case <-time.After(delay):
		}
		s.reconcileAllCollaborators(ctx)
	}
}

func (s *Spindle) reconcileAllCollaborators(ctx context.Context) {
	repos, err := s.db.AllRepos()
	if err != nil {
		s.l.Warn("failed to load repos for collaborator reconcile", "err", err)
		return
	}
	for _, repo := range repos {
		if repo.RepoDid == "" {
			continue
		}
		if err := s.reconcileCollaborators(ctx, repo.Knot, repo.RepoDid); err != nil {
			s.l.Warn("collaborator reconcile failed", "repo", repo.RepoDid, "err", err)
		}
	}
}

func (s *Spindle) feedOutdatedReplay(ctx context.Context, knot string, feed knotfeed.Feed) knotfeed.Cursor {
	live := feed.Live(time.Now())
	repos, err := s.db.AllRepos()
	if err != nil {
		s.l.Warn("failed to load repos after outdated cursor", "knot", knot, "err", err)
		return live
	}
	reset := 0
	for _, repo := range repos {
		if repo.Knot != knot || repo.RepoDid == "" {
			continue
		}
		if err := s.wipeFeedRefs(ctx, repo.RepoDid); err != nil {
			s.l.Error("failed to reset ref state, old shas may be stale until the next push", "repo", repo.RepoDid, "err", err)
			continue
		}
		reset++
	}
	if reset > 0 {
		s.l.Warn("knot cannot replay from our cursor, ref state reset", "knot", knot, "repos", reset)
	}
	return live
}

func (s *Spindle) wipeFeedRefs(ctx context.Context, repoDid syntax.DID) error {
	var err error
	for attempt := range feedWipeAttempts {
		if attempt > 0 {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(feedWipeRetryWait << attempt):
			}
		}
		err = s.db.DeleteFeedRefsByRepoDid(repoDid)
		if err == nil {
			return nil
		}
	}
	return err
}

func (s *Spindle) collaboratorList(ctx context.Context, knot string, repoDid syntax.DID) ([]syntax.DID, error) {
	if s.listCollaborators != nil {
		return s.listCollaborators(ctx, knot, repoDid)
	}

	base, err := knotEndpoint(knot, "/xrpc/sh.tangled.repo.listCollaborators")
	if err != nil {
		return nil, err
	}
	var subjects []syntax.DID
	cursor := ""
	for page := 0; ; page++ {
		if page >= maxListPages {
			return nil, fmt.Errorf("collaborators for %s exceed %d pages", repoDid, maxListPages)
		}
		q := url.Values{}
		q.Set("subject", repoDid.String())
		q.Set("limit", "1000")
		if cursor != "" {
			q.Set("cursor", cursor)
		}
		var out struct {
			Items []struct {
				Subject string `json:"subject"`
			} `json:"items"`
			Cursor string `json:"cursor"`
		}
		if err := getJSON(ctx, base+"?"+q.Encode(), &out); err != nil {
			return nil, err
		}
		for _, item := range out.Items {
			did, err := syntax.ParseDID(item.Subject)
			if err != nil {
				return nil, fmt.Errorf("parsing collaborator subject %s: %w", item.Subject, err)
			}
			subjects = append(subjects, did)
		}
		if out.Cursor == "" {
			return subjects, nil
		}
		cursor = out.Cursor
	}
}

func (s *Spindle) refRecords(ctx context.Context, knot string, repoDid syntax.DID) ([]refRecord, error) {
	if s.listRefRecords != nil {
		return s.listRefRecords(ctx, knot, repoDid)
	}

	base, err := knotEndpoint(knot, "/xrpc/sh.tangled.git.listRefs")
	if err != nil {
		return nil, err
	}
	var refs []refRecord
	cursor := ""
	for page := 0; ; page++ {
		if page >= maxListPages {
			return nil, fmt.Errorf("refs for %s exceed %d pages", repoDid, maxListPages)
		}
		q := url.Values{}
		q.Set("repo", repoDid.String())
		q.Set("limit", "1000")
		if cursor != "" {
			q.Set("cursor", cursor)
		}
		var out struct {
			Refs []struct {
				Ref string `json:"ref"`
				Sha string `json:"sha"`
			} `json:"refs"`
			Cursor string `json:"cursor"`
		}
		if err := getJSON(ctx, base+"?"+q.Encode(), &out); err != nil {
			return nil, err
		}
		for _, ref := range out.Refs {
			rkey, named := knotfeed.EscapeRefname(ref.Ref)
			sha, isObject := knotfeed.ParseObjectID(ref.Sha)
			if !named || !isObject {
				continue
			}
			refs = append(refs, refRecord{Rkey: rkey, Sha: sha})
		}
		if out.Cursor == "" {
			return refs, nil
		}
		cursor = out.Cursor
	}
}

func knotEndpoint(knot, path string) (string, error) {
	host, noTLS, err := hostutil.ParseHostname(knot)
	if err != nil {
		return "", fmt.Errorf("parsing knot host %s: %w", knot, err)
	}
	scheme := "https"
	if noTLS {
		scheme = "http"
	}
	return fmt.Sprintf("%s://%s%s", scheme, host, path), nil
}

func getJSON(ctx context.Context, u string, out any) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
	if err != nil {
		return err
	}
	resp, err := knotHTTPClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
		return fmt.Errorf("%s %s answered %s: %s", http.MethodGet, u, resp.Status, body)
	}
	return json.NewDecoder(resp.Body).Decode(out)
}