package rollout

import (
	"testing"
	"time"

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

// currentView is every slot as the activator would report it.
func currentView(q *appQueue) map[string]string {
	v := map[string]string{}
	for host, h := range q.hosts {
		v[host] = h.serving
	}
	return v
}

func releaseAll(q *appQueue, build string, hosts ...string) {
	for _, h := range hosts {
		q.wantBuild(h, activator.CauseRelease, build, t0)
	}
}

func TestTheCanaryRunsAloneUntilItSettles(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b", "c")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b", "c")
	first := q.admit(currentView(q), t0)
	if len(first) != 1 || !first[0].Canary {
		t.Fatalf("admitted %+v, want exactly one canary", first)
	}
	if more := q.admit(currentView(q), t0); len(more) != 0 {
		t.Fatalf("admitted %+v while the canary has not settled", more)
	}
	q.finish(first[0], activator.SwapDone)
	rest := q.admit(currentView(q), t0)
	if len(rest) != 2 || rest[0].Canary || rest[1].Canary {
		t.Fatalf("after the settle admitted %+v, want the other two and no canary", rest)
	}
}

func TestSwapConcurrencyIsNeverExceeded(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b", "c", "d")
	q.concurrency = 1
	releaseAll(q, "v2", "a", "b", "c", "d")
	// Four attempts need four rounds. The cap turns a queue that stops
	// admitting into a failure instead of a test that spins until its
	// timeout on a shared box.
	const maxRounds = 100
	for round, done := 0, 0; done < 4; round++ {
		if round == maxRounds {
			t.Fatalf("%d of 4 attempts finished after %d admission rounds: the queue stopped admitting", done, maxRounds)
		}
		ds := q.admit(currentView(q), t0)
		if q.running > q.concurrency || len(ds) > 1 {
			t.Fatalf("running %d, admitted %d, with concurrency %d", q.running, len(ds), q.concurrency)
		}
		for _, d := range ds {
			q.finish(d, activator.SwapDone)
			done++
		}
	}
}

func TestAnAbandonedCanaryIsReplaced(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	first := q.admit(currentView(q), t0)
	q.finish(first[0], activator.SwapAbandoned)
	next := q.admit(currentView(q), t0)
	if len(next) != 1 || !next[0].Canary {
		t.Fatalf("after an abandoned canary admitted %+v, want a new canary", next)
	}
}

func TestHaltOnTheFirstFailureSparesInstancesServingTheBuild(t *testing.T) {
	b := &FailureBook{}
	q := newAppQueue("bab", "blog", func(build, host string, served bool) bool {
		return b.Blocks("bab", "blog", build, host, served)
	})
	for _, h := range []string{"a", "b", "c"} {
		q.observe(h, "v1", "v1", t0)
	}
	q.observe("d", "v2", "v2", t0) // d served v2 before it failed elsewhere
	q.concurrency = 3
	releaseAll(q, "v2", "a", "b", "c", "d")
	canary := q.admit(currentView(q), t0)
	if len(canary) != 1 {
		t.Fatalf("admitted %+v", canary)
	}
	if !q.finish(canary[0], activator.SwapFailed) {
		t.Fatal("a failed release attempt goes through the cohort-failure transition")
	}
	if _, _, err := failCohort(b, q, FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: canary[0].ID.Host, At: t0}); err != nil {
		t.Fatal(err)
	}
	q.wantRestart("d", "r1", t0)
	ds := q.admit(currentView(q), t0)
	for _, d := range ds {
		if d.ID.Build == "v2" && d.Class == classAdopt {
			t.Fatalf("admitted %+v after the halt", d.ID)
		}
	}
	if len(ds) != 1 || ds[0].ID.Host != "d" || ds[0].ID.Cause != activator.CauseRestart {
		t.Fatalf("admitted %+v, want only d's restart onto the v2 it serves (spec §8.4)", ds)
	}
}

// orderBook watches the transition's order: when the record is written, the
// attempts past spawn must not yet have been superseded.
type orderBook struct {
	FailureBook
	q        *appQueue
	host     string
	atRecord uint64
}

