package provision

import (
	"context"
	"errors"
	"fmt"
	"os"
	"reflect"
	"slices"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/release"
)

// viewCountingStore is a DirStore that counts every read by method and key,
// and can fail every listing of one prefix. Embedding the concrete store
// forwards every optional extension; bucket decides whether it promises
// bucket-shaped ListDirs (the S3 store does, DirStore does not).
type viewCountingStore struct {
	*release.DirStore
	bucket bool

	mu       sync.Mutex
	calls    []string
	failList string // List, ListETags and ListDirs of exactly this prefix fail
	failGet  string // Get and GetIfChanged of exactly this key fail
	ghost    string // ListETags lists this key, which is not there: deleted between list and read
	cancelOn string // ListETags of exactly this prefix calls cancel first
	cancel   context.CancelFunc
	stall    string // GetIfChanged of exactly this key blocks until the caller's ctx ends
}

func (s *viewCountingStore) setStall(k string) { s.mu.Lock(); s.stall = k; s.mu.Unlock() }

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

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

func (s *viewCountingStore) getFails(key string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.failGet != "" && key == s.failGet
}

func (s *viewCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() }
func (s *viewCountingStore) setFailGet(k string)  { s.mu.Lock(); s.failGet = k; s.mu.Unlock() }
func (s *viewCountingStore) setGhost(k string)    { s.mu.Lock(); s.ghost = k; s.mu.Unlock() }

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

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

func (s *viewCountingStore) BucketShapedListDirs() bool { return s.bucket }

func (s *viewCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.note("Get", key)
	if s.getFails(key) {
		return nil, "", errors.New("get refused")
	}
	return s.DirStore.Get(ctx, key)
}

func (s *viewCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	s.note("GetIfChanged", key)
	s.mu.Lock()
	stall := s.stall != "" && key == s.stall
	s.mu.Unlock()
	if stall {
		<-ctx.Done()
		return nil, "", false, ctx.Err()
	}
	if s.getFails(key) {
		return nil, "", false, errors.New("get refused")
	}
	return s.DirStore.GetIfChanged(ctx, key, etag)
}

func (s *viewCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.note("List", prefix)
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if s.listFails(prefix) {
		return nil, errors.New("list refused")
	}
	return s.DirStore.List(ctx, prefix)
}

func (s *viewCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	s.note("ListETags", prefix)
	s.mu.Lock()
	if s.cancelOn != "" && prefix == s.cancelOn {
		s.cancel()
	}
	s.mu.Unlock()
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if s.listFails(prefix) {
		return nil, errors.New("list refused")
	}
	kes, err := s.DirStore.ListETags(ctx, prefix)
	s.mu.Lock()
	ghost := s.ghost
	s.mu.Unlock()
	if err == nil && ghost != "" && strings.HasPrefix(ghost, prefix) {
		kes = append(kes, release.KeyETag{Key: ghost, ETag: "ghost"})
		slices.SortFunc(kes, func(x, y release.KeyETag) int { return strings.Compare(x.Key, y.Key) })
	}
	return kes, err
}

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

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

// viewFixture is a deployment of three accounts and five apps: records,
// sidecars, both manifests and a trash marker. live writes bump the change
// index; bypass writes the same directory as an old binary would, without
// bumping it.
type viewFixture struct {
	live   *viewCountingStore
	bypass *release.DirStore
	acct   [3]string
}

