package console

import (
	"context"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// sweepFixture is a console with a data bucket for seeds: a DirStore, so
// a snapshot's last-modified time is its file's mtime and a test can age
// it. Records are app shop's in the operator account.
type sweepFixture struct {
	fx   *consoleFixture
	data release.Store
	root string
}

const sweepApp = "shop"

func newSweepFixture(t *testing.T) sweepFixture {
	t.Helper()
	fx := newConsoleFixture(t)
	root := t.TempDir()
	data := release.NewDirStore(root)
	fx.srv.cfg.SeedData = data
	return sweepFixture{fx: fx, data: data, root: root}
}

// seeded writes a record of shop on host whose seed is a fresh id, of
// kind, in state, stamped at, and answers the id.
func (f sweepFixture) seeded(t *testing.T, host, kind, state string, at time.Time) string {
	t.Helper()
	id := newSeedID(t)
	f.recordOf(t, sweepApp, host, &provision.SeedSource{ID: id, Kind: kind, State: state, By: "bob@example.com", At: at})
	return id
}

// recordOf writes a record of app on host with seed (nil for none).
func (f sweepFixture) recordOf(t *testing.T, app, host string, seed *provision.SeedSource) {
	t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount,
		Channel: "stable", CreatedAt: time.Now().UTC(), Seed: seed}
	if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
		t.Fatal(err)
	}
}

// snapshot writes seeds/<host>/<id>/snapshot.db last modified age ago.
func (f sweepFixture) snapshot(t *testing.T, host, id string, age time.Duration) string {
	t.Helper()
	key := seeds.SnapshotKey(host, id)
	if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil {
		t.Fatal(err)
	}
	// DirStore keeps a key at root/key, and its ListMod reports the file's
	// mtime.
	when := time.Now().Add(-age)
	if err := os.Chtimes(filepath.Join(f.root, filepath.FromSlash(key)), when, when); err != nil {
		t.Fatal(err)
	}
	return key
}

func (f sweepFixture) snapshotExists(t *testing.T, key string) bool {
	t.Helper()
	_, _, err := f.data.Get(context.Background(), key)
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		t.Fatal(err)
	}
	return err == nil
}

// state is the seed state of shop's record on host, or "gone".
func (f sweepFixture) state(t *testing.T, host string) string {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return "gone"
	case err != nil:
		t.Fatal(err)
	}
	return rec.Seed.State
}

func (f sweepFixture) pass(now time.Time) { f.fx.srv.seedsPass(context.Background(), now) }

