package spindle

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math/rand/v2"
	"os"
	"time"

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

// a restart can finish the wipe since the row is in the db
func (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason string) error {
	rows, err := s.db.ReposByDid(repoDid)
	if err != nil {
		return fmt.Errorf("list repo rows: %w", err)
	}
	// force skips the turn check, otherwise an older refresh could undo the wipe
	err = s.beginTurn(repoDid).force(func() error { return s.claimWipe(ctx, repoDid, rows, reason) })
	if err != nil {
		return fmt.Errorf("claim %s: %w", repoDid, err)
	}
	return s.runWipe(ctx, repoDid)
}

// the caller runs cleanup after it drops the turn lock
func (s *Spindle) claimWipeIfRows(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
	}
	if err := s.claimWipe(ctx, repoDid, rows, reason); err != nil {
		return nil, fmt.Errorf("claim %s: %w", repoDid, err)
	}
	return func() error { return s.runWipe(context.WithoutCancel(ctx), repoDid) }, nil
}

func (s *Spindle) revokeRepoAccess(ctx context.Context, repoDid syntax.DID, rows []db.Repo) error {
	var errs []error
	collabs, err := s.db.ListCollaboratorsByRepoDid(repoDid)
	errs = append(errs, err)
	for _, c := range collabs {
		errs = append(errs, s.e.RemoveCollaborator(c.Subject.String(), rbac.ThisServer, repoDid.String()))
	}
	errs = append(errs, s.db.DeleteRepoCollaboratorsByRepoDid(repoDid))
	for _, r := range rows {
		errs = append(errs, s.e.RemoveRepo(r.Owner.String(), rbac.ThisServer, repoDid.String()))
	}
	if s.scheduler != nil {
		errs = append(errs, s.scheduler.RemoveRepo(ctx, repoDid.String()))
	}
	return errors.Join(errs...)
}

// the wipe row keeps a refresh from serving the repo during cleanup
func (s *Spindle) claimWipe(ctx context.Context, repoDid syntax.DID, rows []db.Repo, reason string) error {
	if err := s.revokeRepoAccess(ctx, repoDid, rows); err != nil {
		return err
	}
	if err := s.db.ClaimWipe(repoDid, reason, wipeSteps, s.retries.clock()); err != nil {
		return err
	}
	for _, r := range rows {
		if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil {
			s.l.Warn("releasing owner interest failed", "owner", r.Owner, "err", err)
		}
	}
	return nil
}

// order matters, a failed step stays in the row for a retry
var wipeSteps = []string{"workflows", "leases", "artifacts", "jobs", "quota", "secrets", "webhooks", "pulls", "pipelines", "clone"}

// each step has to be safe to repeat since a retry can rerun a half finished one
func (s *Spindle) runWipe(ctx context.Context, repoDid syntax.DID) error {
	w, err := s.db.GetWipe(repoDid)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("reading the wipe of %s: %w", repoDid, err)
	}
	l := s.l.With("repo", repoDid, "reason", w.Reason)
	l.Warn("wiping repo")

	var errs []error
	for _, step := range w.Steps {
		if err := s.wipeStep(ctx, repoDid, w.Reason, step); err != nil {
			l.Error("wipe step failed", "step", step, "err", err)
			errs = append(errs, fmt.Errorf("%s: %w", step, err))
			continue
		}
		if err := s.db.FinishWipeStep(repoDid, step); err != nil {
			errs = append(errs, fmt.Errorf("%s: %w", step, err))
		}
	}
	if len(errs) > 0 {
		attempts := w.Attempts + 1
		if err := s.db.FailWipe(repoDid, attempts, s.retries.clock().Add(wipeBackoff(attempts))); err != nil {
			errs = append(errs, err)
		}
		return fmt.Errorf("wiped %s with %d step errors: %w", repoDid, len(errs), errors.Join(errs...))
	}
	l.Info("repo wiped")
	return nil
}

// no attempt cap, backoff stops growing at repoRetryMax
func wipeBackoff(attempts int) time.Duration {
	backoff := min(repoRetryBase<<min(attempts-1, 10), repoRetryMax)
	return backoff - rand.N(backoff/4)
}

func (s *Spindle) wipeStep(ctx context.Context, repoDid syntax.DID, reason, step string) error {
	switch step {
	case "workflows":
		_, err := s.cancelRepoWorkflows(ctx, repoDid, reason)
		return err
	case "leases":
		return s.wipeLeases(ctx, repoDid)
	case "artifacts":
		refs, err := s.db.ListArtifactRefsByRepo(repoDid.String())
		if err != nil {
			return err
		}
		var errs []error
		for _, ref := range refs {
			errs = append(errs, s.stores.Delete(ctx, ref)...)
		}
		if err := errors.Join(errs...); err != nil {
			// rows stay so a retry still knows which objects to delete
			return err
		}
		return s.db.DeleteArtifactRefsByRepo(repoDid.String())
	case "jobs":
		return s.db.DeleteJobsByRepo(ctx, repoDid.String())
	case "quota":
		return s.db.DeleteQuotaStateForRepo(ctx, repoDid.String())
	case "secrets":
		return s.vault.RemoveAllSecrets(ctx, secrets.RepoIdentifier(repoDid.String()))
	case "webhooks":
		return s.db.DeleteWebhooksByRepo(repoDid)
	case "pulls":
		return s.db.DeletePullsByRepo(repoDid)
	case "pipelines":
		pipes, err := s.db.ListPipelineWorkflows(repoDid.String())
		if err != nil {
			return err
		}
		ids := make([]models.PipelineId, 0, len(pipes))
		for _, p := range pipes {
			ids = append(ids, models.PipelineId(p.PipelineID))
		}
		return s.db.DeletePipelinesByRepo(repoDid.String(), ids)
	case "clone":
		return os.RemoveAll(s.newRepoPath(repoDid))
	}
	return fmt.Errorf("unknown wipe step %q", step)
}