func newViewFixture(t *testing.T, bucket, indexed bool) *viewFixture {
	t.Helper()
	root := t.TempDir()
	d := release.NewDirStore(root)
	d.EnableChangeIndex()
	f := &viewFixture{live: &viewCountingStore{DirStore: d, bucket: bucket}, bypass: release.NewDirStore(root)}
	var w release.Store = f.live
	if !indexed {
		w = f.bypass
	}
	for i := range f.acct {
		f.acct[i] = writeAccountManifest(t, w, int64(301+i), fmt.Sprintf("Account %d", i))
	}
	a, b, c := f.acct[0], f.acct[1], f.acct[2]
	for _, m := range []struct{ acct, app string }{{a, "shop"}, {a, "blog"}, {b, "docs"}, {c, "store"}} {
		if err := PutManifest(ctx, w, m.acct, m.app, Manifest{Schema: 1, App: m.app, Domain: m.app + ".example.com"}); err != nil {
			t.Fatal(err)
		}
	}
	f.record(t, w, a, "shop", "one")
	f.record(t, w, a, "shop", "two")
	f.record(t, w, a, "blog", "one")
	f.record(t, w, b, "docs", "one")
	f.record(t, w, c, "store", "one")
	// A record that reads but does not decode: unreadable, by its path host.
	if _, err := w.Put(ctx, RecordKey(c, "store", "bad.store.example.com"), []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	for _, m := range []struct{ acct, app string }{{a, "shop"}, {b, "wiki"}} {
		if err := PutSidecarManifest(ctx, w, m.acct, m.app, SidecarManifest{Schema: 1, App: m.app, Args: []string{"run"}, Channel: "stable"}); err != nil {
			t.Fatal(err)
		}
		f.sidecar(t, w, m.acct, m.app, "bot")
	}
	f.trash(t, w, a, "blog")
	// Keys outside the record families.
	for _, k := range []string{
		"apps/" + a + "/shop/logs/2026/10/21/chunk-1.jsonl",
		"apps/" + a + "/shop/channels/stable.json",
		"apps/" + a + "/shop/config.json",
		"accounts/" + a + "/members/owner.json",
	} {
		if _, err := w.Put(ctx, k, []byte(`{}`), ""); err != nil {
			t.Fatal(err)
		}
	}
	f.live.reset()
	return f
}

func viewRecord(acct, app, slug string) Record {
	return Record{Schema: 1, Host: slug + "." + app + ".example.com", Slug: slug, App: app,
		Account: acct, Region: "ie", Channel: "stable", CreatedAt: time.Now()}
}

func (f *viewFixture) record(t *testing.T, s release.Store, acct, app, slug string) {
	t.Helper()
	if err := CreateRecord(ctx, s, viewRecord(acct, app, slug)); err != nil {
		t.Fatal(err)
	}
}

func (f *viewFixture) sidecar(t *testing.T, s release.Store, acct, app, name string) {
	t.Helper()
	rec := SidecarRecord{Schema: 1, Name: name, App: app, Account: acct, Host: SidecarHost(name, app, acct), Channel: "stable"}
	if err := CreateSidecarRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
}

func (f *viewFixture) trash(t *testing.T, s release.Store, acct, app string) {
	t.Helper()
	body := mustJSON(t, release.AppTombstone{Schema: 1, Account: acct, App: app, DeletedAt: time.Now().UTC(), Actor: "t"})
	if _, err := s.Put(ctx, release.AppTombstoneKey(acct, app), body, ""); err != nil {
		t.Fatal(err)
	}
}

func (f *viewFixture) view() *RecordView {
	return NewRecordView(f.live, release.NewChangeWatch(f.live))
}

func mustRebuild(t *testing.T, v *RecordView) *ViewSnapshot {
	t.Helper()
	if err := v.Rebuild(ctx); err != nil {
		t.Fatal(err)
	}
	sn := v.Snapshot()
	if sn == nil {
		t.Fatal("Rebuild published nothing")
	}
	return sn
}

func mustRefresh(t *testing.T, v *RecordView) *ViewSnapshot {
	t.Helper()
	if err := v.Refresh(ctx); err != nil {
		t.Fatal(err)
	}
	sn := v.Snapshot()
	if sn == nil {
		t.Fatal("Refresh published nothing")
	}
	return sn
}

// walkAnswer is everything the record consumers read, as one comparable
// value: the strict record set and its unreadable hosts, the known accounts,
// each app's sidecar records, both manifests and the trash marker.
type walkAnswer struct {
	Records    []Record
	Unreadable map[string]bool
	Known      map[string]bool
	ReadFailed bool
	PerApp     map[string]string
}

func readWalk(t *testing.T, s release.Store, f *viewFixture) walkAnswer {
	t.Helper()
	recs, unreadable, err := ListAllRecordsStrict(ctx, s)
	if err != nil {
		t.Fatalf("ListAllRecordsStrict: %v", err)
	}
	for i := range recs {
		recs[i].CreatedAt = recs[i].CreatedAt.UTC()
	}
	known, readFailed, err := knownAccounts(ctx, s)
	if err != nil {
		t.Fatalf("knownAccounts: %v", err)
	}
	per := map[string]string{}
	for _, acct := range f.acct {
		for _, app := range []string{"shop", "blog", "docs", "wiki", "store", "fresh"} {
			m, merr := GetManifest(ctx, s, acct, app)
			sm, serr := GetSidecarManifest(ctx, s, acct, app)
			tomb, terr := release.ReadAppTombstone(ctx, s, acct, app)
			sides, _, lerr := ListSidecarRecords(ctx, s, acct, app, 100)
			per[acct+"/"+app] = fmt.Sprintf("m=%s/%v sm=%s/%v tomb=%v/%v sides=%s/%v",
				mustJSON(t, m), errClass(merr), mustJSON(t, sm), errClass(serr), tomb != nil, errClass(terr), mustJSON(t, sides), errClass(lerr))
		}
	}
	return walkAnswer{Records: recs, Unreadable: unreadable, Known: known, ReadFailed: readFailed, PerApp: per}
}

func errClass(err error) string {
	switch {
	case err == nil:
		return "ok"
	case errors.Is(err, os.ErrNotExist):
		return "absent"
	}
	return "error: " + err.Error()
}

func assertMatchesWalk(t *testing.T, f *viewFixture, sn *ViewSnapshot) {
	t.Helper()
	want := readWalk(t, f.live, f)
	got := readWalk(t, sn.Store(f.live), f)
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("over the snapshot:\n%+v\nover the bucket:\n%+v", got, want)
	}
}

