package rollout

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// scheduler is one app's goroutine (spec §6.1). It runs off the agent's
// tick, so a slow rollout never delays the tick or another app.
type scheduler struct {
	m    *Manager
	key  appKey
	mu   sync.Mutex
	q    *appQueue
	kick chan struct{}
}

func (s *scheduler) poke() {
	select {
	case s.kick <- struct{}{}:
	default:
	}
}

// newest is handed to Swap for its re-checks (spec §6.3).
func (s *scheduler) newest(host string) uint64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.q.newest(host)
}

func (s *scheduler) loop(ctx context.Context) {
	timer := time.NewTimer(time.Hour)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-s.kick:
		case <-timer.C:
		}
		if s.m.closing() {
			// The agent is stopping (Manager.Close): nothing more is admitted.
			return
		}
		view, pendingConfig := s.view()
		for host, rev := range pendingConfig {
			if serving, seen := view[host]; seen && serving == "" {
				s.m.configSlept(host, rev)
			}
		}
		now := s.m.d.Now()
		s.mu.Lock()
		ds := s.q.admit(view, now)
		next := s.q.nextDue(now)
		s.mu.Unlock()
		for _, d := range ds {
			// Counted before it starts, so Close waits for its done(). One
			// admitted as Close began is not run: nothing of it has been
			// written yet, and its slot stays taken, which keeps every later
			// admission for the host out until the process exits.
			if !s.m.enter(true) {
				continue
			}
			go func(d dispatch) {
				defer s.m.leave()
				s.execute(ctx, d)
			}(d)
		}
		wait := time.Hour
		if !next.IsZero() {
			wait = max(next.Sub(now), time.Millisecond)
		}
		timer.Reset(wait)
	}
}

// view asks the activator, with no lock held, what each queued host serves:
// a build when awake, "" when asleep, absent while waking or stopping. It
// also returns each host's pending config revision as it stood BEFORE the
// activator was asked: a host then reported asleep stopped with that file
// already delivered, so its next wake runs it (configSlept).
func (s *scheduler) view() (map[string]string, map[string]string) {
	s.mu.Lock()
	hosts := make([]string, 0, len(s.q.hosts))
	pendingConfig := map[string]string{}
	for name, h := range s.q.hosts {
		hosts = append(hosts, name)
		if h.wantConfig != "" {
			pendingConfig[name] = h.wantConfig
		}
	}
	s.mu.Unlock()
	out := make(map[string]string, len(hosts))
	for _, h := range hosts {
		if g, ok := s.m.d.Act.Current(h); ok && g.Build != "" {
			out[h] = g.Build
		} else if !s.m.d.Act.Owns(h) {
			out[h] = ""
		}
	}
	return out, pendingConfig
}