func TestSweepLeavesALiveReservation(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	id := f.seeded(t, "canary.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-10*time.Minute))
	key := f.snapshot(t, "canary.shop.example.com", id, 10*time.Minute)
	f.pass(now)
	if got := f.state(t, "canary.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, key) {
		t.Fatalf("a live reservation was touched: %s, snapshot kept %v", got, f.snapshotExists(t, key))
	}
	// Past the hour it is abandoned by CAS, and its snapshot goes with it.
	f.pass(now.Add(2 * time.Hour))
	if got := f.state(t, "canary.shop.example.com"); got != seeds.StateAborted {
		t.Fatalf("a stale reservation = %s, want aborted", got)
	}
	if f.snapshotExists(t, key) {
		t.Fatal("the snapshot of a seed this pass abandoned survived")
	}
}

// A --from copy is bounded at about 52 minutes (30m restore, 60s, 20m
// upload): a reservation younger than the hour is never abandoned.
func TestSweepReservationThresholdOutlastsTheCopy(t *testing.T) {
	if seedReserveTTL <= 30*time.Minute+time.Minute+20*time.Minute {
		t.Fatalf("seedReserveTTL %s does not outlast the --from copy's bound", seedReserveTTL)
	}
	f := newSweepFixture(t)
	now := time.Now()
	f.seeded(t, "copy.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-55*time.Minute))
	f.pass(now)
	if got := f.state(t, "copy.shop.example.com"); got != seeds.StateReserving {
		t.Fatalf("a 55-minute reservation = %s, want reserving", got)
	}
}

func TestSweepAbandonsOnlyStaleImports(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	old, young := now.Add(-8*24*time.Hour), now.Add(-6*24*time.Hour)
	f.seeded(t, "waiting.shop.example.com", seeds.KindImport, seeds.StateAwaitingUpload, old)
	f.seeded(t, "uploaded.shop.example.com", seeds.KindImport, seeds.StateUploaded, old)
	f.seeded(t, "recent.shop.example.com", seeds.KindImport, seeds.StateUploaded, young)
	f.seeded(t, "ready.shop.example.com", seeds.KindImport, seeds.StateReady, old)
	f.seeded(t, "refused.shop.example.com", seeds.KindImport, seeds.StateRefused, old)
	f.seeded(t, "fresh.shop.example.com", seeds.KindReplica, seeds.StateReady, old)
	f.pass(now)
	for host, want := range map[string]string{
		"waiting.shop.example.com":  seeds.StateAborted,
		"uploaded.shop.example.com": seeds.StateAborted,
		"recent.shop.example.com":   seeds.StateUploaded,
		"ready.shop.example.com":    seeds.StateReady,
		"refused.shop.example.com":  seeds.StateRefused,
		"fresh.shop.example.com":    seeds.StateReady,
	} {
		if got := f.state(t, host); got != want {
			t.Errorf("%s = %s, want %s", host, got, want)
		}
	}
}

func TestSweepDeletesOnlyUnreferencedOrSupersededSnapshots(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	keep, gone := map[string]string{}, map[string]string{}

	live := f.seeded(t, "a.shop.example.com", seeds.KindReplica, seeds.StateReady, now)
	keep["a snapshot a live record needs"] = f.snapshot(t, "a.shop.example.com", live, 3*time.Hour)
	gone["an orphan over an hour old beside a live seed"] = f.snapshot(t, "a.shop.example.com", newSeedID(t), 3*time.Hour)
	keep["a young unreferenced snapshot (its door may be between record and upload)"] =
		f.snapshot(t, "nobody.shop.example.com", newSeedID(t), 10*time.Minute)
	gone["an orphan with no record, over an hour old"] = f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)

	f.seeded(t, "b.shop.example.com", seeds.KindReplica, seeds.StatePlaced, now)
	gone["a young snapshot superseded by a placed seed"] = f.snapshot(t, "b.shop.example.com", newSeedID(t), 5*time.Minute)
	f.seeded(t, "c.shop.example.com", seeds.KindImport, seeds.StateCommitted, now)
	gone["a young snapshot superseded by a committed seed"] = f.snapshot(t, "c.shop.example.com", newSeedID(t), 5*time.Minute)
	f.seeded(t, "d.shop.example.com", seeds.KindImport, seeds.StateAborted, now)
	keep["a young snapshot beside another seed that is not placed"] = f.snapshot(t, "d.shop.example.com", newSeedID(t), 5*time.Minute)

	own := func(host, kind, state string) string {
		return f.snapshot(t, host, f.seeded(t, host, kind, state, now), 5*time.Minute)
	}
	gone["its own placed seed's (the box's delete was lost)"] = own("e.shop.example.com", seeds.KindReplica, seeds.StatePlaced)
	gone["its own committed seed's"] = own("f.shop.example.com", seeds.KindImport, seeds.StateCommitted)
	gone["its own aborted seed's"] = own("g.shop.example.com", seeds.KindImport, seeds.StateAborted)
	keep["its own refused seed's"] = own("h.shop.example.com", seeds.KindImport, seeds.StateRefused)
	keep["its own swapped seed's"] = own("i.shop.example.com", seeds.KindImport, seeds.StateSwapped)
	keep["its own ready import's"] = own("j.shop.example.com", seeds.KindImport, seeds.StateReady)

	// Not a snapshot: nothing else under seeds/ is the sweep's.
	stray := seeds.SnapshotPrefix("k.shop.example.com", newSeedID(t)) + "other.db"
	if _, err := f.data.Put(context.Background(), stray, []byte("x"), ""); err != nil {
		t.Fatal(err)
	}
	keep["an object that is not a snapshot"] = stray

	f.pass(now)
	for why, key := range keep {
		if !f.snapshotExists(t, key) {
			t.Errorf("deleted %s", why)
		}
	}
	for why, key := range gone {
		if f.snapshotExists(t, key) {
			t.Errorf("kept %s", why)
		}
	}
	if got := f.state(t, "g.shop.example.com"); got != seeds.StateAborted {
		t.Errorf("the aborted record = %s; the sweep deletes its snapshot and leaves the record to the box", got)
	}
}

func TestSweepKeepsSnapshotsOfAnUnreadableRecord(t *testing.T) {
	f := newSweepFixture(t)
	host := "a.shop.example.com"
	id := f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, time.Now())
	key := f.snapshot(t, host, id, 3*time.Hour)
	orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour)
	rk := provision.RecordKey(release.OpsAccount, sweepApp, host)
	_, etag, err := f.fx.rel.Get(context.Background(), rk)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.fx.rel.Put(context.Background(), rk, []byte("{not json"), etag); err != nil {
		t.Fatal(err)
	}
	f.pass(time.Now())
	if !f.snapshotExists(t, key) || !f.snapshotExists(t, orphan) {
		t.Fatalf("an unreadable record's host lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, key), f.snapshotExists(t, orphan))
	}
}