func viewHosts(t *testing.T, s release.Store) []string {
	t.Helper()
	recs, err := ListAllRecords(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	var hosts []string
	for _, r := range recs {
		hosts = append(hosts, r.Host)
	}
	return hosts
}

func TestViewMatchesWalk(t *testing.T) {
	t.Parallel()
	for _, bucket := range []bool{false, true} {
		t.Run(fmt.Sprintf("bucket=%v", bucket), func(t *testing.T) {
			t.Parallel()
			f := newViewFixture(t, bucket, true)
			v := f.view()
			sn := mustRefresh(t, v)
			if !sn.Complete {
				t.Fatal("a clean first build is not complete")
			}
			if !release.HasBucketListDirs(sn.Store(f.live)) {
				t.Fatal("the snapshot store does not promise bucket-shaped ListDirs, so the record walk would list apps/ whole")
			}
			assertMatchesWalk(t, f, sn)
			if len(readWalk(t, f.live, f).Records) != 5 {
				t.Fatal("fixture: want five readable records")
			}

			// Incremental changes in every family, then a Refresh.
			a, b, c := f.acct[0], f.acct[1], f.acct[2]
			f.record(t, f.live, b, "docs", "two")
			moved := viewRecord(a, "shop", "one")
			moved.Region = "us"
			putReplace(t, f.live, RecordKey(a, "shop", moved.Host), mustJSON(t, moved))
			f.record(t, f.live, a, "fresh", "one")
			f.trash(t, f.live, a, "shop")
			if err := f.live.Delete(ctx, release.AppTombstoneKey(a, "blog")); err != nil {
				t.Fatal(err)
			}
			if err := f.live.Delete(ctx, SidecarRecordKey(b, "wiki", "bot")); err != nil {
				t.Fatal(err)
			}
			if err := PutManifest(ctx, f.live, c, "store", Manifest{Schema: 1, App: "store", Domain: "other.example.com"}); err != nil {
				t.Fatal(err)
			}
			soft := domains.AccountManifest{ID: 303, Sqid: c, Name: "gone", CreatedAt: time.Now().UTC(), DeletedAt: time.Now().UTC()}
			putReplace(t, f.live, "accounts/"+c+"/account.json", mustJSON(t, soft))
			sn2 := mustRefresh(t, v)
			if sn2.Epoch <= sn.Epoch || !sn2.Complete {
				t.Fatalf("refresh published epoch %d complete %v after %d", sn2.Epoch, sn2.Complete, sn.Epoch)
			}
			assertMatchesWalk(t, f, sn2)
		})
	}
}

// Complete is false exactly when the walk over the bucket would be partial:
// an account whose apps will not list, or an account manifest that will not
// read cleanly. A strict reader over the snapshot fails as it does live.
func TestViewCompleteMirrorsTheWalk(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	a, b := f.acct[0], f.acct[1]
	v := f.view()
	f.live.setFailList("apps/" + b + "/")
	f.live.setGhost(RecordKey(a, "shop", "ghost.shop.example.com"))
	sn := mustRebuild(t, v)
	if sn.Complete {
		t.Fatal("a walk that could not list an account's apps is complete")
	}
	if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
		t.Fatal("a strict reader over a partial walk did not fail")
	}
	f.live.setFailList("")
	sn = mustRefresh(t, v)
	if !sn.Complete {
		t.Fatal("the refresh after a partial walk did not walk again")
	}
	assertMatchesWalk(t, f, sn)

	putReplace(t, f.live, "accounts/"+b+"/account.json", []byte("{not json"))
	sn = mustRefresh(t, v)
	if sn.Complete {
		t.Fatal("a snapshot holding a malformed account manifest is complete")
	}
	_, _, lerr := ListAllRecordsStrict(ctx, f.live)
	_, _, serr := ListAllRecordsStrict(ctx, sn.Store(f.live))
	if lerr == nil || serr == nil {
		t.Fatalf("strict readers over a malformed manifest: live %v, snapshot %v; want both to fail", lerr, serr)
	}

	// An account listing that fails: its manifest reads as failed, not absent.
	f.live.setFailList("accounts/" + b + "/")
	putReplace(t, f.live, "accounts/"+b+"/account.json", accountManifestBody(302, b, "Account 1"))
	sn = mustRefresh(t, v)
	if sn.Complete {
		t.Fatal("a snapshot whose account listing failed is complete")
	}
	if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
		t.Fatal("a strict reader over a failed account listing did not fail")
	}
	f.live.setFailList("")
	if sn = mustRefresh(t, v); !sn.Complete {
		t.Fatal("still incomplete after the account listing recovered")
	}
	assertMatchesWalk(t, f, sn)
}

// A walk that cannot list one account's apps does not know them: Get and List
// of their records and manifests over the snapshot answer that failure, never
// "absent" or "empty", until a refresh or walk reads them. Found by the
// invariant simulation (a snapshot answered absent for keys no one deleted).
func TestViewUnlistedAccountIsNotAbsent(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	b := f.acct[1]
	f.live.setFailList("apps/" + b + "/")
	st := mustRebuild(t, f.view()).Store(f.live)
	for _, key := range []string{ManifestKey(b, "docs"), RecordKey(b, "docs", "one.docs.example.com"), SidecarRecordKey(b, "wiki", "bot")} {
		if _, _, err := st.Get(ctx, key); err == nil || errors.Is(err, os.ErrNotExist) {
			t.Fatalf("Get %s over a walk that could not list %s's apps = %v, want the listing failure", key, b, err)
		}
	}
	for _, p := range []string{RecordsPrefix(b, "docs"), SidecarsPrefix(b, "wiki")} {
		if keys, err := st.List(ctx, p); err == nil {
			t.Fatalf("List %s over a walk that could not list %s's apps = %v, nil; want the listing failure", p, b, keys)
		}
	}
	if _, err := GetManifest(ctx, st, f.acct[0], "shop"); err != nil {
		t.Fatalf("an account whose apps listed: %v", err)
	}
}

// A refresh whose caller goes away mid-read publishes what it read, with the
// reads that died cached as failures: the view knows it is behind, so it must
// not leave the previous snapshot standing as Complete.
func TestViewCancelledRefreshFailsClosed(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	before := mustRebuild(t, v)
	a := f.acct[0]
	f.record(t, f.live.DirStore, a, "shop", "three")
	cctx, cancel := context.WithCancel(ctx)
	defer cancel()
	f.live.mu.Lock()
	f.live.cancelOn, f.live.cancel = RecordsPrefix(a, "shop"), cancel
	f.live.mu.Unlock()
	if err := v.Refresh(cctx); err == nil {
		t.Fatal("a refresh cancelled mid-read reported success")
	}
	sn := v.Snapshot()
	if sn == before || sn.Complete {
		t.Fatalf("after a cancelled refresh: same snapshot %v, complete %v; want a new, incomplete one", sn == before, sn.Complete)
	}
	if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
		t.Fatal("a strict reader over a refresh that died mid-read did not fail")
	}
	f.live.mu.Lock()
	f.live.cancelOn = ""
	f.live.mu.Unlock()
	sn = mustRefresh(t, v)
	if !sn.Complete {
		t.Fatal("still incomplete after the next refresh")
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("records after the next refresh = %v", hosts)
	}
}

