package host

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/telemetry"
)

// ciCountingStore is a DirStore with the change index on that records every
// read by method and key. Embedding the concrete store forwards every
// optional extension (ConditionalGetter and ETagLister included), so the
// view and the watch see the store production gives them.
type ciCountingStore struct {
	*release.DirStore
	mu       sync.Mutex
	calls    []string
	failList string // List and ListETags of exactly this prefix fail
	// stall, when set, holds every GetIfChanged of the change index until
	// the caller's ctx ends or stall is closed: a hung index GET while the
	// rest of the bucket answers.
	stall chan struct{}
}

// stallIndex hangs the index GET until the returned func is called.
func (s *ciCountingStore) stallIndex() (release func()) {
	ch := make(chan struct{})
	s.mu.Lock()
	s.stall = ch
	s.mu.Unlock()
	return func() {
		s.mu.Lock()
		s.stall = nil
		s.mu.Unlock()
		close(ch)
	}
}

func (s *ciCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() }

func (s *ciCountingStore) listFails(prefix string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.failList != "" && s.failList == prefix
}

func newCICountingStore(t *testing.T) *ciCountingStore {
	t.Helper()
	d := release.NewDirStore(t.TempDir())
	d.EnableChangeIndex()
	return &ciCountingStore{DirStore: d}
}

func (s *ciCountingStore) note(method, key string) {
	s.mu.Lock()
	s.calls = append(s.calls, method+" "+key)
	s.mu.Unlock()
}

func (s *ciCountingStore) reset() { s.mu.Lock(); s.calls = nil; s.mu.Unlock() }

func (s *ciCountingStore) got() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := slices.Clone(s.calls)
	sort.Strings(out)
	return out
}

// count is how many recorded calls start with method+" " and name a key
// under prefix.
func (s *ciCountingStore) count(method, prefix string) int {
	n := 0
	for _, c := range s.got() {
		if strings.HasPrefix(c, method+" "+prefix) {
			n++
		}
	}
	return n
}

func (s *ciCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.note("Get", key)
	return s.DirStore.Get(ctx, key)
}

func (s *ciCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	s.note("GetIfChanged", key)
	s.mu.Lock()
	stall := s.stall
	s.mu.Unlock()
	if stall != nil && key == release.ChangeIndexKey {
		select {
		case <-ctx.Done():
			return nil, "", false, ctx.Err()
		case <-stall:
		}
	}
	return s.DirStore.GetIfChanged(ctx, key, etag)
}

func (s *ciCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.note("List", prefix)
	if s.listFails(prefix) {
		return nil, errors.New("list refused")
	}
	return s.DirStore.List(ctx, prefix)
}

func (s *ciCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	s.note("ListETags", prefix)
	if s.listFails(prefix) {
		return nil, errors.New("list refused")
	}
	return s.DirStore.ListETags(ctx, prefix)
}

func (s *ciCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	s.note("ListDirs", prefix)
	return s.DirStore.ListDirs(ctx, prefix)
}

func (s *ciCountingStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	s.note("ListMod", prefix)
	return s.DirStore.ListMod(ctx, prefix)
}

// ciWorld is a deployment of one account with two apps, shop and cafe, each
// with one instance record and a config bundle, written through the indexed
// store.
type ciWorld struct {
	live *ciCountingStore
	acct string
	now  time.Time
}

func newCIWorld(t *testing.T) *ciWorld {
	t.Helper()
	t.Setenv("CARLOS_HOST_CLAIM", "") // the appconfig and provision guards read the process env
	ctx := context.Background()
	w := &ciWorld{live: newCICountingStore(t), acct: domains.Sqid(2), now: time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)}
	manifest, err := json.Marshal(map[string]any{"id": 2, "sqid": w.acct, "name": "fixture", "created_at": w.now})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.live.Put(ctx, "accounts/"+w.acct+"/account.json", manifest, ""); err != nil {
		t.Fatal(err)
	}
	for _, app := range []string{"shop", "cafe"} {
		if err := provision.PutManifest(ctx, w.live, w.acct, app, provision.Manifest{Schema: 1, App: app, Domain: "example.com"}); err != nil {
			t.Fatal(err)
		}
		w.record(t, app, app+".example.com")
		if err := appconfig.SetEnv(ctx, w.live, w.acct, app, map[string]string{"APP": app}); err != nil {
			t.Fatal(err)
		}
	}
	w.live.reset()
	return w
}

