package host

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rollout"
)

// rolloutFixture is an activator over a fresh registry, built with the
// generations switch as gens says (runAgent sets it from swapArming before
// New). The package's rollout state is reset when the test ends.
func rolloutFixture(t *testing.T, gens bool) (*activator.Service, *registry.Registry, string) {
	t.Helper()
	return rolloutFixtureWith(t, gens, &activator.SplitBackend{})
}

// rolloutFixtureWith is rolloutFixture over backend.
func rolloutFixtureWith(t *testing.T, gens bool, backend activator.Backend) (*activator.Service, *registry.Registry, string) {
	t.Helper()
	prev := activator.GenerationsEnabled()
	activator.SetGenerations(gens)
	t.Cleanup(func() { activator.SetGenerations(prev) })
	resetRollout(t)
	dd := t.TempDir()
	reg, err := registry.Open(filepath.Join(dd, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	svc := activator.New(activator.Config{Bucket: "b", Idle: time.Hour}, backend, reg)
	return svc, reg, dd
}

func resetRollout(t *testing.T) {
	t.Helper()
	clear := func() {
		rolloutMgr.Store(nil)
		rolloutDown.Store(nil)
		rolloutDeclined.Store(nil)
		declinedHolds.Store(nil)
		select {
		case <-statusNudge:
		default:
		}
	}
	clear()
	t.Cleanup(clear)
}

func overlapEnv(pub string) func(string) string {
	env := map[string]string{"CARLOS_SWAP": "overlap", "CARLOS_RELEASE_PUBKEY": pub}
	return func(k string) string { return env[k] }
}

func noEnv(string) string { return "" }

// stubPlacementCheck replaces swapArming's arm-time placement check with
// one that answers err, and counts its calls. A temp-dir cgroup root can
// never pass the real check (cgroup.Manager.VerifyPlacement), so every test
// that expects swapArming to arm uses this with a nil err.
func stubPlacementCheck(t *testing.T, err error) *int {
	t.Helper()
	orig := placementCheck
	calls := new(int)
	placementCheck = func(*cgroup.Manager) error { *calls++; return err }
	t.Cleanup(func() { placementCheck = orig })
	return calls
}

// forbidPlacementCheck fails the test if swapArming runs the placement
// check: with the flag off, or after an earlier check declined, it must not.
func forbidPlacementCheck(t *testing.T) {
	t.Helper()
	orig := placementCheck
	placementCheck = func(*cgroup.Manager) error {
		t.Error("swapArming ran the placement check")
		return nil
	}
	t.Cleanup(func() { placementCheck = orig })
}

// armableBackend is an exec backend that passes every check swapArming makes
// before the placement check: claims off, a delegated subtree (over a temp
// root), and a control channel that speaks the generation verbs.
func armableBackend(t *testing.T) *activator.ExecBackend {
	t.Helper()
	root := t.TempDir()
	for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
		if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
			t.Fatal(err)
		}
	}
	cgm, err := cgroup.SetupAt(root, t.Logf)
	if err != nil {
		t.Fatal(err)
	}
	return &activator.ExecBackend{Cgroups: cgm, Control: instancectl.NewClient(filepath.Join(t.TempDir(), "ctl.sock"))}
}

func unusedStore(context.Context) (release.Store, error) { return nil, errors.New("unused") }

func funcPtr(f any) uintptr { return reflect.ValueOf(f).Pointer() }

// legsAreTodays asserts every leg the scheduler could take over is today's:
// the converge leg, the adopt bounce, config sync's callbacks and the
// restart leg's bounce.
func legsAreTodays(t *testing.T, svc *activator.Service, reg *registry.Registry) {
	t.Helper()
	if fastConvergeLeg(svc, reg) == nil {
		t.Error("no converge leg: convergeAdopted must run")
	}
	if b := adoptBounceFn(svc); b == nil || funcPtr(b) != funcPtr(svc.BounceAdopted) {
		t.Error("the adopt bounce is not BounceAdopted")
	}
	if _, ok := rolloutConfigCallbacks(context.Background(), release.NewDirStore(t.TempDir()), configLegs{}); ok {
		t.Error("config sync left today's SyncExec")
	}
	ctx, s := genKit(t)
	routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")}
	var bounced []string
	bounce := func(h string) bool { bounced = append(bounced, h); return true }
	mem := newRestartMemory()
	askRestart(t, ctx, s, "bab", "eleven")
	restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
	askRestart(t, ctx, s, "bab", "eleven")
	restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
	if !slices.Equal(bounced, []string{"exec.example.com"}) {
		t.Errorf("restart bounced %v, want today's bounce of the exec instance", bounced)
	}
}

