package rollout

import (
	"sort"
	"time"

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

// Identity is an attempt's identity (spec §6.3) without the scheduler
// revision, which travels beside it in dispatch. Two submissions with equal
// identities are one attempt.
type Identity struct {
	Host      string
	Cause     activator.Cause
	Build     string
	ConfigRev string
	RestartID string
}

// attemptClass separates the two failure policies of spec §8.1.
type attemptClass int

const (
	// classAdopt is a release, install or retry onto a build this instance
	// has not served: its failure writes a record and halts the cohort.
	classAdopt attemptClass = iota
	// classRecover is a restart or config attempt, or a move back onto a
	// build the instance has served: its failure writes no record.
	classRecover
)

// dispatch is one attempt the queue has handed out to run.
type dispatch struct {
	ID       Identity
	SchedRev uint64
	Canary   bool
	Class    attemptClass
	Cohort   string
	// FailedUnder is the PromotedAt of the pointer the attempt's build was
	// promoted under, read by the scheduler (Task 8) before the attempt
	// runs, so a failure is published with no blocking read in between
	// (spec §8.1, §8.4). Zero: the failure's own time stands in.
	FailedUnder time.Time
	// Forced: the attempt is the adoption a `carlos instances retry`
	// allowance (hostSlot.forceBuilds) let through, spent when it started.
	// An attempt that ends with nothing of its build run gets it back
	// (rearmRetry; final review F8).
	Forced bool
	// Started is when execute handed the attempt to the activator (Swap or
	// ReplaceWorker): its execution's start, the At of a hold-ran marker it
	// leaves (ruling 205, Task 3 fix round 1). Zero until then.
	Started time.Time
	// BookSeq is the failure book's snapshot (FailureBook.Snapshot), taken
	// with Started: the order of this execution's start against the book's
	// markers and retry clears. Its good write clears only a marker set
	// before it (Astra R3 N6), and its failure is stale after a retry since
	// (ruling 191). Zero until then.
	BookSeq uint64
	// ExecID is this execution's identity, "a<incarnation>-<n>" from a
	// Manager-wide counter (Manager.nextExecID), set with Started. The
	// incarnation keeps it unique across agent restarts (final review round
	// 1, finding 1). A hold-ran marker it leaves
	// carries it (HoldRan.Exec). SchedRev cannot serve: an abandoned
	// attempt is derived again with the same revision.
	ExecID string
}

// hostSlot is one instance's desired state. The attempt to run is derived
// from it (pending), never stored, so supersession and coalescing are the
// same rule: the newest desire wins.
type hostSlot struct {
	host    string
	since   time.Time // its place in the queue: when its current desire appeared
	serving string    // the current generation's build; "" when not awake
	good    string    // GoodVersion

	wantBuild   string
	wantCause   activator.Cause
	wantRestart string
	wantConfig  string
	configAfter time.Time // a failed config attempt's backoff (spec §8.1)
	notBefore   time.Time // a staging refusal's wait (spec §4.2)

	running  *dispatch
	buildRev uint64 // newest(host): moved only by a change of desired build
	// forceBuilds are `carlos instances retry`'s canary entitlements, one per
	// retried build: each lets the adoption onto its build through a halt,
	// as a canary. Each is its build's, whatever the slot wants when the
	// retry lands, and is spent only by that adoption, never by an attempt
	// onto another build or a restart or config attempt (final review round
	// 1, finding 5). They are kept per build because a retry of one build
	// must not take away another's (final review round 2, finding 5,
	// Astra): the book lists the host as retried for the first build, so a
	// second retry of it clears nothing and could not give the entitlement
	// back. dropRetries removes the ones the book no longer lists, so the
	// set holds at most one entry per record that lists the host.
	forceBuilds map[string]bool
}

// cohort is one app's attempts on this box onto one build, of one class
// (spec §3).
type cohort struct {
	canary  string // the host of the cohort's canary, "" until one starts
	settled bool   // the canary passed its health check and its settle
	halted  bool
	reason  string
}

// appQueue is one app's attempts on this box. It is not safe for concurrent
// use: the scheduler holds its lock around every call.
type appQueue struct {
	acct, app   string
	concurrency int
	hosts       map[string]*hostSlot
	cohorts     map[string]*cohort
	running     int
	rev         uint64
	// blocks is the failure book's answer for this app (spec §8.4).
	blocks func(build, host string, served bool) bool
}

func newAppQueue(acct, app string, blocks func(build, host string, served bool) bool) *appQueue {
	return &appQueue{acct: acct, app: app, concurrency: 1, hosts: map[string]*hostSlot{}, cohorts: map[string]*cohort{}, blocks: blocks}
}

func (q *appQueue) nextRev() uint64 {
	q.rev++
	return q.rev
}

func (q *appQueue) slot(host string, now time.Time) *hostSlot {
	h := q.hosts[host]
	if h == nil {
		h = &hostSlot{host: host, since: now}
		q.hosts[host] = h
	}
	return h
}

// newest is the revision an attempt past spawn must still carry when it
// switches (spec §6.3). Only a change of desired build moves it: a restart
// or config request that arrives meanwhile queues behind the attempt instead
// of abandoning it.
func (q *appQueue) newest(host string) uint64 {
	if h := q.hosts[host]; h != nil {
		return h.buildRev
	}
	return 0
}

// observe records the agent pass's view of one host. A running attempt owns
// the serving build until it finishes.
func (q *appQueue) observe(host, serving, good string, now time.Time) {
	h := q.slot(host, now)
	if h.running == nil {
		h.serving = serving
	}
	h.good = good
}

func (q *appQueue) touch(h *hostSlot, now time.Time) {
	if _, queued := q.pending(h); !queued && h.running == nil {
		h.since = now
	}
}

// wantBuild sets host's desired build: its installed build (spec §3). cause
// is release, install or retry.
func (q *appQueue) wantBuild(host string, cause activator.Cause, build string, now time.Time) bool {
	if build == "" {
		return false
	}
	h := q.slot(host, now)
	if h.wantBuild == build {
		if cause == activator.CauseRetry && h.wantCause != activator.CauseRetry {
			h.wantCause = cause
			return true
		}
		return false
	}
	q.touch(h, now)
	h.wantBuild, h.wantCause = build, cause
	h.buildRev = q.nextRev() // rolloutsim:M4
	return true
}

// wantRestart records a `carlos restart` request; the newest id covers the
// older ones (spec §6.3).
func (q *appQueue) wantRestart(host, id string, now time.Time) bool {
	h := q.slot(host, now)
	if id == "" || h.wantRestart == id {
		return false
	}
	q.touch(h, now)
	h.wantRestart = id
	return true
}

// wantConfig records a delivered config revision that is not yet active. A
// newer revision replaces the pending attempt and its backoff (spec §8.1).
func (q *appQueue) wantConfig(host, rev string, now time.Time) bool {
	h := q.slot(host, now)
	if h.wantConfig == rev {
		return false
	}
	q.touch(h, now)
	h.wantConfig, h.configAfter = rev, time.Time{}
	return true
}

// configActive clears the pending config attempt once rev is active.
func (q *appQueue) configActive(host, rev string) {
	if h := q.hosts[host]; h != nil && h.wantConfig == rev {
		h.wantConfig = ""
	}
}

// dropConfig clears whatever config attempt is pending for host, whichever
// revision it names. It is for a host with no process: its next wake starts
// with the delivered file. It never creates a slot.
func (q *appQueue) dropConfig(host string) {
	if h := q.hosts[host]; h != nil {
		h.wantConfig, h.configAfter = "", time.Time{}
	}
}

// configFailed holds a config-only attempt back until its backoff ends.
func (q *appQueue) configFailed(host string, until time.Time) {
	if h := q.hosts[host]; h != nil {
		h.configAfter = until
	}
}

// deferHost holds every attempt for host back until until: a staging refusal
// waits one stage-relay window for root to restage.
func (q *appQueue) deferHost(host string, until time.Time) {
	if h := q.hosts[host]; h != nil {
		h.notBefore = until
	}
}

// pending derives the attempt host should run next, if any. Only an awake
// instance with nothing running has one. A desired build the failure book
// holds falls through to the instance's restart or config work, which goes
// ahead on the build it serves (spec §8.4).
func (q *appQueue) pending(h *hostSlot) (dispatch, bool) {
	if h.running != nil || h.serving == "" {
		return dispatch{}, false
	}
	id := Identity{Host: h.host, ConfigRev: h.wantConfig, RestartID: h.wantRestart}
	switch {
	case h.wantBuild != "" && h.wantBuild != h.serving && !q.blocks(h.wantBuild, h.host, h.wantBuild == h.good):
		id.Cause, id.Build = h.wantCause, h.wantBuild
	case h.wantRestart != "":
		id.Cause, id.Build = activator.CauseRestart, h.serving
	case h.wantConfig != "":
		id.Cause, id.Build = activator.CauseConfig, h.serving
	default:
		return dispatch{}, false
	}
	class := classRecover // rolloutsim:M2
	if id.Build != h.serving && id.Build != h.good {
		class = classAdopt
	}
	return dispatch{ID: id, SchedRev: h.buildRev, Class: class, Cohort: cohortKey(id.Build, class)}, true
}

// cohortKey names a cohort (spec §3): the attempts onto one build, split by
// class so a halt of the adoption cohort never stops the restarts and config
// changes of instances already serving the build.
func cohortKey(build string, class attemptClass) string {
	if class == classAdopt {
		return "adopt:" + build
	}
	return "recover:" + build
}

func (q *appQueue) cohortFor(key string) *cohort {
	c := q.cohorts[key]
	if c == nil {
		c = &cohort{}
		q.cohorts[key] = c
	}
	return c
}

// ordered is the queue order: oldest desire first, host name to break ties.
func (q *appQueue) ordered() []*hostSlot {
	out := make([]*hostSlot, 0, len(q.hosts))
	for _, h := range q.hosts {
		out = append(out, h)
	}
	sort.Slice(out, func(i, j int) bool {
		if !out[i].since.Equal(out[j].since) {
			return out[i].since.Before(out[j].since)
		}
		return out[i].host < out[j].host
	})
	return out
}

// admit hands out the attempts that may start now (spec §6.2). view maps each
// host the activator could see to the build it serves ("" when asleep); a
// host that is waking or stopping is absent, and keeps its place. A sleeper
// loses only what needs a running process: its restart request.
func (q *appQueue) admit(view map[string]string, now time.Time) []dispatch {
	var out []dispatch
	for _, h := range q.ordered() {
		if h.running != nil {
			continue
		}
		serving, seen := view[h.host]
		if !seen {
			continue
		}
		h.serving = serving
		if serving == "" {
			h.wantRestart = ""
			continue
		}
		d, ok := q.gate(h, now)
		if !ok {
			continue
		}
		if q.running >= q.concurrency { // rolloutsim:M1
			break
		}
		q.start(h, d)
		out = append(out, d)
	}
	return out
}

// gate is the per-host half of admit, shared with the worker recycle hook:
// the host's pending attempt, if its cohort lets it run now. pending does not
// read the two waits, so they are enforced here: a staging refusal holds
// every attempt for the host until notBefore, and a failed config attempt
// waits out its backoff. A cohort's first attempt is its canary and runs
// alone until it settles, and that holds for a retried host too: a retry
// makes the host's next adoption a canary and lifts the halt for it, but it
// still waits for any other canary of the cohort that is running and
// unsettled. A halted adoption cohort admits nothing but such a retried host.
func (q *appQueue) gate(h *hostSlot, now time.Time) (dispatch, bool) {
	d, ok := q.pending(h)
	if !ok || now.Before(h.notBefore) {
		return dispatch{}, false
	}
	if d.ID.Cause == activator.CauseConfig && now.Before(h.configAfter) {
		return dispatch{}, false
	}
	c := q.cohortFor(d.Cohort)
	// The flag is tied to the build it was set for, so a retry on a host with
	// nothing pending cannot lift a later cohort's halt or turn an unrelated
	// attempt into a canary.
	forced := d.Class == classAdopt && h.forceBuilds[d.ID.Build]
	switch {
	case forced:
		// The next attempt after a retry is a canary (spec §8.4) whether or
		// not the cohort is halted: halts live in memory and a restart loses
		// them, and a second retry can follow a settled canary. It still
		// waits out another host's running canary.
		if c.canary != "" && c.canary != h.host && !c.settled {
			return dispatch{}, false
		}
		d.Canary, d.Forced = true, true
	case c.halted && d.Class == classAdopt:
		return dispatch{}, false
	case c.canary == "":
		d.Canary = true
	case !c.settled:
		return dispatch{}, false
	}
	return d, true
}

// start marks d running. A cohort's canary is claimed here, at the moment it
// actually starts, so a host gated out by the concurrency limit never holds
// the canary slot. A retried host's canary claims it like any other, and a
// settled flag left over from before the halt starts again from false.
func (q *appQueue) start(h *hostSlot, d dispatch) {
	if d.Canary {
		c := q.cohortFor(d.Cohort)
		c.canary, c.settled = h.host, false
	}
	if d.Forced {
		// Only the adoption the entitlement let through spends it (final
		// review round 1, finding 5): a restart, a config attempt or an
		// adoption onto another build starting first leaves it for later,
		// and the other builds' entitlements stay (round 2, finding 5).
		delete(h.forceBuilds, d.ID.Build)
	}
	h.running = &d
	q.running++
}

// finish records d's outcome and reports whether it is a build failure that
// must go through the cohort-failure transition (spec §8.1): a failed
// adoption attempt. A failed restart or config attempt is reported, never
// recorded.
func (q *appQueue) finish(d dispatch, out activator.SwapOutcome) (transition bool) {
	h := q.hosts[d.ID.Host]
	if h == nil || h.running == nil || h.running.SchedRev != d.SchedRev || h.running.ID != d.ID {
		return false
	}
	h.running = nil
	q.running--
	c := q.cohorts[d.Cohort]
	unclaim := func() {
		if c != nil && c.canary == h.host && !c.settled {
			c.canary = ""
		}
	}
	switch out {
	case activator.SwapDone:
		h.serving = d.ID.Build
		if h.wantRestart == d.ID.RestartID {
			h.wantRestart = ""
		}
		if d.ID.ConfigRev != "" && h.wantConfig == d.ID.ConfigRev {
			h.wantConfig = ""
		}
		if c != nil && c.canary == h.host {
			c.settled = true
		}
	case activator.SwapAbandoned:
		unclaim()
	case activator.SwapFailed, activator.SwapDown:
		if out == activator.SwapDown {
			h.serving = ""
		}
		if d.ID.Cause == activator.CauseRestart && h.wantRestart == d.ID.RestartID {
			h.wantRestart = ""
		}
		unclaim()
		return d.Class == classAdopt
	}
	return false
}

// halt stops build's adoption cohort, and supersedes every adoption attempt
// onto build already past spawn, so its re-check abandons it rather than
// switching (spec §8.1 step 2). It returns the superseded hosts.
func (q *appQueue) halt(build, reason string) (superseded []string) {
	c := q.cohortFor(cohortKey(build, classAdopt))
	c.halted, c.reason = true, reason
	for _, h := range q.hosts {
		if h.running != nil && h.running.Class == classAdopt && h.running.ID.Build == build {
			h.buildRev = q.nextRev()
			superseded = append(superseded, h.host)
		}
	}
	sort.Strings(superseded)
	return superseded
}

// unhalt reopens build's adoption cohort after `carlos deploy --retry`: its
// next attempt is a fresh canary. A canary still running (a retried host's
// swap) stays the cohort's canary, so nothing else starts until it settles.
func (q *appQueue) unhalt(build string) {
	if c := q.cohorts[cohortKey(build, classAdopt)]; c != nil {
		running := ""
		if !c.settled {
			running = c.canary
		}
		*c = cohort{canary: running}
	}
}

// retryHost is `carlos instances retry <host>` naming build: the host's
// adoption onto build runs as a canary even through a halt, now if the slot
// wants build, or whenever it comes to want it (final review round 1,
// finding 5, Astra: a retry of a build the slot did not want yet used to
// leave the host exempt in the book but with no admission through the
// halt, and no later retry could give it one). The entitlement is bound to
// build: it never lets an adoption onto another build through, and a
// retry of another build leaves it standing (final review round 2, finding
// 5).
func (q *appQueue) retryHost(host, build string) {
	if h := q.hosts[host]; h != nil && build != "" {
		h.arm(build)
	}
}

// arm binds a retry entitlement for build to h.
func (h *hostSlot) arm(build string) {
	if h.forceBuilds == nil {
		h.forceBuilds = map[string]bool{}
	}
	h.forceBuilds[build] = true
}

// dropRetries drops every entitlement whose build the failure book no
// longer lists the host as retried for (retried false): the record was
// cleared by a signed pointer or pruned, or a later failure of the host
// spent the retry. Such an entitlement could only make a later record of
// the build, one no retry has touched, admit the host through its halt
// (final review round 2, finding 5). The caller holds the scheduler's
// lock; retried may take the book's, which comes after it.
func (q *appQueue) dropRetries(retried func(build, host string) bool) {
	for _, h := range q.hosts {
		for build := range h.forceBuilds {
			if !retried(build, h.host) {
				delete(h.forceBuilds, build)
			}
		}
		if len(h.forceBuilds) == 0 {
			h.forceBuilds = nil
		}
	}
}

// rearmRetry gives back the retry allowance a forced attempt spent at its
// start (final review F8), for an attempt that ended with nothing of its
// build run: abandoned before its exec, or refused as a staging error. Spec
// §8.4 promises the retried host's next adoption runs as a canary, and that
// adoption has not happened yet. The allowance stays bound to the build,
// and comes back even when the host now wants another build, as a retry
// for a build not yet wanted keeps it (finding 5). Other builds'
// allowances are untouched (round 2, finding 5). If the book has stopped
// listing the host as retried for the build meanwhile, the next pass drops
// it again (dropRetries). An attempt whose build ran spent the allowance
// for good, whatever its outcome. The caller holds the scheduler's lock,
// in the step that frees the attempt's slot.
func (q *appQueue) rearmRetry(d dispatch) {
	h := q.hosts[d.ID.Host]
	if !d.Forced || h == nil {
		return
	}
	h.arm(d.ID.Build)
}

// nextDue is the earliest moment a deferred attempt becomes admissible, so the
// scheduler's goroutine knows when to look again; zero when nothing waits.
func (q *appQueue) nextDue(now time.Time) time.Time {
	var next time.Time
	consider := func(t time.Time) {
		if t.After(now) && (next.IsZero() || t.Before(next)) {
			next = t
		}
	}
	for _, h := range q.hosts {
		consider(h.notBefore)
		if h.wantConfig != "" {
			consider(h.configAfter)
		}
	}
	return next
}

// recorder is the failure book's write half, an interface so a test can
// watch the transition's order.
type recorder interface {
	Record(FailureRecord) (RecordResult, error)
}

// failCohort is the cohort-failure transition of spec §8.1, in its order:
// (1) the failure record, before anything else can act on the failure;
// (2) queued attempts onto the build are cancelled, because the record now
// blocks them, and attempts past spawn are superseded; (3) wakes of the
// cohort boot the good build, which needs nothing here: the planner reads
// the record, and nothing is repointed; (4) instances that have served the
// build are untouched, because the record never blocks them. The caller
// holds the scheduler's lock for q, so the record, the halt and the freed
// slot (finish) are one critical section. A failed file write still returns
// the record held in memory, so the transition never stops halfway.
//
// A failure the book reports PreRetry began before a retry that has since
// cleared its build (ruling 191): it is stale, and none of the transition
// runs (plan review finding 3). Halting would stop the cohort the retry
// reopened and supersede attempts the retry let start, though the record
// that would justify it was never written. Every other result halts, an
// Unchanged one included: the record it found still holds the build, and a
// halt lives in memory beside it (an agent restart loses it).
func failCohort(book recorder, q *appQueue, rec FailureRecord) (res RecordResult, superseded []string, err error) {
	res, err = book.Record(rec) // rolloutsim:M3
	if res == PreRetry {        // rolloutsim:M11
		return res, nil, err
	}
	superseded = q.halt(rec.Build, rec.Reason)
	return res, superseded, err
}