package provision

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand/v2"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

// The RecordView invariant simulation (S3 change-index design §8). Random
// interleavings of writers (bumping, bypassing, failing their bump),
// incremental refreshes, backstop rebuilds and a moves-pass consumer run on a
// DirStore, and five invariants are checked after every step:
//
//	(a) after a Refresh that returned nil, every write whose bump succeeded
//	    and that completed before the Refresh's index poll is in the snapshot
//	    (or that key answers a read failure, which no consumer acts on);
//	(b) after a clean Rebuild (no read failed, no write landed during it),
//	    the record walk over the snapshot equals the record walk over the
//	    bucket, ListAllRecordsStrict included;
//	(c) published epochs only increase, the consumer never consumes a lower
//	    epoch than it already consumed, and no key ever reverts to an older
//	    write;
//	(d) a snapshot with Complete == false never answers ListAllRecordsStrict
//	    with a nil error when its gap lies in what that reader reads (an
//	    account, an instances listing, a record body); a gap anywhere else
//	    (a sidecar listing, a manifest) is still visible as a read error to
//	    that family's reader; and a Complete snapshot answers every reader
//	    without error;
//	(e) a scope whose read failed is read again by the next incremental
//	    Refresh, whether or not the index moved.
//
// How the interleavings are made. The view reads through simStore, which
// carries the current operation in its ctx and calls the simulation before
// every read: a read may fail by the operation's plan, and a trigger planned
// for "before the n-th read of scope S" fires there, writing to S, polling
// the shared watch as a sibling consumer would, or starting a concurrent
// Refresh. The concurrent Refresh runs nested, right there, if RecordView.mu
// is free (TryLock), which is exactly when a real concurrent caller would
// get in; otherwise it would block on mu, so it is queued and runs after the
// operation returns. Triggers fire in the reading goroutine, before the read,
// and only write to the scope being read, so where a write lands relative to
// that scope's reads does not depend on goroutine timing.
//
// A write that bumps and is still in flight (the one interleaving a store
// call cannot expose: between a write and its bump) is a DirStore.PutFrom
// whose body reader blocks on its first Read; refreshes run while it waits.
// A write whose bump fails is a real index-enabled DirStore whose own bump
// is refused through the production path: while its write is in progress
// the index file has a second hard link (outside the bucket), and safefile
// refuses to open any file with more than one link, for every user, root
// included. The bump's read of the index fails, so the bump gives up and the
// write stands. No other reader runs at that instant except a sibling poll
// fired from another reading goroutine, which then finds the index
// unusable and makes the next Refresh walk, as it should. Whether a write
// bumped is never assumed: the index is read before and after each write
// (writes are serialized), and the write is classed by what happened.
//
// The three interleavings §8 names are scripted subtests that run before the
// sweep, under both ListDirs shapes: codex 2 (the backstop has listed an app
// when a record appears and a concurrent Refresh starts), codex 4 (an app's
// re-read fails after the index moved, and the next index read is a 304) and
// codex 1 (a pass captures a snapshot and stalls while a later pass consumes
// a newer one). A failing seed reruns alone with -seed N.

var (
	simSeedFlag  = flag.Uint64("seed", 0, "TestRecordViewSimulation: run only this seed (0 runs the sweep)")
	simSeedsFlag = flag.Int("seeds", 50, "TestRecordViewSimulation: how many seeds the sweep runs")
)

// simConsumeHandedOver is the consumer's mutation switch. The moves pass
// refreshes and takes the current snapshot inside w.pass and never acts on a
// snapshot handed over from the Reconcile leg (§4.3, codex 1). true makes the
// simulated consumer act on the handed-over one; a mutation check flips it.
var simConsumeHandedOver = false

var simEpoch = time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)

var (
	simApps     = []string{"shop", "blog", "docs", "wiki", "store", "fresh"}
	simSlugs    = []string{"one", "two", "three"}
	simSidecars = []string{"bot", "cron"}
)

var errSimRead = errors.New("sim: read refused")

// simParallel is how many seeds run at once.
const simParallel = 8

// simTemplate is the fixture's bucket, read once and copied into a fresh
// root per seed (copying files is far cheaper than writing them through a
// store that fsyncs and bumps).
type simTemplate struct {
	files map[string][]byte
	acct  [3]string
}

func newSimTemplate(t *testing.T) *simTemplate {
	t.Helper()
	f := newViewFixture(t, false, true)
	keys, err := f.bypass.List(ctx, "")
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &simTemplate{files: map[string][]byte{}, acct: f.acct}
	for _, k := range keys {
		body, _, err := f.bypass.Get(ctx, k)
		if err != nil {
			t.Fatal(err)
		}
		tmpl.files[k] = body
	}
	if _, ok := tmpl.files[release.ChangeIndexKey]; !ok {
		t.Fatal("fixture: no change index")
	}
	return tmpl
}

// simOp is one view operation (a Refresh or a Rebuild) as the store sees it.
type simOp struct {
	kind     string
	nested   bool
	fail     map[string]bool // keys and prefixes whose every read fails
	triggers []*simTrigger

	// Filled in as it runs, under sim.mu.
	polled    bool
	pollSeq   int // writes completed when the op first read the index
	walked    bool
	counts    map[string]int
	read      map[string]bool // scopes it read
	failed    map[string]bool // scopes with a read that failed
	failedAny bool
	wrote     bool // a trigger wrote during it
	nestedRan bool // a concurrent Refresh ran inside it
	err       error
}

func newSimOp(kind string) *simOp {
	return &simOp{kind: kind, fail: map[string]bool{}, counts: map[string]int{}, read: map[string]bool{}, failed: map[string]bool{}}
}

type simTrigger struct {
	scope string
	n     int
	what  string
	act   func(op *simOp)
	fired bool
}

type simOpKey struct{}

// simWrite is one planned write: decided by the seeded generator up front,
// so a trigger firing in a reader goroutine never touches the rng.
type simWrite struct {
	w    *simWriter
	key  string
	del  bool
	soft bool // an account manifest written soft-deleted
}