// A walk that fails at the top (the index unusable, then accounts/ will not
// list) must not leave the last good snapshot standing as Complete: the
// unchanged readers over it fail as they fail on the bucket.
func TestViewFailedWalkFailsClosed(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	good := mustRebuild(t, v)
	f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
	f.live.setFailGet(release.ChangeIndexKey)
	f.live.setFailList("accounts/")
	if err := v.Refresh(ctx); err == nil {
		t.Fatal("a refresh whose walk failed reported success")
	}
	sn := v.Snapshot()
	if sn == nil || sn.Epoch <= good.Epoch || sn.Complete {
		t.Fatalf("after a failed walk: %+v; want a newer, incomplete snapshot", sn)
	}
	st := sn.Store(f.live)
	if _, _, err := ListAllRecordsStrict(ctx, st); err == nil {
		t.Fatal("a strict reader over a failed walk did not fail")
	}
	if _, _, err := knownAccounts(ctx, st); err == nil {
		t.Fatal("knownAccounts over a failed walk did not fail")
	}
	if _, err := st.ListDirs(ctx, "apps/"); err == nil {
		t.Fatal("ListDirs(apps/) over a failed walk did not fail")
	}
	f.live.setFailGet("")
	f.live.setFailList("")
	sn = mustRefresh(t, v)
	if !sn.Complete {
		t.Fatal("still incomplete after the walk recovered")
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("records after the recovered walk = %v", hosts)
	}
}

// Over a live store whose ListDirs is not bucket-shaped (the fleet door's
// ListDirs("apps/") is its own account's apps), a write through the snapshot
// store must not send the walk's ListDirs to live: walked as accounts, those
// app directories hold no records, and reconcile would retire every route.
func TestSnapshotStoreWriteOverNonBucketLive(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, false, true)
	a := f.acct[0]
	live := fleetShaped{Store: f.live, acct: a}
	v := NewRecordView(live, release.NewChangeWatch(live))
	sn := mustRefresh(t, v)
	st := sn.Store(live)
	rec := viewRecord(a, "fresh", "one")
	if err := CreateRecord(ctx, st, rec); err != nil {
		t.Fatal(err)
	}
	want := viewHosts(t, f.live)
	if !slices.Contains(want, rec.Host) || len(want) != 6 {
		t.Fatalf("fixture: the bucket holds %v", want)
	}
	if got := viewHosts(t, st); !reflect.DeepEqual(got, want) {
		t.Fatalf("records over the snapshot store after a write = %v, want %v", got, want)
	}
	if dirs, err := st.ListDirs(ctx, "apps/"+a+"/"); err != nil || !slices.Contains(dirs, "apps/"+a+"/fresh/") {
		t.Fatalf("ListDirs after the write = %v, %v; want the new app", dirs, err)
	}
}

func TestViewRefreshRereadsOnlyMovedApp(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	mustRebuild(t, v)
	a := f.acct[0]
	f.record(t, f.live.DirStore, a, "shop", "three")
	f.live.reset()
	sn := mustRefresh(t, v)
	app := "apps/" + a + "/shop/"
	want := []string{
		"GetIfChanged " + app + "deleted.json",
		"GetIfChanged " + app + "instances.json",
		"GetIfChanged " + app + "instances/three.shop.example.com.json",
		"GetIfChanged " + app + "sidecars.json",
		"GetIfChanged " + release.ChangeIndexKey,
		"ListETags " + app + "instances/",
		"ListETags " + app + "sidecars/",
	}
	if got := f.live.got(); !reflect.DeepEqual(got, want) {
		t.Fatalf("one record in shop read\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("records over the snapshot = %v, want the new one", hosts)
	}
}

func TestViewRefreshNoChangeIsOneGet(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	mustRebuild(t, v)
	f.live.reset()
	mustRefresh(t, v)
	if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
		t.Fatalf("a quiet refresh read %v, want %v", got, want)
	}
}

func TestViewRetriesFailedScopeOn304(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	mustRebuild(t, v)
	a := f.acct[0]
	f.record(t, f.live.DirStore, a, "shop", "three")
	f.live.setFailList(RecordsPrefix(a, "shop"))
	sn := mustRefresh(t, v)
	if sn.Complete {
		t.Fatal("a scope whose re-read failed is reported complete")
	}
	if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
		t.Fatal("a strict reader over an incomplete snapshot did not fail")
	}
	if _, complete, err := ListAllRecordsTolerant(ctx, sn.Store(f.live)); err != nil || complete {
		t.Fatalf("a tolerant reader over an incomplete snapshot = complete %v, %v", complete, err)
	}

	f.live.setFailList("")
	f.live.reset()
	sn = mustRefresh(t, v)
	if !slices.Contains(f.live.got(), "ListETags "+RecordsPrefix(a, "shop")) {
		t.Fatalf("the failed scope was not re-read on an unchanged index: %v", f.live.got())
	}
	if slices.Contains(f.live.got(), "ListETags accounts/") || slices.Contains(f.live.got(), "List accounts/") {
		t.Fatalf("the retry walked the deployment: %v", f.live.got())
	}
	if !sn.Complete {
		t.Fatal("still incomplete after the retry succeeded")
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("records after the retry = %v", hosts)
	}
}

// A stalled index GET never holds a refresh, or the backstop rebuild, for
// longer than the poll bound, whatever the caller's context: each gives up
// on the index, walks the bucket (which answers), and publishes. The watch
// is unknown meanwhile, and the next good poll brings the view back to the
// index.
func TestViewStalledIndexIsBounded(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	v.pollTimeout = 50 * time.Millisecond
	mustRebuild(t, v)
	f.record(t, f.live.DirStore, f.acct[1], "docs", "two")
	f.live.setStall(release.ChangeIndexKey)
	f.live.reset()
	for _, step := range []struct {
		name string
		run  func(context.Context) error
	}{{"Refresh", v.Refresh}, {"Rebuild", v.Rebuild}} {
		done := make(chan error, 1)
		go func() { done <- step.run(context.Background()) }()
		select {
		case err := <-done:
			if err != nil {
				t.Fatalf("%s over a stalled index: %v", step.name, err)
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("%s over a stalled index did not return: its poll is unbounded", step.name)
		}
		sn := v.Snapshot()
		if !sn.Complete || !slices.Contains(viewHosts(t, sn.Store(f.live)), "two.docs.example.com") {
			t.Fatalf("%s over a stalled index: complete %v, records %v; want the walk's", step.name, sn.Complete, viewHosts(t, sn.Store(f.live)))
		}
		if v.watch.Usable() {
			t.Fatalf("%s: the watch vouches after its poll gave up", step.name)
		}
	}
	if n := slices.Index(f.live.got(), "GetIfChanged "+release.ChangeIndexKey); n < 0 {
		t.Fatalf("no index poll at all: %v", f.live.got())
	}
	polls := 0
	for _, c := range f.live.got() {
		if c == "GetIfChanged "+release.ChangeIndexKey {
			polls++
		}
	}
	if polls != 2 {
		t.Fatalf("the index was polled %d time(s) for one Refresh and one Rebuild, want 2: a refresh that gave up polls again before its walk", polls)
	}
	f.live.setStall("")
	f.live.reset()
	mustRefresh(t, v)
	if !v.watch.Usable() {
		t.Fatal("the watch did not recover on the next good poll")
	}
	f.live.reset()
	mustRefresh(t, v)
	if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
		t.Fatalf("a quiet refresh after recovery read %v, want %v", got, want)
	}
}