// Cross-plan review focus 1: the flag off changes nothing.
func TestTheFlagOffArmsNothing(t *testing.T) {
	svc, reg, dd := rolloutFixture(t, false)
	opened := false
	m := armRollout(context.Background(), noEnv, svc, reg, dd, nil,
		func(context.Context) (release.Store, error) { opened = true; return nil, errors.New("unused") })
	if m != nil || armedRollout() != nil || opened {
		t.Fatal("CARLOS_SWAP unset armed the scheduler")
	}
	if svc.Planner != nil || svc.OnGen != nil || svc.OnDrift != nil || svc.CoalesceRecycle != nil || svc.GenerationsFile != "" {
		t.Fatal("CARLOS_SWAP unset set an activator hook")
	}
	if adoptBounceFn(svc) == nil {
		t.Fatal("the adopt bounce must stay BounceAdopted")
	}
	forbidPlacementCheck(t)
	if swapArming(noEnv, nil) || activator.GenerationsEnabled() {
		t.Fatal("CARLOS_SWAP unset turned generations on")
	}
	if r := rolloutStatusProbe(); r != nil {
		t.Fatalf("status with the flag off = %+v, want nil", r)
	}
}

// Carried obligation 5: with the flag off every leg is today's, and the only
// hook set is armAlways's OnWorkerServing.
func TestTheFlagOffLegsAreTodays(t *testing.T) {
	svc, reg, dd := rolloutFixture(t, false)
	armAlways(svc, reg, dd)
	armRollout(context.Background(), noEnv, svc, reg, dd, nil, unusedStore)
	legsAreTodays(t, svc, reg)
	if len(statusNudge) != 0 {
		t.Fatal("the status loop was nudged with the flag off")
	}
}

// PF14 (plan 4 R17): runAgent sets the generations file whatever the flag,
// so a worker's captured serving build is pinned with the flag off too.
// armAlways publishes the worker's serving version and touches nothing
// else; armRollout never sets the file either.
func TestTheFlagOffStillPublishesWorkerArtifacts(t *testing.T) {
	svc, reg, dd := rolloutFixture(t, false)
	armAlways(svc, reg, dd)
	armRollout(context.Background(), noEnv, svc, reg, dd, nil, unusedStore)
	if svc.OnWorkerServing == nil {
		t.Fatal("no worker serving publication with the flag off")
	}
	if svc.GenerationsFile != "" {
		t.Fatalf("generations file = %q: armAlways and armRollout must leave it to runAgent", svc.GenerationsFile)
	}
}

func TestTheFlagOnArmsTheScheduler(t *testing.T) {
	svc, reg, dd := rolloutFixture(t, true)
	svc.GenerationsFile = release.GenerationsPath(dd) // as runAgent sets it, before armRollout
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready",
		Hibernate: true, Version: "v2"}); err != nil {
		t.Fatal(err)
	}
	store := release.NewDirStore(t.TempDir())
	ctx, cancel := context.WithCancel(context.Background()) // ends the schedulers' goroutines
	t.Cleanup(cancel)
	m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil,
		func(context.Context) (release.Store, error) { return store, nil })
	if m == nil || armedRollout() != m {
		t.Fatal("CARLOS_SWAP=overlap did not arm the scheduler")
	}
	if svc.Planner == nil || svc.OnGen == nil || svc.OnDrift == nil || svc.CoalesceRecycle == nil {
		t.Fatal("a hook is missing")
	}
	if svc.GenerationsFile != release.GenerationsPath(dd) {
		t.Fatalf("generations file = %q", svc.GenerationsFile)
	}
	// The legs are the scheduler's.
	if b := adoptBounceFn(svc); b == nil || funcPtr(b) != funcPtr(m.Adopted) {
		t.Fatal("the adopt bounce is not the scheduler's Adopted")
	}
	if fastConvergeLeg(svc, reg) == nil {
		t.Fatal("no converge leg: the scheduler's pass must run")
	}
	cb, ok := rolloutConfigCallbacks(context.Background(), store, configLegs{})
	if !ok || cb.Delivered == nil || cb.Ack == nil || cb.Restart == nil || cb.Restart("a.example.com") != nil {
		t.Fatalf("config callbacks = %+v, %v: want the scheduler's delivery and deferred ack, and no restart", cb, ok)
	}
	// The pass at arming saw the managed route, and the status probe is
	// the scheduler's report.
	r := rolloutStatusProbe()
	if r == nil || r.Unavailable != "" {
		t.Fatalf("status = %+v, want the scheduler's report", r)
	}
	if got := r.Hosts["a.example.com"]; got.Installed != "v2" {
		t.Fatalf("status host = %+v, want the managed route installed on v2", got)
	}
}