func (o *orderBook) Record(r FailureRecord) (RecordResult, error) {
	o.atRecord = o.q.newest(o.host)
	return o.FailureBook.Record(r)
}

func TestTheRecordComesBeforeTheSupersession(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b")
	canary := q.admit(currentView(q), t0)
	q.finish(canary[0], activator.SwapDone)
	running := q.admit(currentView(q), t0) // b, now past spawn
	if len(running) != 1 {
		t.Fatalf("admitted %+v", running)
	}
	before := q.newest(running[0].ID.Host)
	ob := &orderBook{q: q, host: running[0].ID.Host}
	_, superseded, _ := failCohort(ob, q, FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", At: t0})
	if ob.atRecord != before {
		t.Fatal("attempts were superseded before the failure record was written (spec §8.1 order)")
	}
	if len(superseded) != 1 || q.newest(running[0].ID.Host) == before {
		t.Fatalf("superseded %v: an attempt past spawn onto the failed build must be abandoned at its re-check", superseded)
	}
}

func TestAWakingInstanceKeepsItsPlaceAndASleeperLosesItsRestart(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.wantRestart("a", "r1", t0)
	q.wantRestart("b", "r1", t0)
	v := currentView(q)
	delete(v, "a") // waking
	v["b"] = ""    // asleep
	if ds := q.admit(v, t0); len(ds) != 0 {
		t.Fatalf("admitted %+v", ds)
	}
	if q.hosts["a"].wantRestart != "r1" {
		t.Fatal("a waking instance keeps its place (spec §6.2)")
	}
	if q.hosts["b"].wantRestart != "" {
		t.Fatal("a restart never wakes a sleeper")
	}
}

func TestCohortsAreKeyedByDesiredBuild(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.concurrency = 2
	q.wantBuild("a", activator.CauseRelease, "v2", t0) // one channel
	q.wantBuild("b", activator.CauseRelease, "v3", t0) // another
	ds := q.admit(currentView(q), t0)
	if len(ds) != 2 || !ds[0].Canary || !ds[1].Canary {
		t.Fatalf("admitted %+v, want each cohort's own canary", ds)
	}
}

func TestARetriedInstanceRunsAsACanaryThroughAHalt(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	q.halt("v2", "health check failed")
	if ds := q.admit(currentView(q), t0); len(ds) != 0 {
		t.Fatalf("admitted %+v from a halted cohort", ds)
	}
	q.retryHost("b", "v2")
	ds := q.admit(currentView(q), t0)
	if len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
		t.Fatalf("admitted %+v, want b alone, as a canary", ds)
	}
	q.unhalt("v2")
	q.finish(ds[0], activator.SwapDone)
	if ds := q.admit(currentView(q), t0); len(ds) != 1 || ds[0].ID.Host != "a" {
		t.Fatalf("after the app retry admitted %+v", ds)
	}
}

func TestAFailedConfigAttemptWaitsOutItsBackoff(t *testing.T) {
	q := awakeQueue(noBlocks, "a")
	q.wantConfig("a", "c1", t0)
	q.configFailed("a", t0.Add(time.Minute))
	if ds := q.admit(currentView(q), t0); len(ds) != 0 {
		t.Fatal("admitted a config attempt inside its backoff")
	}
	if got := q.nextDue(t0); !got.Equal(t0.Add(time.Minute)) {
		t.Fatalf("nextDue = %v, want the backoff's end", got)
	}
	if ds := q.admit(currentView(q), t0.Add(time.Minute)); len(ds) != 1 {
		t.Fatal("the config attempt did not come back after its backoff")
	}
}