func (w *ciWorld) record(t *testing.T, app, host string) {
	t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0], Account: w.acct, App: app, Channel: "stable", CreatedAt: w.now}
	if err := provision.CreateRecord(context.Background(), w.live, rec); err != nil {
		t.Fatal(err)
	}
}

func (w *ciWorld) scope(app string) string { return "apps/" + w.acct + "/" + app }

// armView arms the change index over live for one test, as RunAgent does,
// and disarms it afterwards. Tests that call it stay serial.
func armView(t *testing.T, live release.Store) *provision.RecordView {
	t.Helper()
	watch := release.NewChangeWatch(live)
	v := provision.NewRecordView(live, watch)
	armedChangeWatch.Store(watch)
	armedRecordView.Store(v)
	t.Cleanup(func() {
		armedChangeWatch.Store(nil)
		armedRecordView.Store(nil)
	})
	return v
}

// disarmView makes sure no earlier test left the view armed.
func disarmView(t *testing.T) {
	t.Helper()
	armedChangeWatch.Store(nil)
	armedRecordView.Store(nil)
}

func ciReconcileConfig(t *testing.T) provision.ReconcileConfig {
	return provision.ReconcileConfig{DataDir: t.TempDir(), Layout: flatLayout}
}

// recordFamilyLists counts listings under the record families.
func recordFamilyLists(s *ciCountingStore) int {
	n := 0
	for _, m := range []string{"List", "ListETags", "ListDirs"} {
		n += s.count(m, "apps/") + s.count(m, "accounts/")
	}
	return n
}

func TestChangeIndexArmedOnlyForRead(t *testing.T) {
	t.Parallel()
	for v, want := range map[string]bool{"": false, "read": true, "READ": false, "write": false, "1": false, " read": false} {
		if got := changeIndexArmed(func(k string) string {
			if k == "CARLOS_CHANGE_INDEX" {
				return v
			}
			return ""
		}); got != want {
			t.Errorf("CARLOS_CHANGE_INDEX=%q armed = %v, want %v", v, got, want)
		}
	}
}

// A fleet box stays disarmed with the variable set: its store door has no
// conditional read, so every Refresh would walk anyway (spec §3.4).
func TestChangeIndexStaysDisarmedOnFleetBox(t *testing.T) {
	disarmView(t)
	t.Cleanup(func() { disarmView(t) })
	opened := 0
	open := func(context.Context) (release.Store, error) { opened++; return newCICountingStore(t), nil }
	env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } }

	if v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read", "CARLOS_FLEET": "pi"}), open); v != nil || armedRecordView.Load() != nil || armedChangeWatch.Load() != nil {
		t.Fatal("a fleet box armed the change index")
	}
	if v := armChangeIndex(context.Background(), env(nil), open); v != nil || armedRecordView.Load() != nil {
		t.Fatal("an unset CARLOS_CHANGE_INDEX armed the change index")
	}
	if opened != 0 {
		t.Fatalf("a disarmed box opened the deployment store %d time(s)", opened)
	}
	v := armChangeIndex(context.Background(), env(map[string]string{"CARLOS_CHANGE_INDEX": "read"}), open)
	if v == nil || armedRecordView.Load() != v || armedChangeWatch.Load() == nil {
		t.Fatal("a platform box with CARLOS_CHANGE_INDEX=read did not arm")
	}
}