// Two records holding one host: neither decides for the other, so the
// sweep keeps everything of the host and abandons nothing on it.
func TestSweepKeepsAHostTwoRecordsHold(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	host := "twice.shop.example.com"
	id := f.seeded(t, host, seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
	f.recordOf(t, "blog", host, &provision.SeedSource{ID: newSeedID(t), Kind: seeds.KindReplica, State: seeds.StatePlaced,
		By: "bob@example.com", At: now})
	own := f.snapshot(t, host, id, 3*time.Hour)
	orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour)
	f.pass(now)
	if !f.snapshotExists(t, own) || !f.snapshotExists(t, orphan) {
		t.Fatalf("a host two records hold lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, own), f.snapshotExists(t, orphan))
	}
	if got := f.state(t, host); got != seeds.StateReserving {
		t.Fatalf("a host two records hold had its seed moved to %s", got)
	}
}

// The app stops the import between the abort's read and its write: the
// abort's CAS loses, the retry finds the seed moved on and refuses, and
// nothing is deleted.
func TestSweepAbandonsOnlyByCAS(t *testing.T) {
	f := newSweepFixture(t)
	host := "imp.shop.example.com"
	old := time.Now().Add(-8 * 24 * time.Hour)
	id := f.seeded(t, host, seeds.KindImport, seeds.StateUploaded, old)
	key := f.snapshot(t, host, id, 8*24*time.Hour)
	var calls []int
	seedsPassBeforeAbort = func(call int) {
		calls = append(calls, call)
		if call != 1 {
			return
		}
		if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host, provision.SeedStopped(id)); err != nil {
			t.Errorf("the app's stop: %v", err)
		}
	}
	t.Cleanup(func() { seedsPassBeforeAbort = nil })
	f.pass(time.Now())
	if len(calls) != 2 {
		t.Fatalf("the abort's mutation ran %d times, want 2 (a write that lost, then a refused retry)", len(calls))
	}
	if got := f.state(t, host); got != seeds.StateReady || !f.snapshotExists(t, key) {
		t.Fatalf("state %s, snapshot kept %v; want the app's stop to win and nothing deleted", got, f.snapshotExists(t, key))
	}
}

func TestSweepFinishesARolledBackSeed(t *testing.T) {
	f := newSweepFixture(t)
	host := "gone.shop.example.com"
	id := f.seeded(t, host, seeds.KindImport, seeds.StateRolledBack, time.Now())
	key := f.snapshot(t, host, id, time.Minute)
	f.pass(time.Now())
	if got := f.state(t, host); got != "gone" {
		t.Fatalf("a rolled-back record after the sweep: %s", got)
	}
	if f.snapshotExists(t, key) {
		t.Fatal("a finished seed's snapshot survived")
	}
}

// The host is deleted and seeded again between the two listings. Listed
// snapshots first, the new seed's snapshot is not in the listing, so the
// placed record of the old seed cannot condemn it (the doors write the
// record before the snapshot).
func TestSweepListsSnapshotsBeforeRecords(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	host := "again.shop.example.com"
	f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, now.Add(-time.Hour))
	var fresh string
	seedsPassBetweenListings = func() {
		if err := f.fx.rel.Delete(context.Background(), provision.RecordKey(release.OpsAccount, sweepApp, host)); err != nil {
			t.Error(err)
		}
		id := f.seeded(t, host, seeds.KindReplica, seeds.StateReady, now)
		fresh = f.snapshot(t, host, id, time.Minute)
	}
	t.Cleanup(func() { seedsPassBetweenListings = nil })
	f.pass(now)
	if fresh == "" {
		t.Fatal("the seam never ran")
	}
	if !f.snapshotExists(t, fresh) {
		t.Fatal("the new seed's snapshot was deleted as superseded")
	}
}

// failingAccountList fails the record walk's first listing.
type failingAccountList struct{ release.Store }

func (s failingAccountList) List(ctx context.Context, prefix string) ([]string, error) {
	if prefix == "accounts/" {
		return nil, errors.New("the bucket is unreachable")
	}
	return s.Store.List(ctx, prefix)
}

// failingModList is a data bucket whose listing fails.
type failingModList struct{ release.Store }

func (failingModList) ListMod(context.Context, string) ([]release.KeyMod, error) {
	return nil, errors.New("the bucket is unreachable")
}

