package spindle

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"slices"
	"strings"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/api/org_tangled"
	"tangled.org/core/rbac"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/knotclient"
	"tangled.org/core/tapc"
)

// knot and pds reads stay outside the lock, otherwise a hung knot blocks the repo
func (s *Spindle) ensureRepo(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (repo *db.Repo, err error) {
	turn := s.beginTurn(repoDid)
	l := s.l.With("repo", repoDid, "knot", k.Host)

	// if a later turn already wrote, we return what it left
	// only db work goes in fn, cleanup runs after the lock is gone
	settle := func(fn func() (cleanup func() error, err error)) (*db.Repo, error) {
		var cleanup func() error
		applied, err := turn.apply(func() (err error) {
			cleanup, err = fn()
			return err
		})
		if err != nil {
			return nil, err
		}
		if !applied {
			return s.servedRepo(repoDid)
		}
		if cleanup != nil {
			return nil, cleanup()
		}
		return nil, nil
	}

	var manifest org_tangled.RepoManifest
	src := k
	err = s.knots.GetRecord(ctx, k, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest)
	if errors.Is(err, knotclient.ErrNotFound) && s.cfg.Server.FosterKnot != "" {
		foster, ok := knotAt(s.cfg.Server.FosterKnot)
		if !ok {
			return nil, fmt.Errorf("foster knot %q isn't a usable url", s.cfg.Server.FosterKnot)
		}
		src = foster
		err = s.knots.GetRecord(ctx, foster, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest)
	}
	if errors.Is(err, knotclient.ErrNotFound) {
		deleted, err := s.knots.RepoDeleted(ctx, k, repoDid)
		if err != nil && !errors.Is(err, knotclient.ErrNotFound) {
			return nil, fmt.Errorf("reading status of %s: %w", repoDid, err)
		}
		return settle(func() (func() error, error) { return s.dropIfStoredKnot(ctx, k, repoDid, deleted) })
	}
	if err != nil {
		return nil, fmt.Errorf("reading manifest of %s: %w", repoDid, err)
	}
	if manifest.Spindle == nil || *manifest.Spindle != s.cfg.Server.Did().String() {
		return settle(func() (func() error, error) { return s.forgetRepo(ctx, repoDid, "manifest names another spindle") })
	}
	owner, name, rkey, err := manifestIdentity(&manifest)
	if err != nil {
		l.Warn("ignoring manifest", "err", err)
		reason := err.Error()
		return settle(func() (func() error, error) { return s.forgetRepo(ctx, repoDid, reason) })
	}

	// the knot says who owns it, but only the owner's pds can confirm that
	declared, err := s.pdsDeclaredRepo(ctx, owner, rkey.String())
	switch {
	case errors.Is(err, knotclient.ErrNotFound):
		// a rename can leave no declaration for a bit, and forgetting here loses the old name
		l.Info("owner has no declaration for this repo yet, keeping it as it is", "owner", owner, "rkey", rkey)
		return nil, nil
	case err != nil:
		return nil, fmt.Errorf("reading declaration %s/%s: %w", owner, rkey, err)
	case declared != repoDid:
		l.Warn("owner's declaration points at another repo", "owner", owner, "rkey", rkey, "declared", declared)
		return settle(func() (func() error, error) {
			return s.forgetRepo(ctx, repoDid, "owner's declaration points elsewhere")
		})
	}

	if ban, err := s.db.IsBanned(repoDid, owner); err != nil {
		return nil, fmt.Errorf("checking bans: %w", err)
	} else if ban != nil {
		l.Warn("refusing banned repo", "owner", owner)
		return settle(func() (func() error, error) { return s.claimWipeIfRows(ctx, repoDid, "banned") })
	}
	enrolled, err := db.Enrolls(s.db, &s.cfg.Server, owner)
	if err != nil {
		return nil, fmt.Errorf("checking enrollment: %w", err)
	}
	if !enrolled {
		l.Warn("rejecting repo: owner isn't admitted", "owner", owner)
		return nil, nil
	}

	repo = &db.Repo{Knot: k.Host, Owner: owner, Rkey: rkey, RepoDid: repoDid, Name: name, CreatedAt: manifest.CreatedAt}
	var banned bool
	var wipe func() error
	applied, err := turn.apply(func() error {
		err := s.registerRepo(ctx, l, repo)
		if errors.Is(err, errBanned) {
			banned = true
			wipe, err = s.claimWipeIfRows(ctx, repoDid, "banned")
		}
		return err
	})
	switch {
	case errors.Is(err, errWiping):
		return nil, nil
	case err != nil:
		return nil, err
	case banned:
		if wipe != nil {
			return nil, wipe()
		}
		return nil, nil
	case !applied:
		return s.servedRepo(repoDid)
	}
	desired, err := s.fetchCollaborators(ctx, src, repoDid)
	if err != nil {
		return repo, err
	}
	_, err = turn.apply(func() error { return s.applyCollaborators(repoDid, desired) })
	return repo, err
}

func manifestIdentity(m *org_tangled.RepoManifest) (owner syntax.DID, name string, rkey syntax.RecordKey, err error) {
	if m.Declaration == nil {
		return "", "", "", errors.New("manifest has no declaration")
	}
	if owner, err = syntax.ParseDID(m.Declaration.Owner); err != nil {
		return "", "", "", fmt.Errorf("manifest owner %q is malformed: %w", m.Declaration.Owner, err)
	}
	name = m.Declaration.Name
	if rkey, err = syntax.ParseRecordKey(strings.ToLower(name)); err != nil {
		return "", "", "", fmt.Errorf("manifest name %q is malformed: %w", name, err)
	}
	return owner, name, rkey, nil
}

var (
	errBanned = errors.New("owner is banned")
	errWiping = errors.New("repo is being wiped")
)

// same repo did under a new owner or rkey is a transfer or rename, so the old rows go
func (s *Spindle) registerRepo(ctx context.Context, l *slog.Logger, repo *db.Repo) error {
	if wiping, err := s.db.WipeExists(repo.RepoDid); err != nil {
		return fmt.Errorf("checking for a wipe: %w", err)
	} else if wiping {
		return errWiping
	}
	rows, err := s.db.ReposByDid(repo.RepoDid)
	if err != nil {
		return fmt.Errorf("looking up repo rows: %w", err)
	}
	same := func(r db.Repo) bool { return r.Owner == repo.Owner && r.Rkey == repo.Rkey }
	stale := slices.DeleteFunc(slices.Clone(rows), same)
	var prior *db.Repo
	i := slices.IndexFunc(rows, same)
	known := i >= 0
	if known {
		prior = &rows[i]
	} else if len(stale) > 0 {
		prior = &stale[0]
	}

	// bans don't take the repo lock, so one can land right after this check
	if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil {
		return fmt.Errorf("checking bans: %w", err)
	} else if ban != nil {
		return errBanned
	}
	if err := s.e.AddRepo(repo.Owner.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil {
		return fmt.Errorf("add repo policy: %w", err)
	}
	if err := s.db.AddRepo(*repo); err != nil {
		return fmt.Errorf("add repo: %w", err)
	}

	// the migrations read the old rows, so the stale ones get deleted after
	if !known {
		migrateLegacyRepoSecrets(ctx, s.db, s.vault, l, repo.Owner, repo.Name, repo.Rkey, repo.RepoDid)
		migrateLegacyRepoCasbin(ctx, s.db, s.e, l, repo.Owner, repo.Name, repo.Rkey, repo.RepoDid)
	}
	for _, r := range stale {
		if err := s.db.DeleteRepoByOwnerRkey(r.Owner, r.Rkey); err != nil {
			l.Warn("dropping stale repo row failed", "owner", r.Owner, "rkey", r.Rkey, "err", err)
		}
		if r.Owner != repo.Owner {
			if err := s.e.RemoveRepo(r.Owner.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil {
				l.Warn("dropping stale owner policy failed", "owner", r.Owner, "err", err)
			}
			if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil {
				l.Warn("releasing stale owner interest failed", "owner", r.Owner, "err", err)
			}
		}
	}

	if s.jc != nil {
		s.jc.AddDid(repo.Owner.String())
	}
	if prior != nil && prior.Name != "" && prior.Name != repo.Name {
		l.Info("repo renamed, firing webhook", "old", prior.Name, "new", repo.Name)
		s.wh.FireRename(ctx, repo.Owner, repo, prior.Name, repo.Name)
	}
	if !known && s.scheduler != nil {
		// cloning for schedules is slow, so don't block the caller or the repo lock
		go func() {
			if err := s.scheduler.RefreshRepo(context.WithoutCancel(ctx), *repo); err != nil {
				l.Warn("failed to refresh repository schedules", "err", err)
			}
		}()
	}
	return nil
}

func (s *Spindle) fetchCollaborators(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (map[syntax.DID]struct{}, error) {
	l := s.l.With("repo", repoDid)
	rkeys, err := s.knots.ListRecordKeys(ctx, k, repoDid, org_tangled.GraphCollaboratorNSID)
	if err != nil {
		return nil, fmt.Errorf("listing collaborators of %s: %w", repoDid, err)
	}
	desired := make(map[syntax.DID]struct{}, len(rkeys))
	for _, rkey := range rkeys {
		subject, err := syntax.ParseDID(rkey)
		if err != nil {
			l.Info("skipping collaborator with a malformed did", "rkey", rkey)
			continue
		}
		desired[subject] = struct{}{}
	}
	return desired, nil
}

func (s *Spindle) applyCollaborators(repoDid syntax.DID, desired map[syntax.DID]struct{}) error {
	l := s.l.With("repo", repoDid)
	tracked, err := s.db.ListCollaboratorsByRepoDid(repoDid)
	if err != nil {
		return err
	}

	var errs []error
	have := make(map[syntax.DID]struct{}, len(tracked))
	for _, c := range tracked {
		have[c.Subject] = struct{}{}
		if _, ok := desired[c.Subject]; ok {
			continue
		}
		if err := s.untrackCollaborator(repoDid, c.Subject); err != nil {
			errs = append(errs, fmt.Errorf("remove collaborator %s: %w", c.Subject, err))
			continue
		}
		l.Info("removed collaborator", "subject", c.Subject)
	}
	for subject := range desired {
		if _, ok := have[subject]; ok {
			continue
		}
		if err := s.trackCollaborator(repoDid, subject); err != nil {
			errs = append(errs, fmt.Errorf("add collaborator %s: %w", subject, err))
			continue
		}
		l.Info("added collaborator", "subject", subject)
	}
	return errors.Join(errs...)
}

// access goes first, the clone is removed after the lock is released
func (s *Spindle) forgetRepo(ctx context.Context, repoDid syntax.DID, reason string) (cleanup func() error, err error) {
	rows, err := s.db.ReposByDid(repoDid)
	if err != nil {
		return nil, fmt.Errorf("looking up repo rows: %w", err)
	}
	if len(rows) == 0 {
		return nil, nil
	}
	l := s.l.With("repo", repoDid, "reason", reason)
	l.Warn("no longer serving repo, keeping its secrets and history")

	if err := s.revokeRepoAccess(ctx, repoDid, rows); err != nil {
		return nil, fmt.Errorf("forgetting %s: %w", repoDid, err)
	}
	if err := s.db.DeleteReposByDid(repoDid); err != nil {
		return nil, err
	}
	return func() error {
		for _, r := range rows {
			if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil {
				l.Warn("releasing owner interest failed", "owner", r.Owner, "err", err)
			}
		}
		return os.RemoveAll(s.newRepoPath(repoDid))
	}, nil
}

// a knot without the repo yet also says not found, so only the stored knot counts.
// secrets only go once that knot says the repo was deleted
func (s *Spindle) dropIfStoredKnot(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, deleted bool) (cleanup func() error, err error) {
	rows, err := s.db.ReposByDid(repoDid)
	if err != nil {
		return nil, fmt.Errorf("looking up repo rows: %w", err)
	}
	if !slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Knot == k.Host }) {
		if len(rows) > 0 {
			s.l.Warn("not wiping, the knot that lost the manifest isn't the stored one", "repo", repoDid, "knot", k.Host, "stored", rows[0].Knot)
		}
		return nil, nil
	}
	if !deleted {
		return s.forgetRepo(ctx, repoDid, "manifest missing on the knot")
	}
	return s.claimWipeIfRows(ctx, repoDid, "repo deleted on the knot")
}

// a missing declaration comes back as knotclient.ErrNotFound
func (s *Spindle) pdsDeclaredRepo(ctx context.Context, owner syntax.DID, rkey string) (syntax.DID, error) {
	pds, err := s.pdsOf(ctx, owner)
	if err != nil {
		return "", err
	}
	declared, err := s.declaredRepoAt(ctx, pds, owner, rkey)
	if errors.Is(err, knotclient.ErrNotFound) && s.cfg.Server.LiarPds != "" {
		liar, ok := knotAt(s.cfg.Server.LiarPds)
		if !ok {
			return "", fmt.Errorf("liar pds %q isn't a usable url", s.cfg.Server.LiarPds)
		}
		return s.declaredRepoAt(ctx, liar, owner, rkey)
	}
	return declared, err
}

func (s *Spindle) declaredRepoAt(ctx context.Context, pds knotclient.Knot, owner syntax.DID, rkey string) (syntax.DID, error) {
	var declaration org_tangled.RepoDeclaration
	if err := s.knots.GetRecord(ctx, pds, owner, org_tangled.RepoDeclarationNSID, rkey, &declaration); err != nil {
		return "", err
	}
	return syntax.ParseDID(declaration.Repo)
}

// always reads from the knot, for xrpc calls the firehose hasn't caught up to
func (s *Spindle) RefreshRepo(ctx context.Context, repoDid syntax.DID) error {
	// a mark landing mid-refresh bumps the epoch, and clearAt only clears when it hasn't moved
	epoch, err := s.db.RepoRetryEpoch(repoDid)
	if err != nil {
		s.l.Error("reading a repo retry", "repo", repoDid, "err", err)
		epoch = -1
	}
	k, err := s.knots.KnotOf(ctx, repoDid)
	if err != nil {
		return fmt.Errorf("finding the knot of %s: %w", repoDid, err)
	}
	if _, err := s.ensureRepo(ctx, k, repoDid); err != nil {
		return err
	}
	if err := s.retries.clearAt(s.db, repoDid, epoch); err != nil {
		s.l.Error("clearing a repo retry", "repo", repoDid, "err", err)
	}
	return nil
}

// a live event won't come again, so a failed refresh has to go in the retry set
func (s *Spindle) refreshForEvent(ctx context.Context, rec *tapc.RecordEventData, repoDid syntax.DID) error {
	if markOnly(ctx) {
		return s.markRepo(repoDid)
	}
	err := s.RefreshRepo(ctx, repoDid)
	// no knot means no retry, waiting won't add one
	if err != nil && rec.Live && ctx.Err() == nil && !errors.Is(err, knotclient.ErrNoKnot) {
		s.failRepoRefresh(repoDid)
	}
	return err
}

func (s *Spindle) knowsRepo(repoDid syntax.DID) (bool, error) {
	rows, err := s.db.ReposByDid(repoDid)
	return len(rows) > 0, err
}