type simWriter struct {
	name     string
	store    *release.DirStore
	failBump bool
}

type simVersion struct {
	seq  int
	body []byte // nil: absent
}

type simHandover struct {
	sn   *ViewSnapshot
	from int // step it was captured at
}

// simStats counts what a sweep exercised, so a simulation that silently
// stopped exercising something fails instead of passing.
type simStats struct {
	steps, writes, bumped, bypassed, failedBumps, paused     int
	refreshes, rebuilds, triggers, nested, deferred, sibling int
	readFailures, checksB, checksE, consumed, incomplete     int
}

func (a *simStats) add(b simStats) {
	a.steps += b.steps
	a.writes += b.writes
	a.bumped += b.bumped
	a.bypassed += b.bypassed
	a.failedBumps += b.failedBumps
	a.paused += b.paused
	a.refreshes += b.refreshes
	a.rebuilds += b.rebuilds
	a.triggers += b.triggers
	a.nested += b.nested
	a.deferred += b.deferred
	a.sibling += b.sibling
	a.readFailures += b.readFailures
	a.checksB += b.checksB
	a.checksE += b.checksE
	a.consumed += b.consumed
	a.incomplete += b.incomplete
}

type sim struct {
	t      *testing.T
	rng    *rand.Rand
	bucket bool
	acct   [3]string
	acctID map[string]int64

	root   string
	hold   string             // where holdIndex links the index
	live   *simStore          // what the view reads
	clean  *viewCountingStore // the bucket as a reader sees it, for (b)
	raw    *release.DirStore  // plain reads for bookkeeping
	f      *viewFixture       // for readWalk
	watch  *release.ChangeWatch
	v      *RecordView
	bumpW  *simWriter
	bypW   *simWriter
	failW  *simWriter
	keys   []string // every family key the simulation tracks
	step   int
	writes bool // ops may plan writes (false while a write is in flight)
	paused bool

	writeMu sync.Mutex // serializes writes, so a bump is attributable

	mu       sync.Mutex // guards the fields below
	seq      int        // writes started
	done     int        // writes completed (writes are serialized: done is the last seq)
	hist     map[string][]simVersion
	bumped   map[string][]int // key -> seqs of writes whose bump took
	dirty    map[string]bool  // scopes written since the last operation was planned
	deferred int
	errs     []string
	trace    []string
	stats    simStats

	checkMu  sync.Mutex
	lastEp   uint64
	lastVer  map[string]int
	carry    map[string]bool // (e): scopes whose last read failed
	pass     sync.Mutex      // the fake w.pass
	consumed uint64
	hand     []simHandover
}

func newSim(t *testing.T, tmpl *simTemplate, seed uint64, bucket bool) *sim {
	t.Helper()
	root := t.TempDir()
	for k, b := range tmpl.files {
		p := filepath.Join(root, filepath.FromSlash(k))
		if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(p, b, 0o644); err != nil {
			t.Fatal(err)
		}
	}
	s := &sim{
		t: t, rng: rand.New(rand.NewPCG(seed, 0x336)), bucket: bucket, acct: tmpl.acct, root: root,
		acctID: map[string]int64{}, writes: true,
		hist: map[string][]simVersion{}, bumped: map[string][]int{}, dirty: map[string]bool{}, lastVer: map[string]int{}, carry: map[string]bool{},
	}
	for i, a := range tmpl.acct {
		s.acctID[a] = int64(301 + i)
	}
	s.hold = filepath.Join(t.TempDir(), "changes.json")
	s.live = &simStore{DirStore: release.NewDirStore(root), bucket: bucket, sim: s}
	s.clean = &viewCountingStore{DirStore: release.NewDirStore(root), bucket: bucket}
	s.raw = release.NewDirStore(root)
	s.f = &viewFixture{live: s.clean, bypass: s.raw, acct: tmpl.acct}
	s.watch = release.NewChangeWatch(s.live)
	s.v = NewRecordView(s.live, s.watch)
	bw, fw := release.NewDirStore(root), release.NewDirStore(root)
	bw.EnableChangeIndex()
	fw.EnableChangeIndex()
	s.bumpW = &simWriter{name: "bumping", store: bw}
	s.failW = &simWriter{name: "failing-bump", store: fw, failBump: true}
	s.bypW = &simWriter{name: "bypassing", store: release.NewDirStore(root)}

	tracked := map[string]bool{}
	for _, a := range s.acct {
		tracked[accountManifestKey(a)] = true
		for _, b := range simApps {
			for _, slug := range simSlugs {
				tracked[RecordKey(a, b, slug+"."+b+".example.com")] = true
			}
			for _, n := range simSidecars {
				tracked[SidecarRecordKey(a, b, n)] = true
			}
			tracked[ManifestKey(a, b)] = true
			tracked[SidecarManifestKey(a, b)] = true
			tracked[release.AppTombstoneKey(a, b)] = true
		}
	}
	for k := range tmpl.files {
		if _, ok := (&ViewSnapshot{}).lookup(k); ok {
			tracked[k] = true
		}
	}
	for k := range tracked {
		s.keys = append(s.keys, k)
		var body []byte
		if b, ok := tmpl.files[k]; ok {
			body = b
		}
		s.hist[k] = []simVersion{{seq: 0, body: body}}
	}
	sort.Strings(s.keys)
	return s
}

func (s *sim) fail(format string, args ...any) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.errs = append(s.errs, fmt.Sprintf(format, args...))
}

func (s *sim) tracef(format string, args ...any) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.trace = append(s.trace, fmt.Sprintf("[%d] ", s.step)+fmt.Sprintf(format, args...))
}

// stop fails the test with the invariant breaches and the trace so far.
func (s *sim) stop() {
	s.t.Helper()
	s.mu.Lock()
	errs, trace := slices.Clone(s.errs), slices.Clone(s.trace)
	s.mu.Unlock()
	if len(errs) == 0 {
		return
	}
	if len(trace) > 80 {
		trace = trace[len(trace)-80:]
	}
	s.t.Fatalf("bucket=%v step %d:\n  %s\ntrace (last %d):\n  %s", s.bucket, s.step,
		strings.Join(errs, "\n  "), len(trace), strings.Join(trace, "\n  "))
}