func TestViewFirstRefreshWithoutIndexRebuilds(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, false)
	if _, _, err := f.bypass.Get(ctx, release.ChangeIndexKey); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("fixture wrote an index: %v", err)
	}
	v := f.view()
	sn := mustRefresh(t, v)
	if !sn.Complete {
		t.Fatal("a first build over a bucket with no index is not complete")
	}
	assertMatchesWalk(t, f, sn)
	if len(viewHosts(t, sn.Store(f.live))) != 5 {
		t.Fatal("the first build missed records")
	}
}

func TestViewUnusableWatchRebuilds(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	mustRebuild(t, v)
	putReplace(t, f.bypass, release.ChangeIndexKey, []byte("{not json"))
	f.record(t, f.bypass, f.acct[1], "docs", "two")
	f.live.reset()
	sn := mustRefresh(t, v)
	if !slices.Contains(f.live.got(), "ListETags accounts/") {
		t.Fatalf("an unusable index did not walk: %v", f.live.got())
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "two.docs.example.com") {
		t.Fatalf("records after the walk = %v", hosts)
	}
}

func TestViewBypassWriteAppearsAfterRebuild(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	mustRebuild(t, v)
	f.record(t, f.bypass, f.acct[0], "shop", "three")
	sn := mustRefresh(t, v)
	if hosts := viewHosts(t, sn.Store(f.live)); slices.Contains(hosts, "three.shop.example.com") {
		t.Fatal("a write that bypassed the index appeared without a walk")
	}
	sn = mustRebuild(t, v)
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("records after the backstop walk = %v", hosts)
	}
	assertMatchesWalk(t, f, sn)
}

func TestViewPurgeDropsApp(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	sn := mustRebuild(t, v)
	a := f.acct[0]
	dirs, err := sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
	if err != nil || !slices.Contains(dirs, "apps/"+a+"/shop/") {
		t.Fatalf("ListDirs before the purge = %v, %v", dirs, err)
	}
	keys, err := f.live.DirStore.List(ctx, "apps/"+a+"/shop/")
	if err != nil {
		t.Fatal(err)
	}
	for _, k := range keys {
		if err := f.live.DirStore.Delete(ctx, k); err != nil {
			t.Fatal(err)
		}
	}
	sn = mustRefresh(t, v)
	dirs, err = sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
	if err != nil || slices.Contains(dirs, "apps/"+a+"/shop/") || !slices.Contains(dirs, "apps/"+a+"/blog/") {
		t.Fatalf("ListDirs after the purge = %v, %v; want blog and no shop", dirs, err)
	}
	assertMatchesWalk(t, f, sn)
}

func TestViewEpochsIncrease(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	if v.Snapshot() != nil {
		t.Fatal("a view that never published has a snapshot")
	}
	e1 := mustRebuild(t, v)
	f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
	e2 := mustRefresh(t, v)
	if e2.Epoch <= e1.Epoch {
		t.Fatalf("refresh epoch %d after %d", e2.Epoch, e1.Epoch)
	}
	if again := mustRefresh(t, v); again != e2 {
		t.Fatal("a refresh that read nothing published a new snapshot")
	}
	e3 := mustRebuild(t, v)
	if e3.Epoch <= e2.Epoch {
		t.Fatalf("rebuild epoch %d after %d", e3.Epoch, e2.Epoch)
	}
}