// lease rows go last, they name the logs to delete
func (s *Spindle) wipeLeases(ctx context.Context, repoDid syntax.DID) error {
	refs, err := s.db.ListArtifactRefsByRepo(repoDid.String())
	if err != nil {
		return err
	}
	isArtifact := make(map[string]bool, len(refs))
	for _, ref := range refs {
		isArtifact[ref] = true
	}
	leases, err := s.db.ListMillLeases()
	if err != nil {
		return err
	}
	var errs []error
	for _, lease := range leases {
		if lease.RepoDID != repoDid.String() {
			continue
		}
		if ref := "logs/" + lease.LeaseID + ".log"; !isArtifact[ref] {
			errs = append(errs, s.stores.Delete(ctx, ref)...)
		}
	}
	if err := errors.Join(errs...); err != nil {
		return err
	}
	_, err = s.db.DeleteMillLeasesByRepo(repoDid.String())
	return err
}

func (s *Spindle) WipeOwner(ctx context.Context, ownerDid syntax.DID, reason string) error {
	repos, err := s.db.AllRepos()
	if err != nil {
		return fmt.Errorf("list repos: %w", err)
	}
	var errs []error
	for _, r := range repos {
		if r.Owner != ownerDid || r.RepoDid == "" {
			continue
		}
		if err := s.WipeRepo(ctx, r.RepoDid, reason); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, reason string) ([]models.WorkflowId, error) {
	var wids []models.WorkflowId
	seen := map[models.WorkflowId]bool{}
	add := func(rkey, name string) models.WorkflowId {
		wid := models.WorkflowId{PipelineId: models.PipelineId(rkey), Name: name}
		if !seen[wid] {
			seen[wid] = true
			wids = append(wids, wid)
		}
		return wid
	}

	pipes, err := s.db.ListPipelineWorkflows(repoDid.String())
	if err != nil {
		return nil, fmt.Errorf("list pipeline workflows: %w", err)
	}
	for _, p := range pipes {
		add(string(p.PipelineID), p.Name)
	}

	leases, err := s.db.ListMillLeases()
	if err != nil {
		return nil, fmt.Errorf("list mill leases: %w", err)
	}
	leased := map[models.WorkflowId]bool{}
	for _, lease := range leases {
		if lease.RepoDID == repoDid.String() {
			leased[add(lease.PipelineID, lease.Workflow)] = true
		}
	}

	var errs []error
	for _, wid := range wids {
		st, err := s.db.GetStatus(wid)
		finished := err == nil && st.IsFinish()
		if !finished {
			if err := s.db.StatusCancelled(wid, reason, -1, s.n); err != nil {
				errs = append(errs, err)
			}
		}
		// a finished workflow keeps its lease while a debug shell holds the VM
		if !finished || leased[wid] {
			engine.CancelWorkflow(wid)
			for _, eng := range s.engs {
				if err := eng.DestroyWorkflow(ctx, wid); err != nil {
					errs = append(errs, err)
				}
			}
		}
		if err := os.Remove(models.LogFilePath(s.cfg.Server.LogDir, wid)); err != nil && !os.IsNotExist(err) {
			errs = append(errs, fmt.Errorf("remove log for %s: %w", wid, err))
		}
	}
	return wids, errors.Join(errs...)
}

func (s *Spindle) releaseOwnerInterest(ctx context.Context, ownerDid syntax.DID) error {
	if ownerDid == "" {
		return nil
	}
	n, err := s.db.CountReposByOwner(ownerDid)
	if err != nil {
		return err
	}
	if n > 0 {
		return nil
	}
	onList, err := db.AllowlistContains(s.db, ownerDid)
	if err != nil {
		return err
	}
	// admins are served without an allow-list entry
	if !onList && !s.cfg.Server.IsAdmin(ownerDid.String()) && s.backfill != nil {
		s.backfill.Untrack(ownerDid)
	}
	return nil
}

// only db work here, the cleanup runs on the repo job once kicked
func (s *Spindle) resumeWipes(ctx context.Context) {
	bans, err := s.db.BanList()
	if err != nil {
		s.l.Error("failed to list bans for reconcile", "err", err)
		return
	}
	if len(bans) == 0 {
		return
	}
	banned := make(map[syntax.DID]bool, len(bans))
	for _, b := range bans {
		banned[b.SubjectDid] = true
	}
	repos, err := s.db.AllRepos()
	if err != nil {
		s.l.Error("failed to list repos for the ban reconcile", "err", err)
		return
	}
	claimed := map[syntax.DID]bool{}
	for _, r := range repos {
		if r.RepoDid == "" || claimed[r.RepoDid] || !banned[r.RepoDid] && !banned[r.Owner] {
			continue
		}
		claimed[r.RepoDid] = true
		err := s.beginTurn(r.RepoDid).force(func() error {
			rows, err := s.db.ReposByDid(r.RepoDid)
			if err != nil {
				return err
			}
			return s.claimWipe(ctx, r.RepoDid, rows, "banned")
		})
		if err != nil {
			s.l.Error("ban reconcile failed", "repo", r.RepoDid, "err", err)
			continue
		}
		s.retries.jobs.kick(r.RepoDid)
	}
}