package console

import (
	"context"
	"errors"
	"log"
	"os"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
)

const (
	runSweepInterval = 30 * time.Second
	// runSweepHousekeepingEvery: expiry and lost-run closing run on every
	// pass, because a caller waits on them; retention and orphans have 24
	// hour thresholds, so they run on every 20th pass (about 10 minutes).
	runSweepHousekeepingEvery = 20
	// orphanCandidateWindow: with the record view armed, orphan
	// housekeeping lists runs/ only for apps with a run-start row this
	// recent, the window in which a run's objects can still be live or
	// awaiting cleanup (S3 change-index design §4.6, codex R2-1).
	orphanCandidateWindow = 48 * time.Hour
	// orphanFullSweepEvery: and once this often it lists every app with
	// records, as it always did, as the backstop.
	orphanFullSweepEvery = 24 * time.Hour
)

// runSweepTimeout bounds one pass, so a hung list or read cannot stall the
// sweep for good. A variable so a test can shorten it.
var runSweepTimeout = 5 * time.Minute

// StartRunSweep runs the run sweep every 30 seconds (run spec §5.8). The
// console has no deadline-sweep loop to join (index item 1), so this is a
// claim worker of its own, started from cmd/console like the others.
func (s *Server) StartRunSweep(ctx context.Context) {
	if s.cfg.HostClaim {
		log.Printf("console: run sweep: off on a host-claim console, where carlos run is refused")
		return
	}
	t := time.NewTicker(runSweepInterval)
	defer t.Stop()
	for pass := 0; ; pass++ {
		s.runSweepPass(ctx, time.Now(), pass%runSweepHousekeepingEvery == 0)
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// runSweep is one full pass: every phase.
func (s *Server) runSweep(ctx context.Context, now time.Time) { s.runSweepPass(ctx, now, true) }

// runSweepPass is one pass. Every transition is a CAS on the record, so a box
// or a CLI that acts first wins and the sweep changes nothing: a queued run
// a box picked in the meantime stays picked, a run a box finished stays
// finished. housekeeping adds the terminal-retention and orphan phases to the
// expiry and lost-run phases every pass runs. It is called from one goroutine.
//
// The records are read tolerantly: one account or record that will not read
// must not stop expiry and lost-run closing for the rest of the fleet, and
// each of those acts on a record it read. The orphan phase is the exception:
// a run directory no record names is deleted, so it runs only on a read that
// saw every record.
//
// With the record view armed the records are read from it (recordWalkStore);
// every write, and the orphan phase's listing of runs/, stays on the bucket.
func (s *Server) runSweepPass(ctx context.Context, now time.Time, housekeeping bool) {
	if s.cfg.Releases == nil {
		return
	}
	ctx, cancel := context.WithTimeout(ctx, runSweepTimeout)
	defer cancel()
	src, _ := s.recordWalkStore(ctx, s.cfg.Releases)
	recs, complete, err := provision.ListAllRecordsTolerant(ctx, src)
	if err != nil {
		log.Printf("console: run sweep: list records: %v", err)
		return
	}
	if !complete {
		log.Printf("console: run sweep: some records could not be read; their runs wait for a later pass, and no orphan run is deleted on this one")
	}
	named := map[string]bool{}
	apps := map[[2]string]bool{}
	for _, rec := range recs {
		if ctx.Err() != nil {
			log.Printf("console: run sweep: pass cut short: %v", ctx.Err())
			return
		}
		apps[[2]string{rec.Account, rec.App}] = true
		r := rec.ActiveRun
		if r == nil {
			continue
		}
		named[runs.Prefix(rec.Account, rec.App, r.ID)] = true
		acct, err := s.store.AccountBySqid(rec.Account)
		if err != nil {
			log.Printf("console: run sweep: account %s: %v", rec.Account, err)
			continue
		}
		if acct == nil {
			continue
		}
		c := runCall{acct: acct, app: rec.App, host: rec.Host, id: r.ID}
		switch {
		case r.State == provision.RunQueued && now.After(r.Deadline):
			s.sweepExpireQueued(ctx, c)
		case (r.State == provision.RunPicked || r.State == provision.RunCancel) && now.After(r.Deadline.Add(runs.LostAfter)):
			s.sweepLostRun(ctx, c, r, now)
		case !r.Open() && housekeeping:
			s.sweepTerminalRun(ctx, c, r, now)
		}
	}
	if !housekeeping || !complete {
		return
	}
	candidates, full := s.orphanCandidates(apps, now)
	for app := range candidates {
		if ctx.Err() != nil {
			log.Printf("console: run sweep: pass cut short: %v", ctx.Err())
			return
		}
		s.sweepOrphanRuns(ctx, app[0], app[1], named, now)
	}
	if full && s.recordView != nil {
		s.lastOrphanFull = now
	}
}

// orphanCandidates is the apps whose runs/ the orphan phase lists, out of
// apps, every app with records, and whether that is all of them. Disarmed it
// is all of them, as before the index. Armed (S3 change-index design §4.6,
// codex R2-1) it is only those with a run-start row in the last
// orphanCandidateWindow, except on the first housekeeping after start and
// then once every orphanFullSweepEvery, which take every app, as does a
// failed read of the rows. Orphans are still found by age, not by index
// movement: runs/ is outside every scope. The narrowed set is always a
// subset of apps, so it never lists an app the full sweep would not.
func (s *Server) orphanCandidates(apps map[[2]string]bool, now time.Time) (map[[2]string]bool, bool) {
	if s.recordView == nil || s.lastOrphanFull.IsZero() || now.Sub(s.lastOrphanFull) >= orphanFullSweepEvery {
		return apps, true
	}
	recent, err := s.store.AppsWithActionSince("run-start", now.Add(-orphanCandidateWindow))
	if err != nil {
		log.Printf("console: run sweep: recent runs: %v (every app's runs are swept this pass)", err)
		return apps, true
	}
	out := map[[2]string]bool{}
	for _, r := range recent {
		if app := [2]string{r.Sqid, r.App}; apps[app] {
			out[app] = true
		}
	}
	return out, false
}

// sweepExpireQueued expires a queued run past its pickup deadline. The
// deadline is the record's, stamped at admission after the upload, never
// request.json's requested_at. A box that picked it first wins the CAS, and
// then nothing is written. The result.json and the run-end row are the ones
// the expire route writes.
func (s *Server) sweepExpireQueued(ctx context.Context, c runCall) {
	ctx, cancel := runWriteCtx(ctx)
	defer cancel()
	if _, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id)); err != nil {
		if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("console: run sweep: %s %s: expire: %v", c.host, c.id, err)
		}
		return
	}
	if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil {
		// The record says expired; the next sweep finds the missing result.
		log.Printf("console: run sweep: %s %s: end after expire: %v", c.host, c.id, err)
	}
}