// A listing that fails is no knowledge: the pass deletes nothing and
// abandons nothing.
func TestSweepDecidesNothingOnAFailedListing(t *testing.T) {
	for name, broke := range map[string]func(f sweepFixture){
		"records":   func(f sweepFixture) { f.fx.srv.cfg.Releases = failingAccountList{f.fx.rel} },
		"snapshots": func(f sweepFixture) { f.fx.srv.cfg.SeedData = failingModList{f.data} },
	} {
		t.Run(name, func(t *testing.T) {
			f := newSweepFixture(t)
			now := time.Now()
			stale := f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
			staleKey := f.snapshot(t, "stale.shop.example.com", stale, 2*time.Hour)
			orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)
			f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)
			broke(f)
			f.pass(now)
			if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving {
				t.Errorf("a stale reservation = %s after a failed listing", got)
			}
			if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack {
				t.Errorf("a rolled-back record = %s after a failed listing", got)
			}
			if !f.snapshotExists(t, staleKey) || !f.snapshotExists(t, orphan) {
				t.Error("a snapshot was deleted after a failed listing")
			}
		})
	}
}

// noModList is a data bucket that cannot list last-modified times.
type noModList struct{ release.Store }

// Without last-modified times no snapshot is judged, but stale seeds are
// still abandoned and rolled-back ones finished.
func TestSweepWithoutListModStillAbandonsAndFinishes(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
	f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)
	orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)
	f.fx.srv.cfg.SeedData = noModList{f.data}
	f.pass(now)
	if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted {
		t.Errorf("a stale reservation = %s, want aborted", got)
	}
	if got := f.state(t, "done.shop.example.com"); got != "gone" {
		t.Errorf("a rolled-back record = %s, want gone", got)
	}
	if !f.snapshotExists(t, orphan) {
		t.Error("an orphan was deleted with no last-modified time to judge it by")
	}
}

// A console with host claims armed runs no sweep: its record writes go
// through the claim, and every seed door refuses it.
func TestSweepOffOnAHostClaimConsole(t *testing.T) {
	f := newSweepFixture(t)
	now := time.Now()
	f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
	orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour)
	f.fx.srv.cfg.HostClaim = true
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	f.fx.srv.StartSeedsPass(ctx) // returns at once
	if ctx.Err() != nil {
		t.Fatal("StartSeedsPass ran on a host-claim console")
	}
	f.pass(now)
	if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, orphan) {
		t.Fatalf("a host-claim console swept: %s, orphan kept %v", got, f.snapshotExists(t, orphan))
	}
}

// hang blocks its first caller until ctx is done, as a stalled connection
// does; every later call goes through.
type hang struct{ done atomic.Bool }

func (h *hang) once(ctx context.Context) error {
	if h.done.Swap(true) {
		return nil
	}
	<-ctx.Done()
	return ctx.Err()
}

// hangingRecordWalk stalls the record walk's first listing.
type hangingRecordWalk struct {
	release.Store
	h *hang
}

// DeleteIfMatch keeps the finish's conditional record delete reachable
// through the wrapper.
func (s hangingRecordWalk) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.Store, key, etag)
}

func (s hangingRecordWalk) List(ctx context.Context, prefix string) ([]string, error) {
	if prefix == "accounts/" {
		if err := s.h.once(ctx); err != nil {
			return nil, err
		}
	}
	return s.Store.List(ctx, prefix)
}

// hangingData stalls the data bucket's first snapshot listing (list) or
// its first delete (!list).
type hangingData struct {
	release.Store
	h    *hang
	list bool
}

func (s hangingData) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	if s.list {
		if err := s.h.once(ctx); err != nil {
			return nil, err
		}
	}
	return s.Store.(release.ModLister).ListMod(ctx, prefix)
}

func (s hangingData) Delete(ctx context.Context, key string) error {
	if !s.list {
		if err := s.h.once(ctx); err != nil {
			return err
		}
	}
	return s.Store.Delete(ctx, key)
}