// Spec §6.4, §8.3, §8.4, P2 #1: with the flag on, generations on and the
// scheduler's state unavailable, managed starts are refused, the reason
// code reported and the bounce legs stand down. Falling back to today's
// live-link wakes and legacy convergence (a nil Planner) would exec the
// failed installed build the unreadable records hold down.
func TestUnavailableStateRefusesManagedStarts(t *testing.T) {
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	dirStore := func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }
	for name, tc := range map[string]struct {
		env   func(string) string
		store func(context.Context) (release.Store, error)
		book  bool // an unreadable failure book in the data dir
		code  string
	}{
		"no deploy store": {overlapEnv(pub),
			func(context.Context) (release.Store, error) { return nil, errors.New("no deployment store") }, false, rolloutNoStore},
		"no public key":          {overlapEnv(""), dirStore, false, rolloutNoReleaseKey},
		"unreadable book":        {overlapEnv(pub), dirStore, true, rolloutBookUnreadable},
		"book with a bad schema": {overlapEnv(pub), dirStore, true, rolloutBookUnreadable},
	} {
		t.Run(name, func(t *testing.T) {
			svc, reg, dd := rolloutFixture(t, true)
			if tc.book {
				body := "{not json"
				if name == "book with a bad schema" {
					body = `{"schema": 9}`
				}
				if err := os.WriteFile(filepath.Join(dd, rollout.FailureFile), []byte(body), 0o644); err != nil {
					t.Fatal(err)
				}
			}
			m := armRollout(context.Background(), tc.env, svc, reg, dd, nil, tc.store)
			if m != nil || armedRollout() != nil {
				t.Fatal("armed without its required state")
			}
			if svc.Planner == nil {
				t.Fatal("no Planner: the box fell back to today's live-link wakes")
			}
			managed := registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true, Version: "v2"}
			p := svc.Planner(context.Background(), managed)
			if p.Refuse == "" {
				t.Fatalf("managed wake planned %+v, want refused", p)
			}
			if p.Refuse != refuseTexts[tc.code] {
				t.Fatalf("refusal %q, want the fixed text for %s (never the raw error)", p.Refuse, tc.code)
			}
			unit := registry.Route{Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"}
			if p := svc.Planner(context.Background(), unit); p.Refuse != "" || p.Boot.StagedPath != "" {
				t.Fatalf("an unmanaged wake was changed: %+v", p)
			}
			if r := rolloutStatusProbe(); r == nil || r.Unavailable != tc.code {
				t.Fatalf("status = %+v, want the reason code %q", r, tc.code)
			}
			// The bounce legs stand down.
			if adoptBounceFn(svc) != nil {
				t.Fatal("the adopt bounce still stops instances whose next wake is refused")
			}
			if fastConvergeLeg(svc, reg) != nil {
				t.Fatal("legacy convergence still stops instances whose next wake is refused")
			}
			if svc.OnDrift == nil {
				t.Fatal("the sweep's convergence stop is still armed")
			}
			if svc.CoalesceRecycle == nil || svc.CoalesceRecycle("bot.internal") {
				t.Fatal("a worker recycle can still converge onto the live link")
			}
			if svc.OnGen != nil {
				t.Fatal("OnGen set with no scheduler to hear it")
			}
			store := release.NewDirStore(t.TempDir())
			cb, ok := rolloutConfigCallbacks(context.Background(), store, configLegs{})
			// Nothing is awake here: the restart is none, and says the file
			// is read on the next wake (final review F7 (3)).
			if !ok || cb.Delivered != nil || cb.Restart == nil || !errors.Is(cb.Restart("a.example.com"), appconfig.ErrDeliveredOnWake) || cb.Ack == nil {
				t.Fatalf("config callbacks = %+v, %v: want delivery and the ack as today, and no restart", cb, ok)
			}
			ctx, s := genKit(t)
			routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")}
			var bounced []string
			bounce := func(h string) bool { bounced = append(bounced, h); return true }
			mem := newRestartMemory()
			askRestart(t, ctx, s, "bab", "eleven")
			restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
			askRestart(t, ctx, s, "bab", "eleven")
			if !restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem) {
				t.Fatal("the root half of the restart was not asked for")
			}
			if len(bounced) != 0 {
				t.Fatalf("restart bounced %v while managed starts are refused", bounced)
			}
		})
	}
}

// leaveIntent leaves in's intent for host in dd's failure book, as an agent
// that died mid-execution would have.
func leaveIntent(t *testing.T, dd, host string, in rollout.Intent) {
	t.Helper()
	b, err := rollout.OpenFailureBook(dd)
	if err != nil {
		t.Fatal(err)
	}
	if err := b.Begin(host, in); err != nil {
		t.Fatal(err)
	}
}

// Task 5 rule 5: an armed box whose book holds an intent an earlier agent
// left turns it into the failure it may have been, and saves that before
// the scheduler runs: a reopen finds the record and no intent.
func TestAnArmedBoxReconcilesAnIntentLeftInItsBook(t *testing.T) {
	svc, reg, dd := rolloutFixture(t, true)
	svc.GenerationsFile = release.GenerationsPath(dd)
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready",
		Hibernate: true, Version: "v2", GoodVersion: "v1", GoodDigest: "sha256:v1"}); err != nil {
		t.Fatal(err)
	}
	leaveIntent(t, dd, "a.example.com", rollout.Intent{Account: "bab", App: "blog", Build: "v2", Mode: "hold", ID: "a3"})
	store := release.NewDirStore(t.TempDir())
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, func(context.Context) (release.Store, error) { return store, nil })
	if m == nil || rolloutUnavailable() != "" {
		t.Fatalf("not armed (%q)", rolloutUnavailable())
	}
	b, err := rollout.OpenFailureBook(dd)
	if err != nil {
		t.Fatal(err)
	}
	if b.Reconciled() != 0 {
		t.Fatal("the reconciled book was not saved: the intent is still on disk")
	}
	if !b.HeldDown("bab", "blog", "v2", "a.example.com") {
		t.Fatal("the hold intent's failure was not kept")
	}
	if p := svc.Planner(ctx, *mustGet(t, reg, "a.example.com")); p.Refuse != rollout.HeldDownRefusal("v2") {
		t.Fatalf("plan %+v: want the held-down refusal", p)
	}
}