// Armed, the record walks of a reconcile pass (the offline keep view's
// listing, Reconcile and ReconcileSidecars) read the view: once it is built,
// two passes with no writes list nothing on the bucket.
func TestReconcileUsesViewWhenArmed(t *testing.T) {
	w := newCIWorld(t)
	v := armView(t, w.live)
	if err := v.Rebuild(context.Background()); err != nil {
		t.Fatal(err)
	}
	reg := restartTestRegistry(t)
	// A capable worker whose keep view has read nothing yet: the first pass
	// lists records for it (refreshOfflineView), from the bucket, never the
	// view (final review R1, finding 1: everything a move acts on is live).
	mw := newMovesWorker(moveCfg{OfflineCapable: true})
	mw.cfg.Offline = newOfflineView(mw.cfg, nil)
	cfg := ciReconcileConfig(t)
	w.live.reset()
	if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, mw); err != nil {
		t.Fatal(err)
	}
	if !mw.cfg.Offline.built() || mw.cfg.Offline.failed() {
		t.Fatal("the offline keep view was not read")
	}
	if n := w.live.count("List", "accounts/"); n != 1 {
		t.Fatalf("the unbuilt offline keep view listed accounts/ %d time(s), want once, from the bucket: %v", n, w.live.got())
	}
	w.live.reset()
	for range 2 {
		if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, mw); err != nil {
			t.Fatal(err)
		}
	}
	if n := w.live.count("List", ""); n != 0 {
		t.Fatalf("two armed passes with no writes listed the bucket %d time(s): %v", n, w.live.got())
	}
	if n := recordFamilyLists(w.live); n != 0 {
		t.Fatalf("two armed passes listed the record families %d time(s): %v", n, w.live.got())
	}
	for _, h := range []string{"shop.example.com", "cafe.example.com"} {
		if rt, err := reg.Get(h); err != nil || rt == nil {
			t.Fatalf("no route for %s from the view's records: %v", h, err)
		}
	}
}

// Disarmed, the same passes walk the bucket exactly as before the index.
func TestReconcileWalksWhenDisarmed(t *testing.T) {
	disarmView(t)
	w := newCIWorld(t)
	reg := restartTestRegistry(t)
	cfg := ciReconcileConfig(t)
	for range 2 {
		if err := reconcileSerializedWith(context.Background(), w.live, reg, cfg, nil); err != nil {
			t.Fatal(err)
		}
	}
	if n := w.live.count("List", "accounts/"); n < 2 {
		t.Fatalf("two disarmed passes listed accounts/ %d time(s), want one walk each: %v", n, w.live.got())
	}
	if n := w.live.count("GetIfChanged", ""); n != 0 {
		t.Fatalf("a disarmed pass read the change index: %v", w.live.got())
	}
}

// instancesLoop's floor is reconcileSerialized over its own store: armed,
// with nothing changed, it costs one conditional GET of the index and no
// listing.
func TestInstancesLoopFloorIsRefreshWhenArmed(t *testing.T) {
	w := newCIWorld(t)
	v := armView(t, w.live)
	if err := v.Rebuild(context.Background()); err != nil {
		t.Fatal(err)
	}
	reg := restartTestRegistry(t)
	cfg := ciReconcileConfig(t)
	if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	w.live.reset()
	if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if n := w.live.count("GetIfChanged", release.ChangeIndexKey); n != 1 {
		t.Fatalf("a quiet floor read the index %d time(s), want 1: %v", n, w.live.got())
	}
	if n := w.live.count("List", "") + w.live.count("ListETags", "") + w.live.count("ListDirs", ""); n != 0 {
		t.Fatalf("a quiet floor listed the bucket: %v", w.live.got())
	}
}

// A view that cannot vouch (its Refresh failed, so there is no Complete
// snapshot) is never used: the caller walks live, as before the index.
func TestViewStoreFallsBackWhenTheViewCannotVouch(t *testing.T) {
	disarmView(t)
	w := newCIWorld(t)
	if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) {
		t.Fatal("a disarmed viewStore did not hand back live")
	}
	armView(t, w.live)
	w.live.setFailList("accounts/")
	if s, _, ok := viewStore(context.Background(), w.live); ok || s != release.Store(w.live) {
		t.Fatal("viewStore used the view after a failed Refresh")
	}
	w.live.setFailList("")
	s, epoch, ok := viewStore(context.Background(), w.live)
	if !ok || s == release.Store(w.live) || epoch == 0 {
		t.Fatalf("viewStore after a good Refresh: ok %v epoch %d", ok, epoch)
	}
}