// sweepLostRun ends a picked or cancelling run whose box has not reported an
// hour past its deadline. The lost result is create-only, so a result the
// box wrote first stands; the run-end row then says what the result says.
func (s *Server) sweepLostRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) {
	ctx, cancel := runWriteCtx(ctx)
	defer cancel()
	rel := s.cfg.Releases
	// The ack refuses a last_seq that is not the result's, so a lost result
	// that guessed 0 would let the CLI acknowledge a run with chunks left.
	last, err := runs.LastChunkSeq(ctx, rel, c.acct.Sqid, c.app, c.id)
	if err != nil {
		log.Printf("console: run sweep: %s %s: list chunks: %v", c.host, c.id, err)
		return
	}
	lost := runs.Result{Interrupted: runs.InterruptedLost, Box: r.Box, LastSeq: last, Ended: now.UTC()}
	if err := runs.WriteResult(ctx, rel, c.acct.Sqid, c.app, c.id, lost); err != nil {
		log.Printf("console: run sweep: %s %s: lost result: %v", c.host, c.id, err)
		return
	}
	if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunFinish(c.id)); err != nil {
		if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("console: run sweep: %s %s: finish: %v", c.host, c.id, err)
		}
		return
	}
	res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id)
	if err != nil || res == nil {
		return // the run is done; the terminal phase writes the row
	}
	s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
}

// sweepTerminalRun clears a terminal run and deletes its objects once its
// result is a day old, or, with no result (a run the record has as done),
// once its deadline is. The record is cleared first and the objects deleted
// only if that landed, as the ack does: a run the record still names keeps
// its objects. A delete cut short leaves request.json, which the orphan pass
// finishes from.
func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.RunState, now time.Time) {
	ctx, cancel := runWriteCtx(ctx)
	defer cancel()
	rel := s.cfg.Releases
	res, err := runs.ReadResult(ctx, rel, c.acct.Sqid, c.app, c.id)
	if err != nil {
		return
	}
	ended := r.Deadline
	switch {
	case res != nil && !res.Ended.IsZero():
		ended = res.Ended
	case res == nil && (r.State == provision.RunCancelled || r.State == provision.RunExpired):
		// The console that ended this run died before it wrote the result
		// and the run-end row. Write them now (both create-only); the day
		// starts from here.
		if _, err := runs.ReadRequest(ctx, rel, c.acct.Sqid, c.app, c.id); err == nil {
			if err := s.endUnpicked(ctx, c, r.State); err != nil {
				log.Printf("console: run sweep: %s %s: end %s run: %v", c.host, c.id, r.State, err)
			}
			return
		}
	}
	if res != nil {
		// The run-end row for any terminal run with a result, before anything
		// is cleared: a run nobody followed or acked (a box's own finish), or
		// one whose sweep died after the done CAS, has no row yet. The marker
		// makes this a no-op when the row exists.
		s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
	}
	if now.Sub(ended) <= runs.KeepFor {
		return
	}
	if _, err := provision.SetRecordRun(ctx, rel, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id)); err != nil {
		if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("console: run sweep: %s %s: clear: %v", c.host, c.id, err)
		}
		return
	}
	if err := runs.DeleteRun(ctx, rel, c.acct.Sqid, c.app, c.id); err != nil {
		log.Printf("console: run sweep: %s %s: delete: %v", c.host, c.id, err)
	}
}