// The backstop holds mu for its whole walk, so an incremental refresh that
// starts during it waits, and runs after the rebuild publishes: the rebuild
// cannot publish over what the refresh read. Serial: it sets a package hook.
func TestViewBackstopCannotOverwriteNewer(t *testing.T) {
	f := newViewFixture(t, true, true)
	v := f.view()
	first := mustRebuild(t, v)

	paused, resume := make(chan struct{}), make(chan struct{})
	var once sync.Once
	recordViewRebuildHook = func() {
		once.Do(func() {
			close(paused)
			<-resume
		})
	}
	t.Cleanup(func() { recordViewRebuildHook = nil })

	rebuilt := make(chan error, 1)
	go func() { rebuilt <- v.Rebuild(ctx) }()
	<-paused
	f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
	refreshed := make(chan error, 1)
	go func() { refreshed <- v.Refresh(ctx) }()
	select {
	case err := <-refreshed:
		t.Fatalf("a refresh ran while the backstop walk held the view: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	close(resume)
	if err := <-rebuilt; err != nil {
		t.Fatal(err)
	}
	if err := <-refreshed; err != nil {
		t.Fatal(err)
	}
	sn := v.Snapshot()
	if sn.Epoch != first.Epoch+2 || !sn.Complete {
		t.Fatalf("final epoch %d complete %v, want %d and complete", sn.Epoch, sn.Complete, first.Epoch+2)
	}
	if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
		t.Fatalf("the record written during the walk is missing: %v", hosts)
	}
}

func TestSnapshotStoreReadYourWrites(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	sn := mustRebuild(t, v)
	a := f.acct[0]
	st := sn.Store(f.live)
	rec := viewRecord(a, "fresh", "one")
	if err := CreateRecord(ctx, st, rec); err != nil {
		t.Fatal(err)
	}
	if got, err := GetRecord(ctx, st, a, "fresh", rec.Host); err != nil || got.Host != rec.Host {
		t.Fatalf("Get after Put = %+v, %v", got, err)
	}
	if keys, err := st.List(ctx, RecordsPrefix(a, "fresh")); err != nil || !slices.Contains(keys, RecordKey(a, "fresh", rec.Host)) {
		t.Fatalf("List after Put = %v, %v", keys, err)
	}
	if dirs, err := st.ListDirs(ctx, "apps/"+a+"/"); err != nil || !slices.Contains(dirs, "apps/"+a+"/fresh/") {
		t.Fatalf("ListDirs after Put = %v, %v", dirs, err)
	}
	if err := PutManifest(ctx, st, a, "shop", Manifest{Schema: 1, App: "shop", Domain: "moved.example.com"}); err != nil {
		t.Fatal(err)
	}
	if m, err := GetManifest(ctx, st, a, "shop"); err != nil || m.Domain != "moved.example.com" {
		t.Fatalf("manifest after a CAS write = %+v, %v", m, err)
	}
	gone := RecordKey(a, "shop", "one.shop.example.com")
	if err := st.Delete(ctx, gone); err != nil {
		t.Fatal(err)
	}
	if _, _, err := st.Get(ctx, gone); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("Get after Delete = %v", err)
	}
	if keys, _ := st.List(ctx, RecordsPrefix(a, "shop")); slices.Contains(keys, gone) {
		t.Fatalf("List after Delete still has %s", gone)
	}
	// The snapshot itself never moves: a fresh store over it still answers
	// from the cache.
	if keys, _ := sn.Store(f.live).List(ctx, RecordsPrefix(a, "shop")); !slices.Contains(keys, gone) {
		t.Fatal("a write through one snapshot store changed the snapshot")
	}
}

// Seeded without the index, so the failed scope is not in it: only the
// view's own record of the failure brings it back on the next Refresh.
func TestSnapshotStoreErrorsMatchLive(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, false)
	a, b := f.acct[0], f.acct[1]
	f.live.setFailGet(ManifestKey(b, "docs"))
	v := f.view()
	sn := mustRebuild(t, v)
	if sn.Complete {
		t.Fatal("a snapshot holding a failed manifest read is complete")
	}
	st := sn.Store(f.live)
	for _, key := range []string{
		RecordKey(a, "shop", "missing.shop.example.com"),
		RecordKey(a, "nowhere", "missing.nowhere.example.com"),
		SidecarRecordKey(a, "shop", "missing"),
		ManifestKey(b, "wiki"),
		SidecarManifestKey(a, "blog"),
		release.AppTombstoneKey(a, "shop"),
		"accounts/zzzz/account.json",
	} {
		_, _, lerr := f.live.DirStore.Get(ctx, key)
		_, _, serr := st.Get(ctx, key)
		if !errors.Is(lerr, os.ErrNotExist) || !errors.Is(serr, os.ErrNotExist) {
			t.Fatalf("Get %s: live %v, snapshot %v; want not-found from both", key, lerr, serr)
		}
	}
	if _, err := GetManifest(ctx, st, b, "docs"); err == nil || errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a manifest whose read failed = %v, want the read error, not absence", err)
	}
	if tomb, err := release.ReadAppTombstone(ctx, st, a, "shop"); tomb != nil || err != nil {
		t.Fatalf("absent trash marker = %+v, %v", tomb, err)
	}
	if tomb, err := release.ReadAppTombstone(ctx, st, a, "blog"); tomb == nil || err != nil {
		t.Fatalf("trash marker = %+v, %v", tomb, err)
	}

	f.live.setFailGet("")
	sn = mustRefresh(t, v)
	if !sn.Complete {
		t.Fatal("still incomplete after the manifest read recovered")
	}
	if m, err := GetManifest(ctx, sn.Store(f.live), b, "docs"); err != nil || m.App != "docs" {
		t.Fatalf("recovered manifest = %+v, %v", m, err)
	}
}

func TestSnapshotStoreOtherKeysGoLive(t *testing.T) {
	t.Parallel()
	f := newViewFixture(t, true, true)
	v := f.view()
	sn := mustRebuild(t, v)
	a := f.acct[0]
	if _, err := f.bypass.Put(ctx, "apps/"+a+"/shop/channels/edge.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	putReplace(t, f.bypass, "apps/"+a+"/shop/config.json", []byte(`{"after":true}`))
	f.live.reset()
	st := sn.Store(f.live)
	body, _, err := st.Get(ctx, "apps/"+a+"/shop/config.json")
	if err != nil || string(body) != `{"after":true}` {
		t.Fatalf("config.json over the snapshot = %s, %v", body, err)
	}
	keys, err := st.List(ctx, "apps/"+a+"/shop/channels/")
	if err != nil || !slices.Contains(keys, "apps/"+a+"/shop/channels/edge.json") {
		t.Fatalf("channels over the snapshot = %v, %v", keys, err)
	}
	want := []string{"Get apps/" + a + "/shop/config.json", "List apps/" + a + "/shop/channels/"}
	if got := f.live.got(); !reflect.DeepEqual(got, want) {
		t.Fatalf("live saw %v, want %v", got, want)
	}
}

func TestViewScopeApp(t *testing.T) {
	t.Parallel()
	for scope, want := range map[string][2]string{
		"apps/abc/shop": {"abc", "shop"},
		"apps/abc":      {},
		"apps/abc/":     {},
		"apps//shop":    {},
		"apps/a/b/c":    {},
		"accounts/abc":  {},
	} {
		acct, app, ok := ScopeApp(scope)
		if ok != (want[0] != "") || acct != want[0] || app != want[1] {
			t.Fatalf("ScopeApp(%q) = %q, %q, %v", scope, acct, app, ok)
		}
	}
}

// A pass over a snapshot keeps the hold write's re-read fresh: the record is
// re-read from the bucket, not from the snapshot the listing came from, so a
// move that went on after the view was refreshed is not held again (the
// guard TestReconcileHoldRereadsTheRecordBeforeHoldingARow pins on the walk).
func TestSnapshotHoldRereadReadsTheBucket(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	v := NewRecordView(s, release.NewChangeWatch(s))
	if err := v.Rebuild(ctx); err != nil {
		t.Fatal(err)
	}
	committed := rec
	m := *rec.Move
	m.State = MoveCommitted
	committed.Move = &m
	key := RecordKey(rec.Account, rec.App, rec.Host)
	_, etag, err := s.Get(ctx, key)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, key, mustJSON(t, committed), etag); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, RouteFor(rec, "/data", `["seed"]`, false, nil)) // the target's row, opened
	var wrote holdLog
	cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
		Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }}
	if err := Reconcile(ctx, v.Snapshot().Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 0 {
		t.Fatalf("an opened row was held again from the snapshot's record: %+v %v", got, wrote.hosts)
	}
}