// A hung index GET while the rest of the bucket answers (amadan #336 final
// review, finding 3): the reconcile pass gives up on the index after
// ChangeIndexPollTimeout, walks, and mints what the walk found; the watch
// is unknown meanwhile, so a gated read marked before the stall is due
// again; and the next good poll brings the watch back.
func TestStalledIndexGetKeepsReconcileWalking(t *testing.T) {
	w := newCIWorld(t)
	v := armView(t, w.live)
	if err := v.Rebuild(context.Background()); err != nil {
		t.Fatal(err)
	}
	reg := restartTestRegistry(t)
	cfg := ciReconcileConfig(t)
	if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	gate := newChangeGate()
	clock := newGateClock()
	gen, _ := gate.Due(w.scope("shop"), clock.now())
	gate.Mark(w.scope("shop"), gen, clock.now(), true)
	if _, due := gate.Due(w.scope("shop"), clock.now()); due {
		t.Fatal("fixture: a marked scope is due before the stall")
	}

	w.record(t, "shop", "two.shop.example.com")
	unstall := w.live.stallIndex()
	defer func() {
		if unstall != nil {
			unstall()
		}
	}()
	done := make(chan error, 1)
	go func() { done <- reconcileSerialized(context.Background(), w.live, reg, cfg) }()
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(4 * release.ChangeIndexPollTimeout):
		t.Fatal("a reconcile pass hung behind a stalled index GET")
	}
	if rt, err := reg.Get("two.shop.example.com"); err != nil || rt == nil {
		t.Fatalf("the pass over a stalled index did not walk to the new record: %v", err)
	}
	if armedChangeWatch.Load().Usable() {
		t.Fatal("the watch vouches after its poll gave up")
	}
	if _, due := gate.Due(w.scope("shop"), clock.now()); !due {
		t.Fatal("a gated read trusted the frozen observation")
	}

	unstall()
	unstall = nil
	if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !armedChangeWatch.Load().Usable() {
		t.Fatal("the watch did not recover on the next good poll")
	}
}

// ---- moves -----------------------------------------------------------------

// The moves pass decides whether it has work from its own refresh of the
// view inside w.pass, never from the snapshot the reconcile leg of its pass
// set read (spec §4.3, codex 1): pass A is paused after its reconcile, a
// move is requested, and A's moves leg sees it and walks the bucket.
func TestMovesPassDecidesFromItsOwnRefresh(t *testing.T) {
	w := newCIWorld(t)
	armView(t, w.live)
	mw := newMovesWorker(moveCfg{})
	armedMovesWorker.Store(mw)
	t.Cleanup(func() { armedMovesWorker.Store(nil) })
	reg := restartTestRegistry(t)
	rcfg := ciReconcileConfig(t)
	legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), dd: t.TempDir(),
		cfg: func() provision.ReconcileConfig { return rcfg }}
	legs.full(context.Background()) // walks once: nothing to move
	if !mw.walkedIdle {
		t.Fatal("fixture: a walk with no move left the worker busy")
	}

	paused, resume := make(chan struct{}), make(chan struct{})
	var first atomic.Bool
	passAfterReconcileHook = func() {
		if first.CompareAndSwap(false, true) {
			paused <- struct{}{}
			<-resume
		}
	}
	t.Cleanup(func() { passAfterReconcileHook = nil })
	done := make(chan struct{})
	go func() { defer close(done); legs.full(context.Background()) }() // pass A
	<-paused
	if _, err := provision.SetRecordMove(context.Background(), w.live, w.acct, "shop", "shop.example.com", provision.MoveRequest("uk", w.now)); err != nil {
		t.Fatal(err)
	}
	w.live.reset()
	close(resume)
	<-done
	if n := w.live.count("List", "accounts/"); n == 0 {
		t.Fatalf("pass A's moves leg did not walk for a move requested after its reconcile: %v", w.live.got())
	}
	if mw.walkedIdle {
		t.Fatal("a walk that found a move left the worker idle")
	}
}