// Task 5's pending() does not read notBefore, so admission must: a staging
// refusal's wait holds every attempt for the host, not only config ones.
func TestAStagingRefusalWaitsOutItsDeferral(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b")
	q.deferHost("a", t0.Add(30*time.Second))
	ds := q.admit(currentView(q), t0)
	if len(ds) != 1 || ds[0].ID.Host != "b" {
		t.Fatalf("admitted %+v, want only b: a is deferred", ds)
	}
	if got := q.nextDue(t0); !got.Equal(t0.Add(30 * time.Second)) {
		t.Fatalf("nextDue = %v, want the deferral's end", got)
	}
	q.finish(ds[0], activator.SwapDone)
	if ds := q.admit(currentView(q), t0.Add(29*time.Second)); len(ds) != 0 {
		t.Fatalf("admitted %+v inside the deferral", ds)
	}
	if ds := q.admit(currentView(q), t0.Add(30*time.Second)); len(ds) != 1 || ds[0].ID.Host != "a" {
		t.Fatalf("after the deferral admitted %+v, want a", ds)
	}
	if got := q.nextDue(t0.Add(30 * time.Second)); !got.IsZero() {
		t.Fatalf("nextDue = %v, want zero once nothing waits", got)
	}
}

// A deferred host never holds the cohort's canary slot: the next host in the
// queue takes it, so one refusal does not stall the whole rollout.
func TestADeferredHostDoesNotHoldTheCanarySlot(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	q.deferHost("a", t0.Add(time.Minute))
	ds := q.admit(currentView(q), t0)
	if len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
		t.Fatalf("admitted %+v, want b as the canary", ds)
	}
}

// A forced canary overrides the halt and nothing else: spec §6.2 says no other
// swap in a cohort starts until its first one settles, and swap_concurrency
// counts the canary too. So `deploy --retry` while the retried host's swap
// still runs must not let the next host become a second canary beside it.
func TestARetryThenAnUnhaltLeavesOneCanaryUntilItSettles(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b", "c")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b", "c")
	q.halt("v2", "health check failed")
	q.retryHost("b", "v2")
	forced := q.admit(currentView(q), t0)
	if len(forced) != 1 || forced[0].ID.Host != "b" || !forced[0].Canary {
		t.Fatalf("admitted %+v, want b alone, as a canary", forced)
	}
	q.unhalt("v2")
	if more := q.admit(currentView(q), t0); len(more) != 0 {
		t.Fatalf("admitted %+v beside the retried canary, which has not settled", more)
	}
	q.finish(forced[0], activator.SwapDone)
	rest := q.admit(currentView(q), t0)
	if len(rest) != 2 || rest[0].Canary || rest[1].Canary {
		t.Fatalf("after the settle admitted %+v, want a and c and no canary", rest)
	}
}

// Retrying a host in a cohort that is not halted gives it no way round an
// unsettled canary.
func TestARetriedInstanceWaitsForTheCohortsUnsettledCanary(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b")
	first := q.admit(currentView(q), t0)
	if len(first) != 1 || first[0].ID.Host != "a" || !first[0].Canary {
		t.Fatalf("admitted %+v, want a as the canary", first)
	}
	q.retryHost("b", "v2")
	if more := q.admit(currentView(q), t0); len(more) != 0 {
		t.Fatalf("admitted %+v beside a's unsettled canary", more)
	}
	q.finish(first[0], activator.SwapDone)
	if ds := q.admit(currentView(q), t0); len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
		t.Fatalf("after the settle admitted %+v, want b as a canary", ds)
	}
}

// Spec §8.4: the next attempt after `instances retry` runs as a canary. That
// holds in a cohort that is not halted (a halt lives in memory and an agent
// restart loses it, and a second retry can follow a settled canary), where
// the retried host would otherwise run as an ordinary swap with no settle
// window.
func TestARetriedInstanceIsACanaryInACohortThatIsNotHalted(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	first := q.admit(currentView(q), t0)
	if len(first) != 1 || first[0].ID.Host != "a" || !first[0].Canary {
		t.Fatalf("admitted %+v, want a as the canary", first)
	}
	q.finish(first[0], activator.SwapDone)
	q.retryHost("b", "v2")
	ds := q.admit(currentView(q), t0)
	if len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
		t.Fatalf("admitted %+v, want b as a canary", ds)
	}
	// It claims the slot: the cohort has an unsettled canary again.
	if c := q.cohorts[ds[0].Cohort]; c.canary != "b" || c.settled {
		t.Fatalf("cohort %+v, want b's canary unsettled", *c)
	}
}