// sweepOrphanRuns deletes run directories no record names: acknowledged
// runs whose delete was cut short, runs replaced by a later admission,
// admissions that lost their CAS or whose outcome was unknown, runs of
// deleted hosts. With a request.json the directory goes once the request is
// a day old, after its run-end row (orphanRunEnd). Without one it is a stray
// (request.json is written first and deleted last but for the audited
// marker, so it is neither a create in flight nor a run anyone can read): it
// goes once its newest object is a day old, and stays when the store cannot
// say.
func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named map[string]bool, now time.Time) {
	rel := s.cfg.Releases
	dirs, err := rel.ListDirs(ctx, runs.RunsPrefix(acct, app))
	if err != nil {
		return
	}
	var account *Account
	accountRead := false
	for _, dir := range dirs {
		if named[dir] {
			continue
		}
		id := strings.TrimSuffix(strings.TrimPrefix(dir, runs.RunsPrefix(acct, app)), "/")
		if !runs.ValidID(id) {
			continue
		}
		req, err := runs.ReadRequest(ctx, rel, acct, app, id)
		switch {
		case err == nil:
			if now.Sub(req.RequestedAt) <= runs.KeepFor {
				continue
			}
			if !accountRead {
				if account, err = s.store.AccountBySqid(acct); err != nil {
					log.Printf("console: run sweep: account %s: %v", acct, err)
					return // no run of it is deleted before its row can be written
				}
				accountRead = true
			}
			if account != nil && !s.orphanRunEnd(ctx, account, app, req.Host, id, now) {
				continue
			}
		case errors.Is(err, os.ErrNotExist):
			if !s.strayIsOld(ctx, dir, now) {
				continue
			}
		default:
			continue
		}
		if err := runs.DeleteRun(ctx, rel, acct, app, id); err != nil {
			log.Printf("console: run sweep: %s/%s %s: delete orphan: %v", acct, app, id, err)
		}
	}
}

// orphanRunEnd writes an orphan run's run-end row before its objects go
// (run spec §6). A run that ended with no follower and was replaced by a
// later admission before a housekeeping pass, or whose record let it go,
// has none yet. With a result the row says what it says; without one the
// run never came back, so it gets a lost result (create-only: a result the
// box writes first stands) and the row says that. The audited marker makes
// this a no-op for a run that has its row. It reports whether the run may
// be deleted.
func (s *Server) orphanRunEnd(ctx context.Context, acct *Account, app, host, id string, now time.Time) bool {
	rel := s.cfg.Releases
	res, err := runs.ReadResult(ctx, rel, acct.Sqid, app, id)
	if err != nil {
		log.Printf("console: run sweep: %s/%s %s: read orphan result: %v", acct.Sqid, app, id, err)
		return false
	}
	if res == nil {
		last, err := runs.LastChunkSeq(ctx, rel, acct.Sqid, app, id)
		if err != nil {
			log.Printf("console: run sweep: %s/%s %s: list orphan chunks: %v", acct.Sqid, app, id, err)
			return false
		}
		lost := runs.Result{Interrupted: runs.InterruptedLost, LastSeq: last, Ended: now.UTC()}
		if err := runs.WriteResult(ctx, rel, acct.Sqid, app, id, lost); err != nil {
			log.Printf("console: run sweep: %s/%s %s: orphan lost result: %v", acct.Sqid, app, id, err)
			return false
		}
		if res, err = runs.ReadResult(ctx, rel, acct.Sqid, app, id); err != nil || res == nil {
			log.Printf("console: run sweep: %s/%s %s: read orphan result back: %v", acct.Sqid, app, id, err)
			return false
		}
	}
	s.auditRunEnd(ctx, acct, app, host, id, runEndOutcome(res), res)
	// The marker stands only once the row has had its chance; a marker the
	// bucket refused leaves the run for the next pass.
	if _, _, err := rel.Get(ctx, runs.AuditedKey(acct.Sqid, app, id)); err != nil {
		log.Printf("console: run sweep: %s/%s %s: no audit marker yet, delete left for a later pass: %v", acct.Sqid, app, id, err)
		return false
	}
	return true
}

// strayIsOld reports whether a run directory with no request.json has had
// nothing written to it for a day. A store that cannot list modification
// times (both real stores can) leaves strays alone.
func (s *Server) strayIsOld(ctx context.Context, dir string, now time.Time) bool {
	ml, ok := s.cfg.Releases.(release.ModLister)
	if !ok {
		return false
	}
	objs, err := ml.ListMod(ctx, dir)
	if err != nil || len(objs) == 0 {
		return false
	}
	newest := objs[0].Mod
	for _, o := range objs[1:] {
		if o.Mod.After(newest) {
			newest = o.Mod
		}
	}
	return now.Sub(newest) > runs.KeepFor
}