// A store call that never answers ends the pass at seedsPassTimeout, with
// nothing decided on what it could not finish; the next pass does the
// work. The delete case cuts the pass after its decisions, midway through
// its deletes, before its finishes.
func TestSweepPassIsBoundedByATimeout(t *testing.T) {
	old := seedsPassTimeout
	seedsPassTimeout = 100 * time.Millisecond
	t.Cleanup(func() { seedsPassTimeout = old })
	for name, stall := range map[string]func(f sweepFixture){
		"record walk":       func(f sweepFixture) { f.fx.srv.cfg.Releases = hangingRecordWalk{f.fx.rel, &hang{}} },
		"snapshot listing":  func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, true} },
		"a snapshot delete": func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, false} },
	} {
		t.Run(name, func(t *testing.T) {
			f := newSweepFixture(t)
			now := time.Now()
			f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
			orphans := []string{
				f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour),
				f.snapshot(t, "right.shop.example.com", newSeedID(t), 3*time.Hour),
			}
			f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now)
			stall(f)
			done := make(chan struct{})
			go func() { f.pass(now); close(done) }()
			select {
			case <-done:
			case <-time.After(10 * time.Second):
				t.Fatal("a hung store call stalled the pass")
			}
			if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack {
				t.Fatalf("a pass cut short finished a seed after its cut: %s", got)
			}
			f.pass(now)
			if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted {
				t.Errorf("the next pass left the stale reservation %s", got)
			}
			if got := f.state(t, "done.shop.example.com"); got != "gone" {
				t.Errorf("the next pass left the rolled-back record %s", got)
			}
			for _, key := range orphans {
				if f.snapshotExists(t, key) {
					t.Errorf("the next pass left the orphan %s", key)
				}
			}
		})
	}
}

// M6's rolled-back half on its own: a finish that cannot complete (a store
// this seed made, and no backend to tear it down with) leaves the record
// rolled-back, and the sweep still deletes the seed's snapshot.
func TestSweepDeletesARolledBackSeedsSnapshotWhenItsFinishWaits(t *testing.T) {
	f := finishFixture(t)
	id := newSeedID(t)
	f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack)
	f.markedStore(t, id)
	f.fx.srv.cfg.StoreProvisioner = nil
	key := seeds.SnapshotKey(fromHost, id)
	if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil {
		t.Fatal(err)
	}
	f.fx.srv.seedsPass(context.Background(), time.Now())
	if _, _, err := f.data.Get(context.Background(), key); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the rolled-back seed's snapshot: %v, want deleted", err)
	}
	rec, err := f.record(t, fromHost)
	if err != nil || rec.Seed.State != seeds.StateRolledBack {
		t.Fatalf("the record after a finish that could not complete: %+v, %v; want it kept rolled-back", rec, err)
	}
}

// recordWalks keeps the listings a record walk makes (accounts/ and apps/).
func recordWalks(lists []string) []string {
	var out []string
	for _, l := range lists {
		_, prefix, _ := strings.Cut(l, " ")
		if strings.HasPrefix(prefix, "accounts/") || strings.HasPrefix(prefix, "apps/") {
			out = append(out, l)
		}
	}
	return out
}

// Armed, a pass whose snapshot listing is empty has nothing to delete, so
// it makes no strict walk of the bucket: it reads the records from the
// record view, and a stale seed is still abandoned from what the view holds.
func TestSeedSweepSkipsWalkWithoutSnapshots(t *testing.T) {
	t.Parallel()
	f := newSweepFixture(t)
	cs := indexedDeployment(t, f.fx)
	now := time.Now()
	f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour))
	armRecordView(t, f.fx.srv, cs)
	cs.taken()
	f.pass(now)
	if walks := recordWalks(cs.taken()); len(walks) != 0 {
		t.Fatalf("a pass with no snapshot walked the records: %q", walks)
	}
	if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted {
		t.Fatalf("a stale reservation the view holds = %s, want aborted", got)
	}
}

// Armed, a pass that lists a snapshot keeps today's strict walk of the
// bucket, after the snapshot listing: a record written between the two
// listings that the view has not seen (here, a write that bumps no scope)
// still names its snapshot, so the snapshot is kept.
func TestSeedSweepWalksStrictWithSnapshots(t *testing.T) {
	f := newSweepFixture(t)
	bare := release.NewDirStore(f.fx.rel.(*deletableStore).root) // no change index: its writes bump nothing
	cs := indexedDeployment(t, f.fx)
	now := time.Now()
	host := "late.shop.example.com"
	id := newSeedID(t)
	key := f.snapshot(t, host, id, 3*time.Hour)
	armRecordView(t, f.fx.srv, cs)
	seedsPassBetweenListings = func() {
		cs.taken()
		rec := provision.Record{Schema: 1, Host: host, Slug: "late", App: sweepApp, Account: release.OpsAccount,
			Channel: "stable", CreatedAt: now.UTC(),
			Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "bob@example.com", At: now}}
		if err := provision.CreateRecord(context.Background(), bare, rec); err != nil {
			t.Error(err)
		}
	}
	t.Cleanup(func() { seedsPassBetweenListings = nil })
	f.pass(now)
	if walks := recordWalks(cs.taken()); len(walks) == 0 {
		t.Fatal("a pass with a snapshot made no record walk after the snapshot listing")
	}
	if !f.snapshotExists(t, key) {
		t.Fatal("the snapshot a record names was deleted: the pass read the view, not the bucket")
	}
}