// Task 5 rule 5: a reconciled book the box cannot save refuses every
// managed start, as an unreadable one does: the scheduler must not run while
// a restart would find the intents again with no way to keep their records.
func TestAnArmedBoxThatCannotSaveItsReconciledBookRefusesManagedStarts(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("running as root: a read-only directory does not fail a save")
	}
	svc, reg, dd := rolloutFixture(t, true)
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	leaveIntent(t, dd, "a.example.com", rollout.Intent{Account: "bab", App: "blog", Build: "v2", Mode: "overlap", ID: "w4"})
	if err := os.Chmod(dd, 0o500); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.Chmod(dd, 0o700) })
	m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil,
		func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil })
	if m != nil || armedRollout() != nil {
		t.Fatal("armed over a book it could not save")
	}
	if rolloutUnavailable() != rolloutBookUnreadable {
		t.Fatalf("unavailable %q, want %q", rolloutUnavailable(), rolloutBookUnreadable)
	}
	managed := registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true, Version: "v2"}
	if p := svc.Planner(context.Background(), managed); p.Refuse != refuseTexts[rolloutBookUnreadable] {
		t.Fatalf("plan %+v: want the unreadable-book refusal", p)
	}
}

// Final review round 1, finding 3, restated for the hard rule (readiness
// round 2 Task 1; FailureBook's doc, rules 2 to 5): every open of the book
// owes a save before the scheduler arms, intents or none. A new book starts
// at sequence 0 with a crypto/rand identity and is written as schema 3 with
// durable_epochs; a box that cannot save that could publish epochs it never
// stored, so it refuses managed starts. A box that can save it arms, with
// the identity and the durable marker on disk.
func TestAnArmedBoxSavesItsBookOnOpenOrRefusesManagedStarts(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("running as root: a read-only directory does not fail a save")
	}
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	store := func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }

	svc, reg, dd := rolloutFixture(t, true)
	if err := os.Chmod(dd, 0o500); err != nil {
		t.Fatal(err)
	}
	locked := dd
	t.Cleanup(func() { os.Chmod(locked, 0o700) })
	if m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, store); m != nil || armedRollout() != nil {
		t.Fatal("armed over a book whose open it could not save")
	}
	if rolloutUnavailable() != rolloutBookUnreadable {
		t.Fatalf("unavailable %q, want %q", rolloutUnavailable(), rolloutBookUnreadable)
	}

	svc, reg, dd = rolloutFixture(t, true)
	if m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, store); m == nil {
		t.Fatal("did not arm over a writable data dir")
	}
	raw, err := os.ReadFile(filepath.Join(dd, rollout.FailureFile))
	if err != nil {
		t.Fatalf("the open's advance was not saved before arming: %v", err)
	}
	var f struct {
		Schema  int               `json:"schema"`
		Seq     uint64            `json:"seq"`
		Book    string            `json:"book"`
		Durable bool              `json:"durable_epochs"`
		Records []json.RawMessage `json:"records"`
	}
	if err := json.Unmarshal(raw, &f); err != nil {
		t.Fatalf("saved book %s: %v", raw, err)
	}
	if f.Schema != 3 || !f.Durable {
		t.Fatalf("saved book %s: want schema 3 with durable_epochs true", raw)
	}
	if id, err := hex.DecodeString(f.Book); err != nil || len(id) != 16 {
		t.Fatalf("saved book %s (%v): want a 16-byte hex identity", raw, err)
	}
	// This fixture has no records, intents or markers, so the highest value
	// the book ever held is 0 and a new book hands out none before its first
	// change: the saved sequence is at least that, and no more than the
	// open itself needs (the old 2^20 advance is gone).
	if len(f.Records) != 0 || f.Seq >= 1<<20 {
		t.Fatalf("saved book %s: want an empty new book with a small sequence", raw)
	}
}

// The unreadable-book refusal is one approved text, shared with the
// scheduler's refusal of a wake whose intent it could not write (Task 5).
func TestTheUnreadableBookRefusalIsShared(t *testing.T) {
	if refuseTexts[rolloutBookUnreadable] != rollout.BookUnreadableRefusal {
		t.Fatalf("host %q, rollout %q", refuseTexts[rolloutBookUnreadable], rollout.BookUnreadableRefusal)
	}
}

func mustGet(t *testing.T, reg *registry.Registry, host string) *registry.Route {
	t.Helper()
	rt, err := reg.Get(host)
	if err != nil || rt == nil {
		t.Fatalf("get %s: %v", host, err)
	}
	return rt
}