// execute runs one attempt to its outcome.
func (s *scheduler) execute(ctx context.Context, d dispatch) {
	m := s.m
	rt, err := m.d.Routes.Get(d.ID.Host)
	if err != nil || rt == nil {
		m.d.Logf("rollout: %s: route not readable for its %s attempt onto %s: %v", d.ID.Host, d.ID.Cause, d.ID.Build, err)
		s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route not readable"}, "")
		return
	}
	if !Managed(*rt) || release.RouteAccount(*rt) != s.key.acct || rt.App != s.key.app {
		// Execution admission (Astra R3 §5 item 4): the attempt was queued
		// from the pass's view of the row, and the row this execution runs
		// on may have changed since. One that has become a named-unit or
		// unit-backed route belongs to root's adoption or to systemd, and
		// one that now names another app or account is another scheduler's.
		// Either way it is not this attempt's to run. Managed also wants the
		// row ready, so one passing through draining, moving or landing is
		// not moved now either; the queue keeps it (heldOnBox) and derives
		// the attempt again after the deferral. Checked before the
		// good-build gate, the staging read and the write-ahead intent, so
		// nothing is written and nothing runs: abandoned unexecuted, and
		// done() defers the host and re-arms a retry allowance. A row that
		// stays unmanaged, or another app's, leaves this queue at the next
		// pass (forget).
		m.d.Logf("rollout: %s: its %s attempt onto %s is no longer this scheduler's (%s/%s): the route is not a managed route of the app now; not running it", d.ID.Host, d.ID.Cause, d.ID.Build, s.key.acct, s.key.app)
		s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "route no longer managed"}, "")
		return
	}
	if movesBuild(d.ID.Cause) && rt.GoodVersion != "" && d.ID.Build == rt.GoodVersion {
		if mark, ok := holdRanFor(m.d.Book, s.key.acct, s.key.app, d.ID.Host); ok && mark.Build != d.ID.Build {
			// Ruling 205 at the scheduler (Task 3 fix round 1): a move onto
			// the installed good build over another build's hold-ran marker
			// is the planner's refused case, for an awake instance, a worker
			// or a sidecar member. Root's sibling repoint installs the good
			// build too, so only a signed rollback may run it. The queue
			// derives such a move as classRecover (the instance has served
			// its good build), so the gate is on the cause, not the class: a
			// restart or config attempt re-runs the build the instance
			// serves and is not gated. Nothing ran: abandoned unexecuted, so
			// done() defers the host (abandonDeferral) and the next
			// derivation tries again. The pointer pass, due at once on the
			// install (observe), clears the marker on a signed rollback.
			s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: "a hold build ran here; waiting for a signed rollback"}, "")
			return
		}
	}
	a, err := m.resolve(ctx, d, *rt)
	if err != nil {
		m.stagingRefused(*rt, d.ID.Build, err)
		s.mu.Lock()
		s.q.deferHost(d.ID.Host, m.d.Now().Add(stagingDeferral))
		s.mu.Unlock()
		s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: err.Error()}, "")
		return
	}
	// The failing pointer's PromotedAt, read now, while nothing is at stake:
	// done() publishes a failure without a blocking read (spec §8.1).
	if d.Class == classAdopt {
		d.FailedUnder = m.failedUnder(ctx, *rt, d.ID.Build)
	}
	// The execution's start, after every read and before the activator runs
	// anything: a hold-ran marker this attempt leaves carries it (ruling 205),
	// so a rollback signed while it runs is newer than the marker. The book's
	// snapshot is taken at the same point, so both describe one start: the
	// attempt's good write clears only markers set before it (Astra R3 N6),
	// and its failure is stale after a retry since (ruling 191). ExecID names
	// this execution for a marker it leaves (plan review finding 4).
	d.Started = m.d.Now()
	d.BookSeq = m.d.Book.Snapshot()
	d.ExecID = m.nextExecID()
	if d.Class == classAdopt {
		// Write-ahead (Task 5, Astra R1 #5): the adoption's intent is on disk
		// before the activator stops or starts anything, so a failure whose
		// own save fails is still found by an agent that restarts before a
		// save lands (OpenFailureBook). Its mode is the one the activator
		// will run (plan review finding 2): Swap runs an overlap on a
		// foreign-layout row as a hold, and a reconciled hold holds the
		// instance down. A worker's replacement runs its attempt's mode as
		// given. No intent, no execution: an attempt whose intent could not
		// be written is abandoned unexecuted, and done() defers and retries
		// it as any refusal before the exec. Not under scheduler.mu.
		mode := a.Mode
		if !rt.IsSidecar() {
			mode = m.d.Act.EffectiveMode(rt.Host, a.Mode)
		}
		in := Intent{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Mode: string(mode), ID: d.ExecID, Start: d.Started}
		if err := m.d.Book.Begin(d.ID.Host, in); err != nil { // rolloutsim:M10
			m.d.Logf("rollout: %s: write the intent of its %s attempt onto %s: %v; not running it", d.ID.Host, d.ID.Cause, d.ID.Build, err)
			reason := "failure records could not be written"
			if errors.Is(err, ErrIntentStands) {
				reason = "an earlier execution on this instance has not recorded its outcome"
			}
			s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: reason}, "")
			return
		}
	}
	if rt.IsSidecar() {
		// Workers and native sidecar members keep stop-then-start on the
		// same entry (spec §7.9). done() runs the cohort-failure transition
		// synchronously; only then may a fallback's identity be published
		// (spec §8.1), which the activator held back for this call.
		s.done(d, s.executeWorker(ctx, *rt, a), a.Mode)
		m.d.Act.PublishHeldServing(rt.Host)
		return
	}
	m.emit(*rt, EvSwapping, "info", textSwapping(a.Build, a.Cause))
	s.done(d, m.d.Act.Swap(ctx, a, s.newest), a.Mode)
}

// movesBuild reports whether an attempt of cause moves the instance onto
// its desired build (a release, install or retry), rather than re-running
// the build it serves (a restart or config attempt).
func movesBuild(c activator.Cause) bool {
	return c != activator.CauseRestart && c != activator.CauseConfig
}