// With nothing to move, the armed moves pass costs no walk: the record view
// vouches that no record carries a move, and the worker holds no move state.
// Anything the worker holds (here, a cut's wake fence) makes it walk.
func TestMovesPassSkipsTheWalkOnlyWithNoMoveWork(t *testing.T) {
	w := newCIWorld(t)
	armView(t, w.live)
	mw := newMovesWorker(moveCfg{})
	if err := movesPass(context.Background(), w.live, mw); err != nil {
		t.Fatal(err)
	}
	w.live.reset()
	if err := movesPass(context.Background(), w.live, mw); err != nil {
		t.Fatal(err)
	}
	if n := recordFamilyLists(w.live); n != 0 {
		t.Fatalf("an idle armed moves pass listed the record families: %v", w.live.got())
	}
	mw.fenced.Store("shop.example.com", uint64(1))
	w.live.reset()
	if err := movesPass(context.Background(), w.live, mw); err != nil {
		t.Fatal(err)
	}
	if n := w.live.count("List", "accounts/"); n == 0 {
		t.Fatalf("a worker holding a wake fence skipped its walk: %v", w.live.got())
	}
}

// Final review R1 finding 1 (amadan #336): the moves pass acts on the
// bucket, never on a snapshot older than what a pass already saw there.
// An abort's completion misses its bump, so the view still holds the move
// aborting; an unrelated scope then fails to re-read and a pass walks the
// bucket instead (the live fallback); a new move of the same host is
// requested (its bump missed too) and staging fills; the unrelated scope
// recovers, so the view is Complete again with the old aborting state.
// Acting on that snapshot would delete the new move's staging. The pass
// walks the bucket, sees the new move preparing, and leaves staging alone.
func TestMovesPassNeverActsOnAnOlderSnapshot(t *testing.T) {
	h := newMoveHarness(t)
	root := t.TempDir()
	indexed := release.NewDirStore(root)
	indexed.EnableChangeIndex()
	live := &ciCountingStore{DirStore: indexed}
	bypass := release.NewDirStore(root) // writes whose bump failed
	ctx := context.Background()
	edges, _, err := h.s.Get(ctx, "control/edges/uk-box.json")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := live.Put(ctx, "control/edges/uk-box.json", edges, ""); err != nil {
		t.Fatal(err)
	}
	h.s = live
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort())
	other := "b.go.example.com"
	if err := provision.CreateRecord(ctx, live, provision.Record{Schema: 1, Host: other, Slug: "b", App: "other", Account: "bab",
		Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil {
		t.Fatal(err)
	}
	v := armView(t, live)
	if err := v.Rebuild(ctx); err != nil {
		t.Fatal(err)
	}
	if recs, err := provision.ListAllRecords(ctx, v.Snapshot().Store(live)); err != nil || len(recs) != 2 {
		t.Fatalf("fixture: the view holds %d record(s), %v; want both", len(recs), err)
	}
	// The abort completes, and its bump fails.
	if _, err := provision.SetRecordMove(ctx, bypass, "bab", "app", host, provision.MoveAbortComplete()); err != nil {
		t.Fatal(err)
	}
	// An unrelated scope moves and will not re-read: the view cannot vouch,
	// and the pass walks the bucket, where the abort is complete.
	third := provision.Record{Schema: 1, Host: "c.go.example.com", Slug: "c", App: "other", Account: "bab",
		Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
	if err := provision.CreateRecord(ctx, live, third); err != nil {
		t.Fatal(err)
	}
	live.setFailList(provision.RecordsPrefix("bab", "other"))
	if _, _, ok := viewStore(ctx, live); ok {
		t.Fatal("fixture: the view vouched with a scope failing to re-read")
	}
	wk := h.worker()
	if err := movesPass(ctx, live, wk); err != nil {
		t.Fatal(err)
	}
	// The host is moved again, the bump failing again, and staging fills.
	if _, err := provision.SetRecordMove(ctx, bypass, "bab", "app", host, provision.MoveRequest("uk", time.Now())); err != nil {
		t.Fatal(err)
	}
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1")
	// The unrelated scope recovers: Complete again, with the old abort.
	live.setFailList("")
	if _, _, ok := viewStore(ctx, live); !ok {
		t.Fatal("fixture: the view did not recover")
	}
	snapRec, err := provision.GetRecord(ctx, v.Snapshot().Store(live), "bab", "app", host)
	if err != nil || snapRec.Move == nil || snapRec.Move.State != provision.MoveAborting {
		t.Fatalf("fixture: the recovered snapshot should hold the old abort: %+v %v", snapRec, err)
	}

	if err := movesPass(ctx, live, wk); err != nil {
		t.Fatal(err)
	}
	for _, op := range h.f.ops {
		if strings.HasPrefix(op, "delete data-uk/moves/"+host+"/") {
			t.Fatalf("the pass acted on the snapshot's old abort and deleted the new move's staging: %v", h.f.ops)
		}
	}
	if staged := h.f.bodies("data-uk", "moves/"+host+"/"); len(staged) != 1 {
		t.Fatalf("the new move's staging after the pass: %v", staged)
	}
	if got := h.read(t, host); got.Move == nil || got.Move.State == provision.MoveAborting {
		t.Fatalf("the record after the pass: %+v", got.Move)
	}
}

// ---- fastTick --------------------------------------------------------------

// ciFastDeps is a fastDeps over an indexed store with counting hooks for
// both pass shapes.
func ciFastDeps(t *testing.T) (*fastDeps, *release.DirStore, *[]string, *[][]string) {
	t.Helper()
	env := newFastFixture(t)
	ds := release.NewDirStore(t.TempDir())
	ds.EnableChangeIndex()
	env.store = ds
	var full []string
	var scoped [][]string
	d := env.deps(func() { full = append(full, "full") })
	d.watch = release.NewChangeWatch(ds)
	d.scopedPass = func(_ context.Context, moved []string) { scoped = append(scoped, moved) }
	return d, ds, &full, &scoped
}

func TestFastTickIndexOnlyRunsScopedPass(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	d, ds, full, scoped := ciFastDeps(t)
	now := time.Unix(1754640000, 0)
	fastTick(ctx, d, now) // seeds the serial and the index: no pass
	if len(*full)+len(*scoped) != 0 {
		t.Fatalf("the seeding tick ran a pass: full %v scoped %v", *full, *scoped)
	}
	if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	fastTick(ctx, d, now.Add(10*time.Second))
	if len(*full) != 0 {
		t.Fatalf("an index-only change ran the full pass: %v", *full)
	}
	if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) {
		t.Fatalf("scoped passes %v, want %v", *scoped, want)
	}
	// Inside the clamp the change waits, and is acted on at the next tick
	// past it.
	if _, err := ds.Put(ctx, "apps/bab/cafe/instances/one.cafe.example.com.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	fastTick(ctx, d, now.Add(12*time.Second))
	if len(*scoped) != 1 {
		t.Fatalf("a change inside the clamp ran at once: %v", *scoped)
	}
	fastTick(ctx, d, now.Add(16*time.Second))
	if want := [][]string{{"apps/bab/shop"}, {"apps/bab/cafe"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) {
		t.Fatalf("scoped passes %v, want %v", *scoped, want)
	}
	fastTick(ctx, d, now.Add(30*time.Second))
	if len(*scoped) != 2 {
		t.Fatalf("an unchanged index ran the scoped pass again: %v", *scoped)
	}
	// Scopes outside the record view's (control/edges) run no pass.
	if _, err := ds.Put(ctx, "control/edges/one.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	fastTick(ctx, d, now.Add(40*time.Second))
	if len(*scoped) != 2 || len(*full) != 0 {
		t.Fatalf("a control/ scope ran a pass: full %v scoped %v", *full, *scoped)
	}
}

// A shared watch: another poller (the record view) taking Poll's moved list
// first does not hide the movement from fastTick.
func TestFastTickActsOnObservedNotPollReturn(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	d, ds, _, scoped := ciFastDeps(t)
	now := time.Unix(1754640000, 0)
	fastTick(ctx, d, now)
	if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	if moved, err := d.watch.Poll(ctx); err != nil || len(moved) == 0 {
		t.Fatalf("the sibling poll saw %v, %v", moved, err)
	}
	fastTick(ctx, d, now.Add(10*time.Second))
	if want := [][]string{{"apps/bab/shop"}}; !slices.EqualFunc(*scoped, want, slices.Equal[[]string]) {
		t.Fatalf("scoped passes %v, want %v: a sibling's Poll swallowed the movement", *scoped, want)
	}
}

func TestSerialChangeStillRunsFullPass(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	d, ds, full, scoped := ciFastDeps(t)
	now := time.Unix(1754640000, 0)
	fastTick(ctx, d, now)
	if _, err := ds.Put(ctx, "apps/bab/shop/instances/one.shop.example.com.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	if err := release.BumpSerial(ctx, ds, now); err != nil {
		t.Fatal(err)
	}
	fastTick(ctx, d, now.Add(10*time.Second))
	if len(*full) != 1 || len(*scoped) != 0 {
		t.Fatalf("a serial change with the index moved: full %v scoped %v, want one full pass only", *full, *scoped)
	}
	// The full pass covered the movement: the next tick runs nothing.
	fastTick(ctx, d, now.Add(20*time.Second))
	if len(*full) != 1 || len(*scoped) != 0 {
		t.Fatalf("the movement the full pass covered ran again: full %v scoped %v", *full, *scoped)
	}
}

// The backstop asks for the unrestricted full pass after its rebuild; the
// loop's own tick runs it, once.
func TestFastTickRunsTheBackstopsFullPass(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	d, ds, full, scoped := ciFastDeps(t)
	var due atomic.Bool
	d.fullDue = &due
	now := time.Unix(1754640000, 0)
	fastTick(ctx, d, now)
	v := provision.NewRecordView(ds, d.watch)
	recordViewBackstop(ctx, v, &due)
	if v.Snapshot() == nil {
		t.Fatal("the backstop did not rebuild the view")
	}
	fastTick(ctx, d, now.Add(10*time.Second))
	if len(*full) != 1 || len(*scoped) != 0 || due.Load() {
		t.Fatalf("after the backstop: full %v scoped %v due %v, want one full pass", *full, *scoped, due.Load())
	}
	fastTick(ctx, d, now.Add(20*time.Second))
	if len(*full) != 1 {
		t.Fatalf("the backstop's full pass ran again: %v", *full)
	}
}

func TestBackstopEveryIsJitteredAroundTheBackstop(t *testing.T) {
	t.Parallel()
	lo, hi := release.ChangeIndexBackstop*9/10, release.ChangeIndexBackstop*11/10
	for range 200 {
		if d := backstopEvery(); d < lo || d >= hi {
			t.Fatalf("backstopEvery() = %v, want in [%v, %v)", d, lo, hi)
		}
	}
}

// ---- the scoped pass -------------------------------------------------------

// Codex 10's witness: on a box hosting shop and cafe, an index bump for shop
// runs a scoped pass that reads only shop's keys (its app prefix and its
// build state), plus the change index, while shop's new config is
// delivered. Nothing under cafe's prefix, nothing for cafe's build state,
// and nothing box-wide (edges files, domain claims) is read.
func TestScopedPassTouchesOnlyMovedApp(t *testing.T) {
	w := newCIWorld(t)
	armView(t, w.live)
	armedMovesWorker.Store(newMovesWorker(moveCfg{}))
	t.Cleanup(func() { armedMovesWorker.Store(nil) })
	// A sink, so the telemetry leg runs (telemetryPass returns early without
	// one): its reads are gated too, so the scoped pass reads only shop's.
	var pushed []map[string]bool
	setAnalyticsSink(func(on map[string]bool) { pushed = append(pushed, on) })
	t.Cleanup(func() { analyticsSink.Store(nil) })
	raw, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	pk, err := appconfig.ParsePrivateKey(raw, "test")
	if err != nil {
		t.Fatal(err)
	}
	reg := restartTestRegistry(t)
	dd := t.TempDir()
	// Build states live in the same counting store, under inst/<host>/.
	bs := &buildStates{hibBucket: "hib", prefix: func(h string) string { return "inst/" + h },
		storeFor: storesOf(map[string]release.Store{"hib": w.live}), reg: reg, optDir: t.TempDir(), bakeDir: t.TempDir(), now: time.Now}
	legs := &passLegs{store: w.live, reg: reg, optDir: t.TempDir(), priv: pk, dd: dd, bs: bs,
		cfg: func() provision.ReconcileConfig { return provision.ReconcileConfig{DataDir: dd, Layout: flatLayout} }}
	w.live.reset()
	legs.full(context.Background())
	// The fixture exercises every box-wide leg on the full pass.
	for _, want := range []string{"inst/cafe.example.com/", "domains/", "control/edges",
		telemetry.AccountKey(w.acct), telemetry.AppKey(w.acct, "cafe")} {
		if !slices.ContainsFunc(w.live.got(), func(c string) bool { _, k, _ := strings.Cut(c, " "); return strings.HasPrefix(k, want) }) {
			t.Fatalf("fixture: the full pass read nothing under %s: %v", want, w.live.got())
		}
	}
	for _, h := range []string{"shop.example.com", "cafe.example.com"} {
		rt, err := reg.Get(h)
		if err != nil || rt == nil {
			t.Fatalf("fixture: no route for %s: %v", h, err)
		}
		if !strings.Contains(readConfigEnv(t, dd, *rt), "APP=") {
			t.Fatalf("fixture: %s got no config from the full pass", h)
		}
	}

	if err := appconfig.SetEnv(context.Background(), w.live, w.acct, "shop", map[string]string{"APP": "shop2"}); err != nil {
		t.Fatal(err)
	}
	w.live.reset()
	legs.scoped(context.Background(), []string{w.scope("shop")})

	if len(pushed) != 2 || len(pushed[1]) != 0 {
		t.Fatalf("switch maps pushed %v, want one per pass, both all off", pushed)
	}
	shop := "apps/" + w.acct + "/shop/"
	var shopReads int
	for _, c := range w.live.got() {
		_, key, _ := strings.Cut(c, " ")
		switch {
		case strings.HasPrefix(key, shop):
			shopReads++
		case key == release.ChangeIndexKey, strings.HasPrefix(key, "inst/shop.example.com/"):
		case strings.HasPrefix(key, "control/config-ack/") && strings.HasSuffix(key, "/shop.example.com.json"):
			// shop's config acknowledgement, written for the delivery
		default:
			t.Errorf("the scoped pass for shop read %s, which is not shop's", c)
		}
	}
	if shopReads == 0 {
		t.Fatalf("the scoped pass read nothing under shop: %v", w.live.got())
	}
	rt, _ := reg.Get("shop.example.com")
	if got := readConfigEnv(t, dd, *rt); !strings.Contains(got, "APP=shop2") {
		t.Fatalf("shop's config after the scoped pass = %q, want APP=shop2", got)
	}
}

func readConfigEnv(t *testing.T, dd string, rt registry.Route) string {
	t.Helper()
	b, err := os.ReadFile(appconfig.ConfigEnvPath(dd, rt))
	if err != nil && !os.IsNotExist(err) {
		t.Fatal(err)
	}
	return string(b)
}