// PF15: with the flag on but no delegated cgroup subtree, the flag is
// refused before the activator exists. Generations and containment stay
// off, no manager is armed, no hook is set, every leg is today's, and the
// status carries the reason code.
func TestNoDelegationDeclinesTheFlag(t *testing.T) {
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	resetRollout(t)
	forbidPlacementCheck(t)
	eb := &activator.ExecBackend{Control: instancectl.NewClient(filepath.Join(t.TempDir(), "ctl.sock"))}
	if swapArming(overlapEnv(pub), eb) {
		t.Fatal("armed generations with no delegated cgroup subtree")
	}
	if got := rolloutUnarmed(); got != rolloutNoDelegation {
		t.Fatalf("declined with %q, want %q", got, rolloutNoDelegation)
	}
	// runAgent: SetGenerations(swapOn), then New, then armRollout.
	svc, reg, dd := rolloutFixture(t, false)
	declineRollout(rolloutNoDelegation) // rolloutFixture reset it; runAgent would not
	opened := false
	m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil,
		func(context.Context) (release.Store, error) {
			opened = true
			return release.NewDirStore(t.TempDir()), nil
		})
	if m != nil || armedRollout() != nil || opened {
		t.Fatal("armed a scheduler over an activator without generations")
	}
	if svc.Planner != nil || svc.OnGen != nil || svc.OnDrift != nil || svc.CoalesceRecycle != nil {
		t.Fatal("a declined flag set an activator hook")
	}
	if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutNoDelegation {
		t.Fatalf("status = %+v, want the reason code %q", r, rolloutNoDelegation)
	}
	if rolloutUnavailable() != "" {
		t.Fatal("a declined flag refuses managed starts")
	}
	legsAreTodays(t, svc, reg)
}

// The armed half of PF15: a delegated subtree and a control channel that
// speaks the generation verbs arm generations; a channel without them does
// not.
func TestDelegationArmsTheFlag(t *testing.T) {
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	resetRollout(t)
	calls := stubPlacementCheck(t, nil)
	eb := armableBackend(t)
	if !swapArming(overlapEnv(pub), eb) || rolloutUnarmed() != "" {
		t.Fatalf("delegation present: declined with %q", rolloutUnarmed())
	}
	if *calls != 1 {
		t.Fatalf("the armed flag ran the placement check %d times, want once", *calls)
	}
	if swapArming(noEnv, eb) {
		t.Fatal("the flag off armed generations")
	}
	if *calls != 1 {
		t.Fatal("the flag off ran the placement check")
	}
	eb.Control = nil
	if swapArming(overlapEnv(pub), eb) || rolloutUnarmed() != rolloutNoGenControl {
		t.Fatalf("no generation control: armed, or declined with %q", rolloutUnarmed())
	}
	if *calls != 1 {
		t.Fatal("a box that declined on its generation control still ran the placement check")
	}
}

// The second dev-box check (#272): a box armed CARLOS_SWAP=overlap and then
// failed every generation start, because no process could be placed in its
// cgroup under carlos-edge.service's hardening. Arming now does one real
// placement into a scratch cgroup and declines when it fails, with a code
// of its own, a loud operator line, and the code in the box's status.
func TestSwapArmingDeclinesWhenPlacementFails(t *testing.T) {
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	resetRollout(t)
	var buf bytes.Buffer
	log.SetOutput(&buf)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	eb := armableBackend(t)
	var got *cgroup.Manager
	orig := placementCheck
	placementCheck = func(m *cgroup.Manager) error { got = m; return errors.New("probe exited before exec") }
	t.Cleanup(func() { placementCheck = orig })
	if swapArming(overlapEnv(pub), eb) {
		t.Fatal("armed with a failed placement check")
	}
	if got != eb.Cgroups {
		t.Fatal("the placement check was not run against the backend's cgroup manager")
	}
	if rolloutUnarmed() != rolloutPlacementFailed {
		t.Fatalf("declined as %q", rolloutUnarmed())
	}
	if rolloutUnavailable() != "" {
		t.Fatal("a placement decline refuses managed starts")
	}
	if s := buf.String(); !strings.Contains(s, "REFUSING CARLOS_SWAP=overlap (cgroup-placement-failed)") ||
		!strings.Contains(s, "probe exited before exec") {
		t.Fatalf("no operator line naming the placement refusal and its error: %q", s)
	}
	if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutPlacementFailed {
		t.Fatalf("status %+v: want Unavailable %q", r, rolloutPlacementFailed)
	}
}