// A halted cohort whose canary is still running (halt supersedes it, but it
// has not been abandoned yet) holds a retried host back too.
func TestARetriedInstanceWaitsForARunningCanaryInAHaltedCohort(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	q.concurrency = 2
	releaseAll(q, "v2", "a", "b")
	first := q.admit(currentView(q), t0)
	q.halt("v2", "another host failed")
	q.retryHost("b", "v2")
	if more := q.admit(currentView(q), t0); len(more) != 0 {
		t.Fatalf("admitted %+v beside a running canary", more)
	}
	q.finish(first[0], activator.SwapAbandoned)
	if ds := q.admit(currentView(q), t0); len(ds) != 1 || ds[0].ID.Host != "b" || !ds[0].Canary {
		t.Fatalf("after the abandon admitted %+v, want b as a canary", ds)
	}
}

// Final review F8: a forced attempt that ran nothing gets its allowance
// back, bound to the build it was for. Since final review round 1, finding
// 5, that holds when the host's desire has moved on to another build too:
// the allowance waits for the adoption onto its build, and never lets an
// adoption onto the other build through. An attempt that was never forced
// gets nothing.
func TestARearmedRetryStaysBoundToItsBuild(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b")
	releaseAll(q, "v2", "a", "b")
	q.halt("v2", "health check failed")
	q.retryHost("b", "v2")
	forced := q.admit(currentView(q), t0)
	if len(forced) != 1 || forced[0].ID.Host != "b" || !forced[0].Forced {
		t.Fatalf("admitted %+v, want b's forced canary", forced)
	}
	q.finish(forced[0], activator.SwapAbandoned)
	q.rearmRetry(forced[0])
	again := q.admit(currentView(q), t0)
	if len(again) != 1 || again[0].ID.Host != "b" || !again[0].Canary || !again[0].Forced {
		t.Fatalf("after the re-arm admitted %+v, want b's forced canary again", again)
	}
	q.finish(again[0], activator.SwapAbandoned)
	q.wantBuild("b", activator.CauseRelease, "v3", t0) // a newer build replaced the desire
	q.rearmRetry(again[0])
	if got := q.hosts["b"].forceBuilds; len(got) != 1 || !got["v2"] {
		t.Fatalf("forceBuilds = %v: want the allowance for v2 back, waiting for v2", got)
	}
	q.halt("v3", "health check failed")
	if ds := q.admit(currentView(q), t0); len(ds) != 0 {
		t.Fatalf("admitted %+v: an allowance for v2 let an adoption onto v3 through v3's halt", ds)
	}
	q.rearmRetry(dispatch{ID: Identity{Host: "a", Build: "v2"}}) // never forced
	if got := q.hosts["a"].forceBuilds; len(got) != 0 {
		t.Fatalf("forceBuilds = %v: an attempt that was never forced armed a retry", got)
	}
}

// A retry of another build must not leave a flag that lets an unrelated
// later attempt skip the settle, or a halt (the entitlement is its build's,
// final review round 1, finding 5).
func TestAStaleRetryDoesNotBypassTheSettleOrAHalt(t *testing.T) {
	q := awakeQueue(noBlocks, "a", "b", "c")
	q.concurrency = 3
	q.retryHost("c", "v9") // a retry of a build c does not want: nothing for v2
	releaseAll(q, "v2", "a", "b", "c")
	first := q.admit(currentView(q), t0)
	if len(first) != 1 || first[0].ID.Host != "a" {
		t.Fatalf("admitted %+v, want only a's canary: c's old retry bought nothing", first)
	}
	q.halt("v2", "health check failed")
	q.finish(first[0], activator.SwapFailed)
	if ds := q.admit(currentView(q), t0); len(ds) != 0 {
		t.Fatalf("admitted %+v from a halted cohort on a stale retry", ds)
	}
}