// staleView builds a view over s and returns it Complete, so that writes
// made after it, with no bump (s has no change index: an old binary, or a
// bump that failed), are invisible to its snapshots until the backstop.
func staleView(t *testing.T, s release.Store) *RecordView {
	t.Helper()
	v := NewRecordView(s, release.NewChangeWatch(s))
	if err := v.Refresh(ctx); err != nil {
		t.Fatal(err)
	}
	return v
}

func completeSnapshot(t *testing.T, v *RecordView) *ViewSnapshot {
	t.Helper()
	if err := v.Refresh(ctx); err != nil {
		t.Fatal(err)
	}
	sn := v.Snapshot()
	if sn == nil || !sn.Complete {
		t.Fatal("fixture: the snapshot is not Complete")
	}
	return sn
}

// A record written without its bump is absent from a Complete snapshot. A
// pass over that snapshot confirms the absence in the bucket before it
// retires the row a walking pass minted, and keeps it; a record that is
// really gone is still retired.
func TestSnapshotAbsenceDoesNotRetireALiveRecord(t *testing.T) {
	s, sqid := setup(t)
	v := staleView(t, s)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil { // a walking pass (the view could not vouch)
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	sn := completeSnapshot(t, v)
	if recs, err := ListAllRecords(ctx, sn.Store(s)); err != nil || len(recs) != 0 {
		t.Fatalf("fixture: the snapshot has records %v, %v; it should lack the unbumped one", recs, err)
	}
	if err := Reconcile(ctx, sn.Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host) // kept: the bucket has the record

	if err := DeleteRecord(ctx, s, sqid, "tito", rec.Host); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
		t.Fatalf("a record really gone did not retire its row over the snapshot: %+v %v", rt, err)
	}
}

// The same for the trash: a marker the snapshot still has, taken away
// without a bump, does not retire the app's rows.
func TestSnapshotTrashMarkerIsConfirmedBeforeRemoval(t *testing.T) {
	_, pub := releaseKeys(t)
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	if err := release.TrashApp(ctx, s, sqid, "tito", "owner@example.com", time.Now()); err != nil {
		t.Fatal(err)
	}
	v := staleView(t, s) // sees the marker
	if err := release.RestoreApp(ctx, s, sqid, "tito"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host) // kept: the bucket has no marker any more
}