// A box that declined for a failed placement honours the failure records an
// earlier armed run left, exactly as the other declined codes do
// (honourDeclinedFailures keys on the activator built without generations,
// not on the code), and publishes its own code beside the honoured book.
func TestPlacementDeclineHonoursFailureRecords(t *testing.T) {
	t.Setenv("CARLOS_OPT_DIR", t.TempDir())
	svc, reg, dd := rolloutFixture(t, false)
	acct := release.RouteAccount(managedRoute("x", ""))
	writeFailureBook(t, dd, []rollout.FailureRecord{{
		Account: acct, App: "blog", Build: "v2", Host: "sib.example.com", Cause: "adopt", Mode: "hold", Reason: "health",
	}}, nil)
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	stubPlacementCheck(t, errors.New("the shim exited before placement"))
	if swapArming(overlapEnv(pub), armableBackend(t)) {
		t.Fatal("armed with a failed placement check")
	}
	m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, unusedStore)
	if m != nil || armedRollout() != nil {
		t.Fatal("a declined box armed the scheduler")
	}
	if heldHolds() == nil || svc.Planner == nil {
		t.Fatal("a placement decline did not honour the failure records")
	}
	blocked := managedRoute("a.example.com", "v2")
	blocked.GoodVersion, blocked.GoodDigest = "v1", "sha256:v1"
	if p := svc.Planner(context.Background(), blocked); p.Refuse != refuseNoSwaps {
		t.Fatalf("a wake onto a held build planned %+v, want refused", p)
	}
	if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutPlacementFailed || r.FailureBook != rolloutBookHonoured {
		t.Fatalf("status = %+v, want the decline code %q and the book %q", r, rolloutPlacementFailed, rolloutBookHonoured)
	}
}

// A box with host claims on declines the swap flag (#272): claims-backed
// launches cannot yet carry a swap's recovery choice, execution evidence,
// generations file or good build. The claims check comes before delegation's,
// so a box that has neither says claims.
func TestSwapArmingDeclinesWhenHostClaimsAreOn(t *testing.T) {
	resetRollout(t)
	forbidPlacementCheck(t)
	var buf bytes.Buffer
	log.SetOutput(&buf)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	// No cgroups either: the claims check must come first, so the code
	// says claims, not delegation.
	eb := &activator.ExecBackend{Cfg: activator.Config{HostWrites: hostwrite.Config{Enabled: true}}}
	if swapArming(overlapEnv(""), eb) {
		t.Fatal("armed a box with host claims on")
	}
	if got := rolloutUnarmed(); got != rolloutHostClaims {
		t.Fatalf("decline code %q, want %q", got, rolloutHostClaims)
	}
	if !strings.Contains(buf.String(), "REFUSING CARLOS_SWAP=overlap (host-claims-on)") {
		t.Fatalf("no operator line naming the claims refusal: %q", buf.String())
	}
	r := rolloutStatusProbe()
	if r == nil || r.Unavailable != rolloutHostClaims {
		t.Fatalf("status %+v: want Unavailable %q", r, rolloutHostClaims)
	}
}

func TestSwapArmingIgnoresClaimsWithTheFlagOff(t *testing.T) {
	resetRollout(t)
	forbidPlacementCheck(t)
	eb := &activator.ExecBackend{Cfg: activator.Config{HostWrites: hostwrite.Config{Enabled: true}}}
	if swapArming(noEnv, eb) || rolloutUnarmed() != "" {
		t.Fatal("flag off: a claims box must say nothing")
	}
}