// done records an attempt's outcome and acts on it. mode is the attempt's
// swap mode, "" when no Swap ran.
func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.SwapMode) {
	m := s.m
	if res.Mode != "" {
		// The mode the attempt actually ran in: an overlap attempt on a
		// foreign-layout row runs as a hold (spec §2.9), and a failed hold
		// is held down (§6.4, §8.3).
		mode = res.Mode
	}
	failed := res.Outcome == activator.SwapFailed || res.Outcome == activator.SwapDown
	// A failure is published with no blocking read before it (spec §8.1):
	// the failing pointer's PromotedAt was read before the attempt ran
	// (d.FailedUnder), and the record is written in the same critical
	// section that frees the slot and halts the cohort. No admission, no
	// newest re-check and no wake plan (which reads the book) sees the slot
	// free before the record. The record's build and host are the
	// dispatch's, never res.Gen's: a failed hold returns no generation.
	var rec *FailureRecord
	if failed {
		r := FailureRecord{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Host: d.ID.Host,
			Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, PromotedAt: d.FailedUnder, At: m.d.Now(),
			Seq: d.BookSeq, HoldRanAt: d.Started, HoldRanExec: d.ExecID}
		if r.PromotedAt.IsZero() {
			// No pointer named the build when the attempt started: the
			// failure's own time, so only a pointer signed after it clears
			// the record.
			r.PromotedAt = r.At
		}
		if res.Outcome == activator.SwapDown && mode == activator.ModeHold {
			// A hold build executed against this instance's database:
			// nothing falls back on it automatically (spec §6.4, §8.3).
			// Held down in the same step that frees the slot, so no wake
			// planned in between can boot the failed hold build.
			r.HeldDown = []string{d.ID.Host}
		}
		rec = &r
	}
	// A failed attempt that carried a config revision leaves it delivered and
	// not active (spec §8.1). Its backoff is counted here, before the slot
	// frees, and set on the queue in the same step that frees it, so the
	// loop cannot re-admit the attempt in between. A failed adoption goes
	// through the cohort-failure transition instead, and the next attempt
	// for the host is its recovery onto the serving build with the config.
	var (
		cfgWait    time.Duration
		cfgCounted bool
	)
	if failed && d.Class == classRecover && d.ID.ConfigRev != "" {
		cfgWait, cfgCounted = m.configFailure(d.ID.Host, d.ID.ConfigRev, res.Reason)
	}
	if res.Outcome == activator.SwapDone {
		// The config the attempt started with is active, published under the
		// same rule as the serving state (spec §8.1): before the slot frees,
		// so a newer config attempt cannot complete first and be overwritten
		// here, and with no registry read in front of it, so an unreadable
		// row loses nothing.
		m.configActivated(d.ID.Host, d.ID.ConfigRev)
		// Published BEFORE the slot is freed (spec §8.5): while the attempt
		// still holds its host's slot no newer attempt can be admitted and
		// complete first, so this write can never land on top of a newer
		// one. writeGood also requires the generation to be the one the
		// activator serves (acknowledged), and clears only a hold-ran marker
		// set before this attempt started (d.BookSeq, Astra R3 N6).
		if res.CleanExit {
			// A one-shot replacement that exited cleanly (spec §6.5): no
			// generation is left for the activator to acknowledge, so the
			// acknowledged check is skipped, and only that check (see
			// servedOneShot for why that is safe under the held slot).
			m.servedOneShot(d.ID.Host, res.Gen, d.BookSeq)
		} else {
			m.served(d.ID.Host, res.Gen, d.BookSeq)
		}
	}
	now := m.d.Now()
	s.mu.Lock()
	transition := s.q.finish(d, res.Outcome)
	var (
		recorded   RecordResult
		superseded []string
		recErr     error
		markErr    error
	)
	// A hold build that ran here and neither passed nor was held down (an
	// attempt superseded or stopped after its exec) leaves the instance's
	// hold-ran marker (final review wave 2, N2; spec §6.4): the build may
	// have migrated the database, so no later wake may boot the good build
	// in its place. Only an adoption sets it: a restart or config attempt
	// re-runs a build the instance has already served. Set in the step that
	// frees the slot, with the failure record when there is one (one save),
	// so no wake planned after the slot frees misses it.
	heldDown := transition && rec != nil && len(rec.HeldDown) > 0
	mark := mode == activator.ModeHold && d.Class == classAdopt && res.Executed &&
		res.Outcome != activator.SwapDone && !heldDown
	stale := false
	if transition && rec != nil {
		// The whole cohort-failure transition (spec §8.1 steps 1 and 2): the
		// record (in memory at once, on disk as it can) and the halt. None of
		// it runs for an attempt that began before a retry of its build
		// (PreRetry, ruling 191): the book sets its marker, if any, or one
		// in place of its hold-down (r.HoldRanExec, r.HoldRanAt), and
		// nothing else.
		r := *rec
		r.MarkHoldRan = mark
		recorded, superseded, recErr = failCohort(m.d.Book, s.q, r)
		stale = recorded == PreRetry
	} else if mark {
		markErr = m.d.Book.MarkHoldRan(d.ID.Host, HoldRan{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Exec: d.ExecID, At: d.Started})
	}
	added := transition && rec != nil && recorded == Recorded
	if res.Outcome == activator.SwapAbandoned && !res.Executed {
		// It is proven that nothing of the build ran (a refusal before its
		// exec, a staging refusal, an unreadable row): a retried host's
		// allowance is not spent, and its promised canary still runs after
		// the deferral (final review F8). Executed true means the build ran
		// or may have (final review wave 2, N4: an uncertain start, a respawn
		// whose end the attempt could not learn), and that spends it.
		s.q.rearmRetry(d)
	}
	h := s.q.hosts[d.ID.Host]
	if cfgCounted && h != nil && h.wantConfig == d.ID.ConfigRev {
		s.q.configFailed(d.ID.Host, now.Add(cfgWait))
	}
	if res.Outcome == activator.SwapAbandoned && h != nil && h.buildRev == d.SchedRev && h.notBefore.Before(now.Add(abandonDeferral)) {
		// Abandoned while still the host's newest: the same attempt would be
		// derived again at once (plan 3's deferral). A staging refusal has
		// already deferred the host by as long.
		s.q.deferHost(d.ID.Host, now.Add(abandonDeferral))
	}
	serving := ""
	if h != nil {
		serving = h.serving
	}
	s.mu.Unlock()
	// The adoption's intent ends now (Task 5 rule 2), after every book write
	// its outcome makes (the record, the hold-ran marker, served's marker
	// clear above), never before, and off scheduler.mu. If those writes
	// failed to save, this save fails too and the intent stays on disk, where
	// a restart finds it. Ended before the poke below, so the next admission
	// for the host does not meet it.
	var endErr error
	if d.Class == classAdopt && d.ExecID != "" {
		endErr = m.d.Book.End(d.ID.Host, d.ExecID)
	}
	if mark || (stale && heldDown) {
		// A marker was set: the requested one, or one the book set in place
		// of a stale hold-down (Task 4 fix round 1).
		m.pointerReadDue(s.key)
	}
	defer s.poke()
	defer m.changed() // runs first: the slot is free and the record is written
	if recErr != nil {
		m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, recErr)
	}
	if stale {
		m.d.Logf("rollout: %s: %s attempt onto %s failed, but it began before a retry of the build: the retry stands", d.ID.Host, d.ID.Cause, d.ID.Build)
	}
	if markErr != nil {
		m.d.Logf("rollout: %s: write hold-ran marker for %s: %v (held in memory)", d.ID.Host, d.ID.Build, markErr)
	}
	if endErr != nil {
		m.d.Logf("rollout: %s: end the intent of its attempt onto %s: %v (it stays on disk until a save lands)", d.ID.Host, d.ID.Build, endErr)
	}
	if len(superseded) > 0 {
		m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded)
	}
	if failed {
		// The raw reason is for the operator; members get the reason word.
		m.d.Logf("rollout: %s: %s attempt onto %s %s (%s mode): %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Outcome, mode, res.Reason)
	}
	rt := s.route(d.ID.Host)
	if added {
		m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason)))
	}
	switch res.Outcome {
	case activator.SwapDone:
		m.mu.Lock()
		if m.retried[d.ID.Host] == d.ID.Build {
			delete(m.retried, d.ID.Host)
		}
		m.mu.Unlock()
		m.emit(rt, EvSwapped, "info", textSwapped(d.ID.Build, res.Drain))
	case activator.SwapFailed, activator.SwapDown:
		if res.Outcome == activator.SwapDown {
			m.emit(rt, EvSwapDown, "error", textSwapDown(d.ID.Build))
		} else {
			m.emit(rt, EvSwapFailed, "warn", textSwapFailed(d.ID.Build, serving, reasonWord(res.Reason)))
		}
		if cfgCounted {
			m.emit(rt, EvConfigPending, "warn", textConfigPending(reasonWord(res.Reason), cfgWait))
		}
	case activator.SwapAbandoned:
		m.d.Logf("rollout: %s: %s attempt onto %s abandoned: %s", d.ID.Host, d.ID.Cause, d.ID.Build, res.Reason)
	}
}

// route is host's row for an event. A row that cannot be read still gets its
// events, addressed by what the scheduler knows: its app and the host.
func (s *scheduler) route(host string) registry.Route {
	if rt, err := s.m.d.Routes.Get(host); err == nil && rt != nil {
		return *rt
	}
	return registry.Route{Host: host, Account: s.key.acct, App: s.key.app}
}