func TestSnapshotAbsenceDoesNotRetireALiveSidecar(t *testing.T) {
	s := dirStore(t)
	sqid := writeAccountManifest(t, s, acctID, "Shop")
	if err := PutSidecarManifest(ctx, s, sqid, "shop", SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable"}); err != nil {
		t.Fatal(err)
	}
	v := staleView(t, s)
	host := SidecarHost("bot", "shop", sqid)
	if err := CreateSidecarRecord(ctx, s, SidecarRecord{Schema: 1, Name: "bot", Host: host, App: "shop", Account: sqid, Channel: "stable"}); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{DataDir: t.TempDir()}
	if err := ReconcileSidecars(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, host)
	if err := ReconcileSidecars(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, host) // kept: the bucket has the record

	if err := s.Delete(ctx, SidecarRecordKey(sqid, "shop", "bot")); err != nil {
		t.Fatal(err)
	}
	if err := ReconcileSidecars(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(host); err != nil || rt != nil {
		t.Fatalf("a sidecar record really gone did not retire its row over the snapshot: %+v %v", rt, err)
	}
}

// removalWorld is one account with app shop (instances and sidecars
// manifests) over a bucket whose writes bump the change index (live), with a
// second handle on the same directory whose writes do not (bypass: a bump
// that failed, an old binary, a direct-bucket writer). v follows live.
type removalWorld struct {
	live   *viewCountingStore
	bypass *release.DirStore
	acct   string
	v      *RecordView
}

func newRemovalWorld(t *testing.T) *removalWorld {
	t.Helper()
	root := t.TempDir()
	d := release.NewDirStore(root)
	d.EnableChangeIndex()
	w := &removalWorld{live: &viewCountingStore{DirStore: d}, bypass: release.NewDirStore(root)}
	w.acct = writeAccountManifest(t, w.live, 401, "Shop")
	if err := PutManifest(ctx, w.live, w.acct, "shop", Manifest{Schema: 1, App: "shop", Domain: "example.com"}); err != nil {
		t.Fatal(err)
	}
	if err := PutSidecarManifest(ctx, w.live, w.acct, "shop", SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable"}); err != nil {
		t.Fatal(err)
	}
	w.v = NewRecordView(w.live, release.NewChangeWatch(w.live))
	return w
}

// snap is the view's current Complete snapshot as a store over live.
func (w *removalWorld) snap(t *testing.T) release.Store {
	t.Helper()
	return completeSnapshot(t, w.v).Store(w.live)
}

func (w *removalWorld) record(app, slug, region, channel string) Record {
	return Record{Schema: 1, Host: slug + ".example.com", Slug: slug, App: app, Account: w.acct,
		Region: region, Channel: channel, CreatedAt: time.Now()}
}

// listings counts the record-family listings live answered since the last
// reset: what a walk costs and a snapshot pass must not.
func (w *removalWorld) listings() int {
	n := 0
	for _, c := range w.live.got() {
		for _, m := range []string{"List ", "ListDirs ", "ListETags "} {
			if strings.HasPrefix(c, m+"accounts/") || strings.HasPrefix(c, m+"apps/") {
				n++
			}
		}
	}
	return n
}

// Final review findings 2 and 4 (amadan #336): a removal a snapshot proves
// by absence is decided by the walk's own rules over the live bucket. A
// sidecar whose placement becomes no-inbound, in its record or in the app's
// sidecars manifest, with the bump landing, leaves an inbound box: the live
// walk does not place it either. Before, the confirmation asked only region
// and pool, and kept the row for ever.
func TestSnapshotSidecarPlacementChangeEvacuates(t *testing.T) {
	for _, tc := range []struct {
		name  string
		apply func(t *testing.T, w *removalWorld, rec SidecarRecord)
	}{
		{"record", func(t *testing.T, w *removalWorld, rec SidecarRecord) {
			rec.Placement = PlacementNoInbound
			putReplace(t, w.live, SidecarRecordKey(w.acct, "shop", "bot"), mustJSON(t, rec))
		}},
		{"manifest", func(t *testing.T, w *removalWorld, _ SidecarRecord) {
			putReplace(t, w.live, SidecarManifestKey(w.acct, "shop"), mustJSON(t,
				SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable", Placement: PlacementNoInbound}))
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			w := newRemovalWorld(t)
			host := SidecarHost("bot", "shop", w.acct)
			rec := SidecarRecord{Schema: 1, Name: "bot", Host: host, App: "shop", Account: w.acct, Channel: "stable"}
			if err := CreateSidecarRecord(ctx, w.live, rec); err != nil {
				t.Fatal(err)
			}
			reg := regWith(t)
			cfg := ReconcileConfig{DataDir: t.TempDir()} // an inbound box
			if err := ReconcileSidecars(ctx, w.snap(t), reg, cfg); err != nil {
				t.Fatal(err)
			}
			mustGet(t, reg, host)
			sn := w.snap(t)
			w.live.reset()
			if err := ReconcileSidecars(ctx, sn, reg, cfg); err != nil {
				t.Fatal(err)
			}
			if n := w.listings(); n != 0 {
				t.Fatalf("a snapshot pass with nothing to remove listed the bucket: %v", w.live.got())
			}

			tc.apply(t, w, rec)
			if err := ReconcileSidecars(ctx, w.snap(t), reg, cfg); err != nil {
				t.Fatal(err)
			}
			if rt, err := reg.Get(host); err != nil || rt != nil {
				t.Fatalf("a no-inbound sidecar stayed on an inbound box over the snapshot: %+v %v", rt, err)
			}
		})
	}
}

// A record the snapshot lacks, whose live copy has an empty channel and a
// region out of this pass's scope: the walk protects an unusable record
// before it asks about scope, so the row stays. Before, the confirmation
// decoded the record, found it out of scope, and removed the row.
func TestSnapshotRemovalProtectsAnInvalidOutOfScopeRecord(t *testing.T) {
	t.Parallel()
	w := newRemovalWorld(t)
	sn := w.snap(t) // before the record: it never reaches the view
	rec := w.record("shop", "one", "ie", "stable")
	if err := CreateRecord(ctx, w.bypass, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
	if err := Reconcile(ctx, w.live, reg, cfg); err != nil { // a walking pass mints
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	bad := rec
	bad.Region, bad.Channel = "uk", ""
	putReplace(t, w.bypass, RecordKey(w.acct, "shop", rec.Host), mustJSON(t, bad))
	if err := Reconcile(ctx, sn, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host) // protected, as the walk protects it
	if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host) // and the walk agrees
}

// A host whose record moved to another key (another app here) without a
// bump: the walk matches hosts across the whole deployment, so the row
// stays. Before, the confirmation read only the row's old key, found it
// gone, and removed the row. Once the record is gone everywhere, the row
// goes.
func TestSnapshotRemovalKeepsARehomedHost(t *testing.T) {
	t.Parallel()
	w := newRemovalWorld(t)
	w.snap(t) // built before the record: no bypass write below reaches the view
	rec := w.record("shop", "one", "ie", "stable")
	if err := CreateRecord(ctx, w.bypass, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
	if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	if err := DeleteRecord(ctx, w.bypass, w.acct, "shop", rec.Host); err != nil {
		t.Fatal(err)
	}
	moved := w.record("cafe", "one", "ie", "stable")
	if err := CreateRecord(ctx, w.bypass, moved); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, w.snap(t), reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host) // kept: the bucket still has the host

	if err := DeleteRecord(ctx, w.bypass, w.acct, "cafe", rec.Host); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, w.snap(t), reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
		t.Fatalf("a host really gone did not retire its row over the snapshot: %+v %v", rt, err)
	}
}

// Disarmed (any store that is not a snapshot), removal is the walk's, from
// the one walk the pass already made: no second walk.
func TestRemovalOverTheBucketWalksOnce(t *testing.T) {
	t.Parallel()
	w := newRemovalWorld(t)
	rec := w.record("shop", "one", "ie", "stable")
	if err := CreateRecord(ctx, w.live, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
	if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, rec.Host)
	if err := DeleteRecord(ctx, w.live, w.acct, "shop", rec.Host); err != nil {
		t.Fatal(err)
	}
	w.live.reset()
	if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
		t.Fatalf("a gone record's row survived a walking pass: %+v %v", rt, err)
	}
	n := 0
	for _, c := range w.live.got() {
		if c == "List accounts/" {
			n++
		}
	}
	if n != 1 {
		t.Fatalf("a walking pass listed accounts/ %d time(s), want 1: %v", n, w.live.got())
	}
}