func TestRestartsFeedTheSchedulerWithTheirIdentity(t *testing.T) {
	routes := []registry.Route{
		{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true},
		{Host: "bot.internal", Kind: "sidecar", App: "blog", Account: "bab", Status: "ready"},
		{Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"},
		{Host: "o.example.com", Kind: "instance", App: "other", Account: "bab", Status: "ready", Hibernate: true},
	}
	k := [2]string{"bab", "blog"}
	got := map[string]string{}
	n := feedRestarts(routes, [][2]string{k}, nil, map[[2]string]string{k: "etag-7"}, func(h, id string) bool { got[h] = id; return true })
	if n != 2 || got["a.example.com"] != "gen:etag-7" || got["bot.internal"] != "gen:etag-7" {
		t.Fatalf("fed %d: %v; want the instance and the sidecar, never the named unit or another app", n, got)
	}
	scoped := []scopedRestart{{app: k, requests: []release.RestartRequest{
		{ID: "r1", Scope: release.RestartScope{Kind: release.RestartScopeHost, Value: "a.example.com"}},
		{ID: "r2", Scope: release.RestartScope{Kind: release.RestartScopeHost, Value: "a.example.com"}},
	}}}
	got = map[string]string{}
	if n := feedRestarts(routes, nil, scoped, nil, func(h, id string) bool { got[h] = id; return true }); n != 1 || got["a.example.com"] != "scope:r2" {
		t.Fatalf("fed %d: %v; want the narrowed host, identified by the newest request", n, got)
	}
}

// Carried obligation 4 (T13): OnChange is a non-blocking nudge that
// coalesces, so a burst of transitions is one early status tick.
func TestOnChangeCoalescesIntoOneNudge(t *testing.T) {
	resetRollout(t)
	done := make(chan struct{})
	go func() {
		for range 100 {
			nudgeStatus()
		}
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("nudgeStatus blocked")
	}
	if len(statusNudge) != 1 {
		t.Fatalf("%d nudges queued, want one", len(statusNudge))
	}
}

// Carried obligation 3 (T11): OnWorkerServing returns at once even while
// the registry write is slow, and the newest identity a host was handed is
// the one that lands last.
func TestTheWorkerServingWriteIsCheapAndKeepsTheNewest(t *testing.T) {
	release1 := make(chan struct{})
	var mu sync.Mutex
	var writes []string
	w := newServingWriter(func(host, ver string) error {
		if ver == "v1" {
			<-release1
		}
		mu.Lock()
		writes = append(writes, host+"="+ver)
		mu.Unlock()
		return nil
	})
	w.publish("a", "v1") // the drain blocks inside this write
	start := time.Now()
	w.publish("a", "v2")
	w.publish("b", "v1b")
	w.publish("a", "v3")
	if d := time.Since(start); d > time.Second {
		t.Fatalf("publish waited %s on a slow write", d)
	}
	close(release1)
	deadline := time.Now().Add(5 * time.Second)
	for {
		w.mu.Lock()
		idle := !w.running
		w.mu.Unlock()
		if idle {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the writer never drained")
		}
		time.Sleep(5 * time.Millisecond)
	}
	mu.Lock()
	defer mu.Unlock()
	var lastA string
	for _, x := range writes {
		if x[:2] == "a=" {
			lastA = x
		}
	}
	if lastA != "a=v3" || !slices.Contains(writes, "b=v1b") {
		t.Fatalf("writes %v: want a's newest (v3) last and b's written", writes)
	}
	if slices.Contains(writes, "a=v2") {
		t.Fatalf("writes %v: a stale intermediate version was written", writes)
	}
}

// armAlways's callback reaches the registry, through the writer.
func TestArmAlwaysRecordsTheServingVersion(t *testing.T) {
	t.Parallel()
	svc, reg, dd := rolloutFixture(t, false)
	host := "bot.internal"
	if err := reg.Upsert(registry.Route{Host: host, Kind: "sidecar", App: "blog", Account: "bab", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	armAlways(svc, reg, dd)
	svc.OnWorkerServing(host, release.StagedIdentity{Version: "v4"})
	deadline := time.Now().Add(5 * time.Second)
	for {
		if rt, _ := reg.Get(host); rt != nil && rt.ServingVersion == "v4" {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal("the serving version never reached the registry")
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// outageStore fails every read the way a bucket outage does.
type outageStore struct{ release.Store }

func (outageStore) Get(context.Context, string) ([]byte, string, error) {
	return nil, "", errors.New("503 slow down")
}

// Carried obligation 1 (T3): the policy loader answers an error for any
// store failure, and only a confirmed missing object as the defaults. A
// policy never read is hold (rollout's policyCache).
func TestThePolicyLoaderFailsOnAnyStoreError(t *testing.T) {
	ctx := context.Background()
	if m, err := rolloutPolicy(outageStore{})(ctx, "bab", "blog"); err == nil {
		t.Fatalf("a store failure read as %+v", m)
	}
	s := release.NewDirStore(t.TempDir())
	m, err := rolloutPolicy(s)(ctx, "bab", "blog")
	if err != nil {
		t.Fatalf("a missing config object: %v", err)
	}
	if got := rollout.PolicyFromManifest(m); got != rollout.DefaultPolicy() {
		t.Fatalf("a missing config object read as %+v, want the defaults", got)
	}
	if _, err := s.Put(ctx, appconfig.Key("bab", "blog"), []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	if m, err := rolloutPolicy(s)(ctx, "bab", "blog"); err == nil {
		t.Fatalf("an undecodable config object read as %+v", m)
	}
}

// A refusing box must not tell a member an awake managed instance was
// restarted: the instance still runs its old config and nothing will restart
// it (spec §8.1, an ack only once active). Its ack carries a fixed
// RestartError and Restarted stays false. A sleeper reads the file on its
// next wake, so its ack says delivered on wake (Hibernating, final review F7
// (3)), never Restarted. An exec route the scheduler does not manage is
// restarted as today and acked as restarted (F7 (2)).
func TestRefusingConfigSyncDoesNotAckAnAwakeManagedHostAsRestarted(t *testing.T) {
	ctx := context.Background()
	svc, reg, dd := rolloutFixture(t, true)
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	noStore := func(context.Context) (release.Store, error) {
		return nil, errors.New("no deployment store: secret detail")
	}
	if m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, noStore); m != nil {
		t.Fatal("armed without a store")
	}
	store := release.NewDirStore(t.TempDir())
	privText, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(privText, "test")
	if err != nil {
		t.Fatal(err)
	}
	if err := appconfig.SetEnv(ctx, store, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
		t.Fatal(err)
	}
	for _, h := range []string{"awake.example.com", "asleep.example.com"} {
		rt := execRoute(h, "bab", "eleven")
		rt.Hibernate = true
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	// Always on, so not managed (rollout.Managed): today's restart.
	if err := reg.Upsert(execRoute("plain.example.com", "bab", "eleven")); err != nil {
		t.Fatal(err)
	}
	var restarted []string
	cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{
		awake:   func(h string) bool { return h == "awake.example.com" },
		restart: func(h string) error { restarted = append(restarted, h); return nil },
	})
	if !ok {
		t.Fatal("refusing mode has no config callbacks")
	}
	acks := map[string]appconfig.ConfigAck{}
	cb.Ack = func(a appconfig.ConfigAck) { acks[a.Host] = a }
	if err := appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, t.TempDir(), "", cb); err != nil {
		t.Fatal(err)
	}
	awake, ok := acks["awake.example.com"]
	if !ok {
		t.Fatalf("no ack for the awake host: %+v", acks)
	}
	if awake.Restarted || awake.RestartError != rolloutRestartRefused {
		t.Fatalf("awake ack = %+v, want Restarted false and RestartError %q", awake, rolloutRestartRefused)
	}
	asleep, ok := acks["asleep.example.com"]
	if !ok {
		t.Fatalf("no ack for the sleeping host: %+v", acks)
	}
	if asleep.RestartError != "" || asleep.Restarted || !asleep.Hibernating {
		t.Fatalf("sleeper ack = %+v, want delivered on wake: Hibernating, not restarted, no error", asleep)
	}
	if !slices.Equal(restarted, []string{"plain.example.com"}) {
		t.Fatalf("today's restart ran for %v, want only the unmanaged route", restarted)
	}
	if plain := acks["plain.example.com"]; !plain.Restarted || plain.RestartError != "" || plain.Hibernating {
		t.Fatalf("unmanaged ack = %+v, want today's: restarted", plain)
	}
}

// Final review F7 (1) and (2), armed: config sync reports a managed native
// sidecar's delivery to the scheduler, which owns its restart, and restarts
// an exec route the scheduler does not manage as today (svc.Bounce, through
// configLegs.restart), acking it as restarted at once; neither managed route
// is restarted by config sync, and the unmanaged one is not reported.
func TestArmedConfigSyncSplitsManagedFromUnmanaged(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	svc, reg, dd := rolloutFixture(t, true)
	svc.GenerationsFile = release.GenerationsPath(dd)
	_, pub, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	managed := execRoute("m.example.com", "bab", "eleven")
	managed.Hibernate, managed.Version = true, "v1"
	bot := registry.Route{Host: "bot.eleven.bab.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bab", Status: "ready", Version: "v1"}
	plain := execRoute("plain.example.com", "bab", "eleven") // always on: not managed
	for _, rt := range []registry.Route{managed, bot, plain} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	if !rollout.Managed(managed) || !rollout.Managed(bot) || rollout.Managed(plain) {
		t.Fatal("fixture: the routes are not managed as this test assumes")
	}
	store := release.NewDirStore(t.TempDir())
	m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, func(context.Context) (release.Store, error) { return store, nil })
	if m == nil {
		t.Fatal("not armed")
	}
	privText, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(privText, "test")
	if err != nil {
		t.Fatal(err)
	}
	if err := appconfig.SetEnv(ctx, store, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
		t.Fatal(err)
	}
	var restarted, delivered []string
	cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{restart: func(h string) error { restarted = append(restarted, h); return nil }})
	if !ok || cb.Delivered == nil {
		t.Fatal("armed mode has no delivery callback")
	}
	report := cb.Delivered
	cb.Delivered = func(h, rev string) { delivered = append(delivered, h); report(h, rev) }
	acks := map[string]appconfig.ConfigAck{}
	cb.Ack = m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks[a.Host] = a })
	if err := appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, t.TempDir(), "", cb); err != nil {
		t.Fatal(err)
	}
	// The manager's writer writes the acks (Task 11); Close writes what is
	// still pending before it returns, so the map is complete after it.
	m.Close(ctx)
	slices.Sort(delivered)
	if want := []string{bot.Host, managed.Host}; !slices.Equal(delivered, want) {
		t.Fatalf("delivered %v, want %v: the managed sidecar's revision must reach the scheduler, the unmanaged route's must not", delivered, want)
	}
	if !slices.Equal(restarted, []string{plain.Host}) {
		t.Fatalf("today's restart ran for %v, want only the unmanaged route", restarted)
	}
	if a, ok := acks[plain.Host]; !ok || !a.Restarted || a.RestartError != "" {
		t.Fatalf("unmanaged ack = %+v (written %v), want today's: restarted, at once", a, ok)
	}
}

// The awake test is the activator's: a managed host it owns. Without an
// activator, or for a host it does not own, nothing is awake.
func TestManagedAwakeNeedsTheActivatorToOwnAManagedHost(t *testing.T) {
	svc, reg, _ := rolloutFixture(t, true)
	rt := execRoute("m.example.com", "bab", "eleven")
	rt.Hibernate = true
	if err := reg.Upsert(rt); err != nil {
		t.Fatal(err)
	}
	if managedAwake(nil, reg)("m.example.com") {
		t.Error("awake with no activator")
	}
	if managedAwake(svc, reg)("m.example.com") || managedAwake(svc, reg)("gone.example.com") {
		t.Error("awake though the activator owns nothing")
	}
}