// simScope is the change scope a read of target (a key or a listed prefix)
// belongs to, or "" for the top-level listings and the index.
func simScope(target string) string {
	if rest, ok := strings.CutPrefix(target, "accounts/"); ok {
		if a, _, ok := strings.Cut(rest, "/"); ok && a != "" {
			return "accounts/" + a
		}
		return ""
	}
	if sc, ok := release.ChangeScope(target); ok {
		return sc
	}
	return ""
}

// read runs before every read the view makes: it attributes the read to the
// operation in ctx, fires that operation's triggers for it, and fails it if
// the plan says so. Reads outside an operation (the checks) pass untouched.
func (s *sim) read(ctx context.Context, target string) error {
	op, _ := ctx.Value(simOpKey{}).(*simOp)
	if op == nil {
		return nil
	}
	s.mu.Lock()
	if target == release.ChangeIndexKey {
		if !op.polled {
			op.polled, op.pollSeq = true, s.done
		}
		fail := op.fail[target]
		if fail {
			op.failedAny = true
			s.stats.readFailures++
		}
		s.mu.Unlock()
		if fail {
			return errSimRead
		}
		return nil
	}
	if target == "accounts/" {
		op.walked = true
	}
	scope := simScope(target)
	var fire []*simTrigger
	if scope != "" {
		n := op.counts[scope]
		op.counts[scope] = n + 1
		for _, tr := range op.triggers {
			if !tr.fired && tr.scope == scope && tr.n == n {
				tr.fired = true
				fire = append(fire, tr)
			}
		}
	}
	s.mu.Unlock()
	for _, tr := range fire {
		s.tracef("  %s: before read %d of %s (%s): %s", op.kind, tr.n, scope, target, tr.what)
		s.mu.Lock()
		s.stats.triggers++
		s.mu.Unlock()
		tr.act(op)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	fail := op.fail[target]
	if scope != "" {
		op.read[scope] = true
		if fail {
			op.failed[scope] = true
		}
	}
	if fail {
		op.failedAny = true
		s.stats.readFailures++
		return errSimRead
	}
	return nil
}

// simStore is the bucket as the view reads it.
type simStore struct {
	*release.DirStore
	bucket bool
	sim    *sim
}

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

func (s *simStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if err := s.sim.read(ctx, key); err != nil {
		return nil, "", err
	}
	return s.DirStore.Get(ctx, key)
}

func (s *simStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	if err := s.sim.read(ctx, key); err != nil {
		return nil, "", false, err
	}
	return s.DirStore.GetIfChanged(ctx, key, etag)
}

func (s *simStore) List(ctx context.Context, prefix string) ([]string, error) {
	if err := s.sim.read(ctx, prefix); err != nil {
		return nil, err
	}
	return s.DirStore.List(ctx, prefix)
}

func (s *simStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	if err := s.sim.read(ctx, prefix); err != nil {
		return nil, err
	}
	return s.DirStore.ListETags(ctx, prefix)
}

func (s *simStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	if err := s.sim.read(ctx, prefix); err != nil {
		return nil, err
	}
	return s.DirStore.ListDirs(ctx, prefix)
}

func (s *simStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	if err := s.sim.read(ctx, prefix); err != nil {
		return nil, err
	}
	return s.DirStore.ListMod(ctx, prefix)
}

// ---- writers ----

func (s *sim) indexGen(scope string) uint64 {
	idx, _, err := release.ReadChangeIndex(ctx, s.raw)
	if err != nil {
		s.fail("sim: reading the index: %v", err)
		return 0
	}
	return idx.Entries[scope]
}

func (s *sim) body(w simWrite, seq int) []byte {
	var v any
	if a, ok := domains.AccountManifestSqid(w.key); ok {
		m := domains.AccountManifest{ID: s.acctID[a], Sqid: a, Name: fmt.Sprintf("Account v%d", seq), CreatedAt: simEpoch}
		if w.soft {
			m.DeletedAt = simEpoch.Add(time.Hour)
		}
		v = m
	} else {
		a, b, rest, _ := appKey(w.key)
		switch {
		case rest == "instances.json":
			v = Manifest{Schema: 1, App: b, Domain: fmt.Sprintf("v%d.%s.example.com", seq, b)}
		case rest == "sidecars.json":
			v = SidecarManifest{Schema: 1, App: b, Args: []string{"run"}, Channel: fmt.Sprintf("c%d", seq)}
		case rest == "deleted.json":
			v = release.AppTombstone{Schema: 1, Account: a, App: b, DeletedAt: simEpoch, Actor: fmt.Sprintf("t%d", seq)}
		case strings.HasPrefix(rest, "instances/"):
			host := strings.TrimSuffix(strings.TrimPrefix(rest, "instances/"), ".json")
			slug, _, _ := strings.Cut(host, ".")
			v = Record{Schema: 1, Host: host, Slug: slug, App: b, Account: a, Region: fmt.Sprintf("r%d", seq), Channel: "stable", CreatedAt: simEpoch}
		case strings.HasPrefix(rest, "sidecars/"):
			name := strings.TrimSuffix(strings.TrimPrefix(rest, "sidecars/"), ".json")
			v = SidecarRecord{Schema: 1, Name: name, App: b, Account: a, Host: SidecarHost(name, b, a), Channel: fmt.Sprintf("c%d", seq)}
		default:
			panic("sim: no body for " + w.key)
		}
	}
	body, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	return body
}

// begin registers a write before it touches the bucket, so a reader that
// sees it can always name it. Caller holds writeMu.
func (s *sim) begin(w simWrite) (seq int, body []byte, etag string, before uint64) {
	scope, _ := release.ChangeScope(w.key)
	before = s.indexGen(scope)
	if _, e, err := s.raw.Get(ctx, w.key); err == nil {
		etag = e
	}
	s.mu.Lock()
	s.seq++
	seq = s.seq
	s.mu.Unlock()
	if !w.del {
		body = s.body(w, seq)
	}
	s.mu.Lock()
	s.hist[w.key] = append(s.hist[w.key], simVersion{seq: seq, body: body})
	s.mu.Unlock()
	return seq, body, etag, before
}

// holdIndex gives the index file a second link, so the failing-bump
// writer's bump cannot read it (safefile.OpenFile refuses a file with more
// than one link). Caller holds writeMu.
func (s *sim) holdIndex() {
	if err := os.Link(filepath.Join(s.root, filepath.FromSlash(release.ChangeIndexKey)), s.hold); err != nil {
		s.fail("sim: holding the index: %v", err)
	}
}

// end classifies the write by what the index did, and completes it.
func (s *sim) end(w simWrite, seq int, before uint64, err error) {
	if w.w.failBump {
		if err := os.Remove(s.hold); err != nil && !errors.Is(err, os.ErrNotExist) {
			s.fail("sim: releasing the index: %v", err)
		}
	}
	if err != nil {
		s.fail("sim: %s write of %s: %v", w.w.name, w.key, err)
	}
	scope, _ := release.ChangeScope(w.key)
	after := s.indexGen(scope)
	s.mu.Lock()
	defer s.mu.Unlock()
	s.done = seq
	s.stats.writes++
	s.dirty[scope] = true
	class := "no bump"
	switch {
	case after == before+1:
		s.bumped[w.key] = append(s.bumped[w.key], seq)
		s.stats.bumped++
		class = fmt.Sprintf("bumped %s to %d", scope, after)
	case after != before:
		s.errs = append(s.errs, fmt.Sprintf("sim: write %d moved %s from %d to %d", seq, scope, before, after))
	case w.w.failBump:
		s.stats.failedBumps++
	default:
		s.stats.bypassed++
	}
	op := "put"
	if w.del {
		op = "delete"
	}
	s.trace = append(s.trace, fmt.Sprintf("[%d]   write %d: %s %s %s (%s)", s.step, seq, w.w.name, op, w.key, class))
}

func (s *sim) write(w simWrite) {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	seq, body, etag, before := s.begin(w)
	if w.w.failBump {
		s.holdIndex()
	}
	var err error
	if w.del {
		err = w.w.store.Delete(ctx, w.key)
	} else {
		_, err = w.w.store.Put(ctx, w.key, body, etag)
	}
	s.end(w, seq, before, err)
}

// simGate is a request body that holds the write at its first Read until
// released: the write has begun and is not yet in the bucket.
type simGate struct {
	r       *bytes.Reader
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (g *simGate) Read(p []byte) (int, error) {
	g.once.Do(func() {
		close(g.started)
		<-g.release
	})
	return g.r.Read(p)
}

func (g *simGate) Seek(off int64, whence int) (int64, error) { return g.r.Seek(off, whence) }

// pausedWrite streams a put whose body stalls, runs view operations while
// it is in flight, then lets it finish and runs the next pass.
func (s *sim) pausedWrite(w simWrite, during func()) {
	s.writeMu.Lock()
	seq, body, etag, before := s.begin(w)
	s.tracef("  write %d in flight: %s put %s", seq, w.w.name, w.key)
	g := &simGate{r: bytes.NewReader(body), started: make(chan struct{}), release: make(chan struct{})}
	done := make(chan error, 1)
	go func() {
		_, err := w.w.store.PutFrom(ctx, w.key, g, int64(len(body)), etag)
		done <- err
	}()
	var err error
	finished := false
	select {
	case <-g.started:
	case err = <-done:
		finished = true
	}
	if !finished {
		s.paused, s.writes = true, false
		s.mu.Lock()
		s.stats.paused++
		s.mu.Unlock()
		during()
		s.paused, s.writes = false, true
		if w.w.failBump {
			s.holdIndex() // only now: the refreshes above read the index as usual
		}
		close(g.release)
		err = <-done
	}
	s.end(w, seq, before, err)
	s.writeMu.Unlock()
}

// ---- view operations ----

// run is one view operation and the checks that follow every operation.
func (s *sim) run(op *simOp) error {
	cctx := context.WithValue(context.Background(), simOpKey{}, op)
	var err error
	if op.kind == "refresh" {
		err = s.v.Refresh(cctx)
	} else {
		err = s.v.Rebuild(cctx)
	}
	s.mu.Lock()
	op.err = err
	if op.kind == "refresh" {
		s.stats.refreshes++
	} else {
		s.stats.rebuilds++
	}
	s.mu.Unlock()
	sn := s.v.Snapshot()
	var ep uint64
	complete := false
	if sn != nil {
		ep, complete = sn.Epoch, sn.Complete
	}
	var fails []string
	for k := range op.fail {
		fails = append(fails, k)
	}
	sort.Strings(fails)
	s.tracef("  %s%s: err=%v epoch=%d complete=%v polled-at-write=%d failing=%v", map[bool]string{true: "nested "}[op.nested], op.kind, err, ep, complete, op.pollSeq, fails)
	if op.kind == "refresh" && err == nil && op.polled {
		s.checkVisible(op)
	}
	s.checkSnapshot()
	return err
}

// top runs an operation as a pass would, then the (b) and (e) checks, then
// any concurrent Refresh that had to wait for mu.
func (s *sim) top(op *simOp) error {
	carried := make(map[string]bool, len(s.carry))
	for sc := range s.carry {
		carried[sc] = true
	}
	err := s.run(op)

	// (e): an incremental Refresh reads again every scope whose last read
	// failed. A Refresh that walked read the whole deployment instead.
	if op.kind == "refresh" && !op.walked {
		for sc := range carried {
			s.mu.Lock()
			s.stats.checksE++
			s.mu.Unlock()
			if !op.read[sc] {
				s.fail("(e) %s failed its last read and this Refresh did not read it again", sc)
			}
		}
	}
	if op.walked && err == nil {
		// A walk that got past its top-level listings read the whole
		// deployment: a scope it did not reach has no keys left to read.
		clear(s.carry)
	}
	for sc := range op.read {
		if !op.failed[sc] {
			delete(s.carry, sc)
		}
	}
	for sc := range op.failed {
		s.carry[sc] = true
	}

	if op.kind == "rebuild" && err == nil && !op.failedAny && !op.wrote && !op.nestedRan && !s.paused {
		s.checkMatchesLive()
	}

	for {
		s.mu.Lock()
		n := s.deferred
		s.deferred = 0
		s.mu.Unlock()
		if n == 0 {
			break
		}
		for range n {
			s.mu.Lock()
			s.stats.deferred++
			s.mu.Unlock()
			s.top(newSimOp("refresh"))
		}
	}
	return err
}

// concurrent is a Refresh started by another caller at this instant. It gets
// in only if mu is free; otherwise it waits for the operation holding mu.
func (s *sim) concurrent(outer *simOp) {
	if s.v.mu.TryLock() {
		s.v.mu.Unlock()
		s.mu.Lock()
		outer.nestedRan = true
		s.stats.nested++
		s.mu.Unlock()
		op := newSimOp("refresh")
		op.nested = true
		s.run(op)
		return
	}
	s.mu.Lock()
	s.deferred++
	s.mu.Unlock()
}

func (s *sim) siblingPoll() {
	s.mu.Lock()
	s.stats.sibling++
	s.mu.Unlock()
	_, _ = s.watch.Poll(context.Background())
}

// ---- checks ----

// version is the newest write whose value the answer shows: a body names
// its write uniquely, and an absence is credited to the newest delete.
func (s *sim) version(key string, body []byte, absent bool) (int, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	best := -1
	for _, v := range s.hist[key] {
		if (absent && v.body == nil) || (!absent && v.body != nil && bytes.Equal(v.body, body)) {
			best = max(best, v.seq)
		}
	}
	return best, best >= 0
}

func (s *sim) answer(st release.Store, key string) (body []byte, absent, failed bool) {
	body, _, err := st.Get(context.Background(), key)
	switch {
	case err == nil:
		return body, false, false
	case errors.Is(err, os.ErrNotExist):
		return nil, true, false
	}
	return nil, false, true
}

// checkVisible is (a).
func (s *sim) checkVisible(op *simOp) {
	sn := s.v.Snapshot()
	if sn == nil {
		s.fail("(a) a Refresh returned nil and published nothing")
		return
	}
	st := sn.Store(s.live)
	s.mu.Lock()
	need := map[string]int{}
	for key, seqs := range s.bumped {
		for _, q := range seqs {
			if q <= op.pollSeq {
				need[key] = max(need[key], q)
			}
		}
	}
	s.mu.Unlock()
	for key, q := range need {
		body, absent, failed := s.answer(st, key)
		if failed {
			continue // a cached read failure: no consumer acts on it
		}
		ver, ok := s.version(key, body, absent)
		switch {
		case !ok:
			s.fail("(a) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body)
		case ver < q:
			s.fail("(a) epoch %d: Refresh polled after write %d (bumped) completed, but %s shows write %d", sn.Epoch, q, key, ver)
		}
	}
}

// checkSnapshot is (c) and (d) over the current snapshot, once per epoch.
func (s *sim) checkSnapshot() {
	s.checkMu.Lock()
	defer s.checkMu.Unlock()
	sn := s.v.Snapshot()
	if sn == nil {
		return
	}
	if sn.Epoch < s.lastEp {
		s.fail("(c) published epoch went from %d back to %d", s.lastEp, sn.Epoch)
		return
	}
	if sn.Epoch == s.lastEp {
		return
	}
	s.lastEp = sn.Epoch
	if !sn.Complete {
		s.mu.Lock()
		s.stats.incomplete++
		s.mu.Unlock()
	}
	st := sn.Store(s.live)
	bg := context.Background()
	famErr := ""
	for _, key := range s.keys {
		body, absent, failed := s.answer(st, key)
		if failed {
			if famErr == "" {
				famErr = "Get " + key
			}
			continue
		}
		ver, ok := s.version(key, body, absent)
		if !ok {
			s.fail("(c) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body)
			continue
		}
		if ver < s.lastVer[key] {
			s.fail("(c) epoch %d: %s went back from write %d to write %d", sn.Epoch, key, s.lastVer[key], ver)
		}
		s.lastVer[key] = ver
	}
	for _, a := range s.acct {
		if _, err := st.List(bg, "accounts/"+a+"/"); err != nil && famErr == "" {
			famErr = "List accounts/" + a + "/"
		}
		if _, err := st.ListDirs(bg, "apps/"+a+"/"); err != nil && famErr == "" {
			famErr = "ListDirs apps/" + a + "/"
		}
		for _, b := range simApps {
			for _, p := range []string{RecordsPrefix(a, b), SidecarsPrefix(a, b)} {
				if _, err := st.List(bg, p); err != nil && famErr == "" {
					famErr = "List " + p
				}
			}
		}
	}
	if _, rf, err := knownAccounts(bg, st); (err != nil || rf) && famErr == "" {
		famErr = "knownAccounts"
	}
	_, _, strictErr := ListAllRecordsStrict(bg, st)
	switch {
	case sn.Complete && strictErr != nil:
		s.fail("(d) epoch %d is Complete but ListAllRecordsStrict over it fails: %v", sn.Epoch, strictErr)
	case sn.Complete && famErr != "":
		s.fail("(d) epoch %d is Complete but %s over it fails", sn.Epoch, famErr)
	case !sn.Complete && strictErr == nil && recordWalkGap(sn):
		s.fail("(d) epoch %d is not Complete, its gap is in the record walk, and ListAllRecordsStrict over it returned nil", sn.Epoch)
	case !sn.Complete && strictErr == nil && famErr == "":
		s.fail("(d) epoch %d is not Complete, yet every reader over it answers as if it were", sn.Epoch)
	}
}

// recordWalkGap reports a gap in what ListAllRecordsStrict reads: the
// top-level walk, an account it would know (or fail to know), a known
// account's app directories, an instances listing or a record body.
func recordWalkGap(sn *ViewSnapshot) bool {
	if sn.walkErr != nil {
		return true
	}
	known, readFailed, err := knownAccounts(context.Background(), sn.Store(uncachedStore{}))
	if err != nil || readFailed {
		return true
	}
	for a := range known {
		if sn.dirErrs[a] != nil {
			return true
		}
		for _, e := range sn.apps[a] {
			if e.instErr != nil {
				return true
			}
			for _, ke := range e.inst {
				if e.objs[ke.Key].failed() {
					return true
				}
			}
		}
	}
	return false
}

// checkMatchesLive is (b): after a clean Rebuild with nothing written during
// it, the readers over the snapshot answer exactly as over the bucket.
func (s *sim) checkMatchesLive() {
	s.stats.checksB++
	sn := s.v.Snapshot()
	if !sn.Complete {
		s.fail("(b) a clean Rebuild published epoch %d not Complete", sn.Epoch)
		return
	}
	lr, lu, lerr := ListAllRecordsStrict(ctx, s.clean)
	sr, su, serr := ListAllRecordsStrict(ctx, sn.Store(s.live))
	if lerr != nil || serr != nil {
		s.fail("(b) ListAllRecordsStrict: bucket %v, snapshot %v", lerr, serr)
		return
	}
	norm := func(rs []Record) []Record {
		for i := range rs {
			rs[i].CreatedAt = rs[i].CreatedAt.UTC()
		}
		return rs
	}
	if !reflect.DeepEqual(norm(lr), norm(sr)) || !reflect.DeepEqual(lu, su) {
		s.fail("(b) ListAllRecordsStrict over epoch %d:\n%+v %v\nover the bucket:\n%+v %v", sn.Epoch, sr, su, lr, lu)
		return
	}
	if want, got := readWalk(s.t, s.clean, s.f), readWalk(s.t, sn.Store(s.live), s.f); !reflect.DeepEqual(got, want) {
		s.fail("(b) the record walk over epoch %d:\n%+v\nover the bucket:\n%+v", sn.Epoch, got, want)
	}
}

// ---- the consumer ----

// consumerBegin is a pass's Reconcile leg: refresh, take the snapshot, and
// hand it over towards w.pass.
func (s *sim) consumerBegin(op *simOp) {
	s.top(op)
	if sn := s.v.Snapshot(); sn != nil {
		s.hand = append(s.hand, simHandover{sn: sn, from: s.step})
		s.tracef("  pass begins with epoch %d handed over", sn.Epoch)
	}
}

// consumerFinish is the moves pass reaching w.pass: it refreshes and takes
// the current snapshot under it, records the epoch it consumed and refuses a
// lower one. simConsumeHandedOver makes it use the handed-over snapshot.
func (s *sim) consumerFinish(i int, op *simOp) {
	h := s.hand[i]
	s.hand = slices.Delete(s.hand, i, i+1)
	s.pass.Lock()
	defer s.pass.Unlock()
	sn, err := h.sn, error(nil)
	if !simConsumeHandedOver {
		err = s.top(op)
		sn = s.v.Snapshot()
	}
	if sn == nil {
		return
	}
	s.tracef("  moves pass (begun at step %d) consumes epoch %d (complete=%v, err=%v)", h.from, sn.Epoch, sn.Complete, err)
	s.stats.consumed++
	if sn.Epoch < s.consumed {
		s.fail("(c) the moves pass consumed epoch %d after it had consumed %d", sn.Epoch, s.consumed)
	}
	s.consumed = max(s.consumed, sn.Epoch)
	if err == nil && sn.Complete {
		if _, _, err := ListAllRecordsStrict(context.Background(), sn.Store(s.live)); err != nil {
			s.fail("(d) the moves pass acted on Complete epoch %d and the strict reader failed: %v", sn.Epoch, err)
		}
	}
}

// ---- the generator ----

func (s *sim) pickWriter() *simWriter {
	switch r := s.rng.IntN(10); {
	case r < 6:
		return s.bumpW
	case r < 8:
		return s.bypW
	}
	return s.failW
}

func (s *sim) pickScope() string {
	a := s.acct[s.rng.IntN(len(s.acct))]
	if s.rng.IntN(100) < 15 {
		return "accounts/" + a
	}
	return "apps/" + a + "/" + simApps[s.rng.IntN(len(simApps))]
}

func (s *sim) pickWrite(scope string, w *simWriter, puts bool) simWrite {
	if a, ok := scopeAccount(scope); ok {
		return simWrite{w: w, key: accountManifestKey(a), soft: s.rng.IntN(100) < 15}
	}
	a, b, _ := ScopeApp(scope)
	slug := simSlugs[s.rng.IntN(len(simSlugs))]
	name := simSidecars[s.rng.IntN(len(simSidecars))]
	r := s.rng.IntN(20)
	if puts && r >= 8 && r < 11 {
		r = 0
	}
	switch {
	case r < 8:
		return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com")}
	case r < 11:
		return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com"), del: true}
	case r < 14:
		return simWrite{w: w, key: SidecarRecordKey(a, b, name), del: !puts && r == 13}
	case r < 16:
		return simWrite{w: w, key: ManifestKey(a, b)}
	case r < 17:
		return simWrite{w: w, key: SidecarManifestKey(a, b)}
	}
	return simWrite{w: w, key: release.AppTombstoneKey(a, b), del: !puts && r == 19}
}

func (s *sim) pickFailTarget() string {
	a := s.acct[s.rng.IntN(len(s.acct))]
	b := simApps[s.rng.IntN(len(simApps))]
	switch r := s.rng.IntN(100); {
	case r < 70:
		slug := simSlugs[s.rng.IntN(len(simSlugs))]
		return []string{
			RecordsPrefix(a, b), SidecarsPrefix(a, b), RecordKey(a, b, slug+"."+b+".example.com"),
			SidecarRecordKey(a, b, simSidecars[s.rng.IntN(len(simSidecars))]),
			ManifestKey(a, b), SidecarManifestKey(a, b), release.AppTombstoneKey(a, b),
		}[s.rng.IntN(7)]
	case r < 85:
		return []string{"accounts/" + a + "/", accountManifestKey(a)}[s.rng.IntN(2)]
	}
	return []string{"accounts/", "apps/", "apps/" + a + "/", release.ChangeIndexKey}[s.rng.IntN(4)]
}

// triggerScope picks where a trigger waits. An incremental Refresh reads only
// the scopes that moved or failed, so those are favoured; a Rebuild reads
// every scope.
func (s *sim) triggerScope(kind string) string {
	s.mu.Lock()
	var hot []string
	for sc := range s.dirty {
		hot = append(hot, sc)
	}
	s.mu.Unlock()
	for sc := range s.carry {
		if !slices.Contains(hot, sc) {
			hot = append(hot, sc)
		}
	}
	sort.Strings(hot)
	if kind == "refresh" && len(hot) > 0 && s.rng.IntN(100) < 70 {
		return hot[s.rng.IntN(len(hot))]
	}
	return s.pickScope()
}

// plan draws an operation: read failures, and triggers that write into the
// scope being read, start a concurrent Refresh, or poll the shared watch.
func (s *sim) plan(kind string) *simOp {
	op := newSimOp(kind)
	if s.rng.IntN(100) < 30 {
		for range 1 + s.rng.IntN(2) {
			op.fail[s.pickFailTarget()] = true
		}
	}
	defer func() {
		s.mu.Lock()
		clear(s.dirty)
		s.mu.Unlock()
	}()
	for range s.rng.IntN(4) {
		tr := &simTrigger{scope: s.triggerScope(kind), n: s.rng.IntN(6)}
		k := s.rng.IntN(20)
		if !s.writes && k < 15 {
			k = 15 + s.rng.IntN(5)
		}
		switch {
		case k < 12:
			wr := s.pickWrite(tr.scope, s.pickWriter(), false)
			tr.what = fmt.Sprintf("%s writes %s", wr.w.name, wr.key)
			tr.act = func(o *simOp) { s.hookWrite(o, wr) }
		case k < 15:
			// The codex-2 shape: a write lands behind the reader, and a
			// concurrent Refresh starts at once.
			wr := s.pickWrite(tr.scope, s.bumpW, true)
			tr.what = fmt.Sprintf("bumping writes %s, then a concurrent Refresh", wr.key)
			tr.act = func(o *simOp) { s.hookWrite(o, wr); s.concurrent(o) }
		case k < 18:
			tr.what = "a concurrent Refresh"
			tr.act = s.concurrent
		default:
			tr.what = "a sibling polls the shared watch"
			tr.act = func(*simOp) { s.siblingPoll() }
		}
		op.triggers = append(op.triggers, tr)
	}
	return op
}

func (s *sim) hookWrite(o *simOp, wr simWrite) {
	s.mu.Lock()
	o.wrote = true
	s.mu.Unlock()
	s.write(wr)
}

// viewStep draws one view-side step: a Refresh, a Rebuild, or one leg of a
// moves pass.
func (s *sim) viewStep() {
	switch r := s.rng.IntN(100); {
	case r < 45:
		s.tracef("refresh")
		s.top(s.plan("refresh"))
	case r < 60:
		s.tracef("rebuild")
		s.top(s.plan("rebuild"))
	case r < 75:
		s.tracef("pass begins")
		s.consumerBegin(s.plan("refresh"))
	case r < 90 && len(s.hand) > 0:
		s.tracef("moves pass reaches w.pass")
		s.consumerFinish(s.rng.IntN(len(s.hand)), s.plan("refresh"))
	default:
		s.tracef("quiet refresh")
		s.top(newSimOp("refresh"))
	}
}

func (s *sim) randomStep() {
	switch r := s.rng.IntN(100); {
	case r < 22:
		wr := s.pickWrite(s.pickScope(), s.pickWriter(), false)
		s.tracef("write")
		s.write(wr)
	case r < 30:
		w := s.bumpW
		if s.rng.IntN(10) < 3 {
			w = s.pickWriter()
		}
		wr := s.pickWrite(s.pickScope(), w, true)
		s.tracef("write in flight")
		s.pausedWrite(wr, func() {
			s.tracef("refresh while a write is in flight")
			s.top(s.plan("refresh"))
			if s.rng.IntN(2) == 0 {
				s.viewStep()
			}
		})
		s.tracef("refresh after the write completed")
		s.top(s.plan("refresh"))
	case r < 33:
		s.tracef("sibling poll")
		s.siblingPoll()
	default:
		s.viewStep()
	}
}

func runSimSeed(t *testing.T, tmpl *simTemplate, seed uint64) simStats {
	s := newSim(t, tmpl, seed, seed%2 == 0)
	steps := 20 + s.rng.IntN(15)
	for s.step = 1; s.step <= steps; s.step++ {
		s.randomStep()
		s.stats.steps++
		s.stop()
	}
	// Quiesce: a clean backstop, which (b) checks, then a clean refresh.
	s.tracef("final rebuild")
	s.top(newSimOp("rebuild"))
	s.tracef("final refresh")
	s.top(newSimOp("refresh"))
	s.stop()
	return s.stats
}

// ---- the named interleavings (§8) ----

func simScripted(t *testing.T, tmpl *simTemplate) {
	for _, bucket := range []bool{true, false} {
		// codex 2: the backstop lists an app before a record appears, and an
		// incremental refresh sees the record. The refresh must wait for the
		// backstop and publish after it, so the record is never published
		// and then lost.
		t.Run(fmt.Sprintf("codex2-backstop-vs-refresh/bucket=%v", bucket), func(t *testing.T) {
			s := newSim(t, tmpl, 1, bucket)
			s.top(newSimOp("rebuild"))
			a := s.acct[0]
			key := RecordKey(a, "shop", "three.shop.example.com")
			op := newSimOp("rebuild")
			op.triggers = []*simTrigger{{scope: "apps/" + a + "/shop", n: 1, what: "record appears, concurrent Refresh",
				act: func(o *simOp) { s.hookWrite(o, simWrite{w: s.bumpW, key: key}); s.concurrent(o) }}}
			s.top(op)
			s.stop()
			if !op.triggers[0].fired {
				t.Fatal("the trigger never fired")
			}
			if body, absent, failed := s.answer(s.v.Snapshot().Store(s.live), key); absent || failed || body == nil {
				t.Fatalf("the record is missing after the backstop and the refresh behind it (absent %v, failed %v)", absent, failed)
			}
		})
		// codex 4: an app's re-read fails after the index moved, and the
		// next index read is a 304. The next Refresh must read it again.
		t.Run(fmt.Sprintf("codex4-failed-reread-then-304/bucket=%v", bucket), func(t *testing.T) {
			s := newSim(t, tmpl, 1, bucket)
			s.top(newSimOp("rebuild"))
			a := s.acct[0]
			key := RecordKey(a, "shop", "three.shop.example.com")
			s.write(simWrite{w: s.bumpW, key: key})
			op := newSimOp("refresh")
			op.fail[RecordsPrefix(a, "shop")] = true
			s.top(op)
			if !op.failed["apps/"+a+"/shop"] {
				t.Fatal("the re-read did not fail")
			}
			_, before, _ := s.raw.Get(ctx, release.ChangeIndexKey)
			again := newSimOp("refresh")
			s.top(again)
			s.stop()
			if _, after, _ := s.raw.Get(ctx, release.ChangeIndexKey); after != before {
				t.Fatal("the index moved: the second poll was not a 304")
			}
			if !again.read["apps/"+a+"/shop"] {
				t.Fatal("the failed scope was not read again")
			}
			sn := s.v.Snapshot()
			if body, absent, failed := s.answer(sn.Store(s.live), key); absent || failed || body == nil || !sn.Complete {
				t.Fatalf("after the retry: absent %v, failed %v, complete %v", absent, failed, sn.Complete)
			}
		})
		// codex 1: a pass captures a snapshot, stalls; another pass consumes
		// a newer one; the first then reaches w.pass. It must not consume an
		// older epoch than the second did.
		t.Run(fmt.Sprintf("codex1-stalled-pass/bucket=%v", bucket), func(t *testing.T) {
			s := newSim(t, tmpl, 1, bucket)
			s.top(newSimOp("rebuild"))
			s.consumerBegin(newSimOp("refresh"))
			s.write(simWrite{w: s.bumpW, key: RecordKey(s.acct[1], "docs", "three.docs.example.com")})
			s.consumerBegin(newSimOp("refresh"))
			if s.hand[1].sn.Epoch <= s.hand[0].sn.Epoch {
				t.Fatalf("fixture: handovers at epochs %d and %d", s.hand[0].sn.Epoch, s.hand[1].sn.Epoch)
			}
			s.consumerFinish(1, newSimOp("refresh"))
			s.consumerFinish(0, newSimOp("refresh"))
			s.stop()
		})
	}
}

func TestRecordViewSimulation(t *testing.T) {
	if testing.Short() {
		t.Skip("the invariant simulation is not run under -short")
	}
	// The view and the bump log every failed read and refused bump; the
	// simulation makes thousands on purpose. Serial: it sets the log output.
	out := log.Writer()
	log.SetOutput(io.Discard)
	t.Cleanup(func() { log.SetOutput(out) })
	tmpl := newSimTemplate(t)

	t.Run("scripted", func(t *testing.T) { simScripted(t, tmpl) })

	first, n := uint64(1), *simSeedsFlag
	if *simSeedFlag != 0 {
		first, n = *simSeedFlag, 1
	}
	var (
		mu       sync.Mutex
		total    simStats
		ran      int
		failSeed uint64
		next     atomic.Uint64
		stop     atomic.Bool
	)
	next.Store(first)
	start := time.Now()
	t.Run("sweep", func(t *testing.T) {
		// A seed spends most of its time waiting for DirStore's fsyncs, so a
		// few run at once. Each seed is its own bucket, view and writers; the
		// sweep stops starting seeds at the first failure.
		var wg sync.WaitGroup
		for range simParallel {
			wg.Go(func() {
				for !stop.Load() {
					seed := next.Add(1) - 1
					if seed >= first+uint64(n) {
						return
					}
					var st simStats
					ok := t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { st = runSimSeed(t, tmpl, seed) })
					mu.Lock()
					if ok {
						total.add(st)
						ran++
					} else if failSeed == 0 || seed < failSeed {
						failSeed = seed
						stop.Store(true)
					}
					mu.Unlock()
				}
			})
		}
		wg.Wait()
		if failSeed != 0 {
			t.Logf("first failing seed: %d (rerun with -seed %d)", failSeed, failSeed)
		}
	})
	t.Logf("%d seeds passed in %v: %+v", ran, time.Since(start).Round(time.Millisecond), total)
	if t.Failed() || ran < 50 {
		return
	}
	// A simulation that stopped exercising something would pass silently.
	exercised := map[string]int{
		"bumped writes": total.bumped, "bypassing writes": total.bypassed, "failed bumps": total.failedBumps,
		"writes in flight": total.paused, "triggers": total.triggers, "concurrent refreshes that waited": total.deferred,
		"sibling polls": total.sibling, "read failures": total.readFailures, "(b) checks": total.checksB,
		"(e) checks": total.checksE, "consumed snapshots": total.consumed, "incomplete snapshots": total.incomplete,
		"rebuilds": total.rebuilds,
	}
	if runtime.GOOS != "linux" {
		// Only the Linux safefile refuses a file with two links, which is
		// how a bump is made to fail; elsewhere those writes bump.
		delete(exercised, "failed bumps")
		t.Logf("failed bumps are not exercised on %s", runtime.GOOS)
	}
	for name, c := range exercised {
		if c == 0 {
			t.Errorf("the sweep exercised no %s", name)
		}
	}
}