package hostlife

import (
	"fmt"
	"hash/fnv"
	"regexp"
	"sort"
	"strings"
)

// The database and its replica, with the pinned litestream's LTX
// behaviour (rev 7.6, R4-1, R5-1 to R5-3; the facts in
// internal/sim/litestreamfacts).
//
// A database is a map of keys to values with two counters: the lineage
// transaction id (TX), the oracle's order of acknowledged writes, and the
// replication position (Pos), litestream's TXID, which counts shipped LTX
// files and is what every LTX range is numbered in. A replica object is an
// LTX file: a level (0 and 1, 9 for a snapshot), a range in replication
// numbering, the final page image of every key it touches, and, for the
// oracle only, the writes it covers. Pinned litestream writes no database
// checksum into any file (R5-2): a file carries a CRC-64 of itself, which
// the model's files never fail, and nothing else; restore verifies by
// decoding, not by a state checksum.
//
// Replication (R5-1): each shipped file carries one TXID, Pos+1; a
// database with position 0 ships a full-page level-0 file [1, 1] first,
// as litestream's first sync does. Litestream takes a fresh box's position
// from the newest remote level-0 file, and from nothing else: a sidecar is
// follow-mode only, and a snapshot does not count. So a cold restore
// writes a level-0 anchor [cur, cur] with every page before replication
// starts, and the position is cur; without it the position is the newest
// remote level-0 file's, or 0, and replication restarts below the
// snapshot, where the planner never looks.
//
// Rotation and retention (R5-3): the executor's litestream compacts three
// level-0 files into one level-1 file (deleting the inputs), writes a new
// snapshot after each compaction, keeps only the newest snapshot, and
// expires level-1 and level-0 files whose range ends below the previous
// snapshot's TXID. A restore roots at the newest current snapshot of the
// generation at or past the pinned one, and falls back to the pinned
// version only when no successor is current.

type tx struct {
	W   int    // the global write id, which is what the application acknowledged
	TX  int    // the lineage's transaction id; 0 if the write never reached a database
	Key string // the key written
	Val int    // the value; 0 is a delete
}

// sum is a digest of the state, for the model's own digests (stop
// evidence, the settle check); no replica file carries it (R5-2).
func sum(state map[string]int) string {
	keys := make([]string, 0, len(state))
	for k, v := range state {
		if v != 0 {
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)
	h := fnv.New64a()
	for _, k := range keys {
		fmt.Fprintf(h, "%s=%d;", k, state[k])
	}
	return fmt.Sprintf("%016x", h.Sum64())
}

func cloneState(m map[string]int) map[string]int {
	out := make(map[string]int, len(m))
	for k, v := range m {
		if v != 0 {
			out[k] = v
		}
	}
	return out
}

func statesEqual(a, b map[string]int) (string, bool) {
	for k, v := range a {
		if v != 0 && b[k] != v {
			return fmt.Sprintf("key %s expected %d, got %d", k, v, b[k]), false
		}
	}
	for k, v := range b {
		if v != 0 && a[k] != v {
			return fmt.Sprintf("key %s expected %d, got %d", k, a[k], v), false
		}
	}
	return "", true
}

// commit applies acknowledged write w to db as a transaction: mostly an
// insert of a new key named by the write; every fifth write updates the
// newest live key, every fifth-but-one deletes the oldest, so restore
// fidelity depends on final page images, not on a list of ids. It records
// the transaction for the oracle.
func (w *world) commit(db *localDB, wr int) {
	if db.State == nil {
		db.State = map[string]int{}
	}
	t := tx{W: wr, TX: db.TX + 1, Key: fmt.Sprintf("k%d", wr), Val: wr}
	switch wr % 5 {
	case 3:
		if k := newestKey(db.State); k != "" {
			t.Key, t.Val = k, wr
		}
	case 4:
		if k := oldestKey(db.State); k != "" {
			t.Key, t.Val = k, 0
		}
	}
	if t.Val == 0 {
		delete(db.State, t.Key)
	} else {
		db.State[t.Key] = t.Val
	}
	db.TX++
	db.Pending = append(db.Pending, wr)
	db.Log = append(db.Log, wr)
	w.txs[wr] = t
}

func keyID(k string) int {
	n := 0
	fmt.Sscanf(k, "k%d", &n)
	return n
}

func newestKey(state map[string]int) string {
	best, id := "", 0
	for k, v := range state {
		if v != 0 && keyID(k) > id {
			best, id = k, keyID(k)
		}
	}
	return best
}

func oldestKey(state map[string]int) string {
	best, id := "", 0
	for k, v := range state {
		if v != 0 && (best == "" || keyID(k) < id) {
			best, id = k, keyID(k)
		}
	}
	return best
}

// ltxKey is the pinned S3 client's layout under a generation prefix:
// <prefix>/<level as four hex digits>/<min>-<max>.ltx, sixteen hex digits
// each (litestream s3/replica_client.go, ltx.FormatFilename). A snapshot
// is level 9 from TXID 1.
func ltxKey(loc, gen string, level, min, max int) string {
	return fmt.Sprintf("%s%04x/%016x-%016x.ltx", dataKey(loc, gen, ""), level, min, max)
}

func snapKey(loc, gen string, max int) string { return ltxKey(loc, gen, 9, 1, max) }

var ltxKeyRe = regexp.MustCompile(`^(.*/)([0-9a-f]{4})/([0-9a-f]{16})-([0-9a-f]{16})\.ltx$`)

// ltxLevel parses the level and range from an LTX key; ok is false for
// any other key.
func ltxLevel(k string) (level, min, max int, ok bool) {
	m := ltxKeyRe.FindStringSubmatch(k)
	if m == nil {
		return 0, 0, 0, false
	}
	fmt.Sscanf(m[2], "%x", &level)
	fmt.Sscanf(m[3], "%x", &min)
	fmt.Sscanf(m[4], "%x", &max)
	return level, min, max, true
}

// genPrefixOf is the generation prefix an LTX key belongs to: the key
// without its level directory and file name.
func genPrefixOf(k string) string {
	m := ltxKeyRe.FindStringSubmatch(k)
	if m == nil {
		return ""
	}
	return m[1]
}

// isLevelDir: a four-hex-digit child is litestream's level directory.
func isLevelDir(child string) bool {
	if len(child) != 4 {
		return false
	}
	for _, r := range child {
		if !(r >= '0' && r <= '9' || r >= 'a' && r <= 'f') {
			return false
		}
	}
	return true
}

// snapshotOf is the snapshot file of a database at its replication
// position: every page, TXIDs 1 to Pos, and the oracle's list of the
// writes it holds. The caller ships pending writes first, as litestream
// syncs before it snapshots.
// snapshotOf is the snapshot of a database as a seal or an installer
// encodes it: [1, N], N the replication position. A database that holds
// transactions but has never replicated (position 0: a warm database
// whose writes are all pending, a source sealed at its first stay) is
// encoded at TXID 1, the TXID litestream's first sync gives its full-page
// file; only a database with no transaction at all is empty, and the
// encoder refuses a file for it (R8-3; seed 19023 / 30,000 took such a
// seal for empty and lost the source's first write)
func snapshotOf(db *localDB, inc int) *blob {
	n := db.Pos
	if n == 0 && (len(db.Log) > 0 || len(db.State) > 0) {
		n = 1
	}
	return &blob{Kind: "snap", Level: 9, Min: 1, Max: n, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true,
		Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)}
}

// deltaOf is the next level-0 file: one TXID, Pos+1, with the final
// image of every key the pending writes touched. At position 0 it is the
// first sync, a full-page file [1, 1], as litestream writes it.
func (w *world) deltaOf(db *localDB, inc int) *blob {
	pages := map[string]int{}
	full := db.Pos == 0 || db.FirstFull
	if full {
		pages = cloneState(db.State)
	}
	for _, wr := range db.Pending {
		t := w.txs[wr]
		pages[t.Key] = db.State[t.Key] // the image now: 0 when deleted since
	}
	// the oracle's list of the writes the file carries: a full file carries
	// the effect of every write applied here, not only the pending ones
	txs := db.Pending
	if full {
		txs = db.Log
	}
	return &blob{Kind: "ltx", Level: 0, Min: db.Pos + 1, Max: db.Pos + 1, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: full,
		Pages: pages, TXs: append([]int(nil), txs...)}
}

// anchorOf is the level-0 anchor a cold restore writes before replication
// starts (R5-1, the fact in litestreamfacts TestFact_R5_1_AnchorL0ContinuesFromN):
// every page of the restored database as the file [cur, cur], which
// litestream's open takes as its position.
func anchorOf(db *localDB, inc int) *blob {
	return &blob{Kind: "ltx", Level: 0, Min: db.Pos, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true,
		Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)}
}

// apply overlays an LTX file's page images onto state; a full file replaces
// it, so a key the full file does not carry is gone.
func apply(state map[string]int, f *blob) {
	if f.Full {
		for k := range state {
			if _, ok := f.Pages[k]; !ok {
				delete(state, k)
			}
		}
	}
	for k, v := range f.Pages {
		if v == 0 {
			delete(state, k)
		} else {
			state[k] = v
		}
	}
}

// replicate is the serving process's litestream replication (claim C3's
// exception: plain PUTs). A sync writes the pending writes as one local
// level-0 file at Pos+1; ship uploads the local files in order; then
// compaction, snapshot rotation and retention run as the store's
// monitors would.
func (w *world) replicate(b *box) {
	if !w.agentUp(b) || b.DB == nil || b.open != nil {
		return // replication is not open until the admitted session's start completes (rev 7.11 §7.14)
	}
	if !b.Old && b.DB.Epoch != b.Epoch {
		// the replicator that was admitted died with its agent; a new agent
		// process is a new session (claim rev 6.10) and replicates nothing
		// for the old one: its wake is a new admission
		w.stats["replication refused: the session's agent is gone"]++
		return
	}
	if !hasMain(b) {
		w.closeStep(b)
		return
	}
	w.flush(b)
	w.compact(b)
}

// flush is one sync of the replicator: the pending writes become the next
// local level-0 file, and the queue is shipped.
func (w *world) flush(b *box) {
	db := b.DB
	if len(db.Pending) > 0 {
		// the sync writes this TXID's file locally; the position advances
		// at once, and later writes go to the next TXID
		bl := w.deltaOf(db, db.Inc)
		db.FirstFull = false
		db.Queue = append(db.Queue, bl)
		db.Pos, db.Pending = bl.Max, nil
	}
	w.ship(b)
}

// closedKey is a writer session's clean-close certificate (claim rev 6.8
// §6.4a): <location>/<generation>/closed/<session>.json, written once and
// never deleted; the admitting CAS moves holder.session, so an old,
// delayed or replayed certificate names a dead session.
func closedKey(loc, gen string, session int) string {
	return dataKey(loc, gen, fmt.Sprintf("closed/%d.json", session))
}

// isClosedKey: a clean-close certificate of any session.
func isClosedKey(k string) bool {
	i := strings.LastIndex(k, "/closed/")
	return i >= 0 && strings.HasSuffix(k, ".json")
}

// track records a request the writer issued whose success the client did
// not see at once, for the clean close's evidence (R8-5, R9-5): every
// class, level-0, compaction, snapshot and retention. A request the server
// refused is a failure the client saw: the session is uncertain for good.
func (db *localDB) track(r *request, v int) {
	if v > 0 {
		return // client-observed success
	}
	if r.done {
		db.Uncertain = true // refused
		return
	}
	db.Issued = append(db.Issued, r)
}

// settled: every request the writer issued over its lifetime returned
// success to the client, within the client's wait, and none was retried
// after a timeout (R8-5, R9-5). A request still outstanding within the wait
// is not yet settled; one the client gave up on, whether it later
// completed on the server or not, makes the session uncertain for good,
// since its outcome was never observed. The mutation of close-evidence
// looks at the level-0 queue alone, as rev 7.8 did.
func (db *localDB) settled(now int) bool {
	queued := len(db.Queue) == 0 && (db.InFlight == nil || db.InFlight.done)
	if !on("close-evidence") {
		return queued
	}
	if !queued || db.Uncertain {
		return false
	}
	for _, r := range db.Issued {
		if !r.done {
			if now-r.at >= uploadTimeout {
				db.Uncertain = true // the client gave up: timeout
				return false
			}
			return false // still waiting
		}
		if r.doneAt-r.at >= uploadTimeout {
			db.Uncertain = true // completed after the client gave up: a lost response
			return false
		}
	}
	return true
}

// closeStep is the clean close of a stopped writer (claim rev 6.8 §6.4a):
// first every request it issued over its lifetime must have returned
// success (R8-5), then it writes closed/<session>.json {session, box,
// max_txid} once (If-None-Match), then it releases its lease. A writer
// with a request unresolved, or that ever retried after a timeout, writes
// no certificate; a crash while draining leaves none (forget). The
// mutation of close-waits writes the certificate as soon as the writer
// stops.
func (w *world) closeStep(b *box) {
	db := b.DB
	if db == nil || !db.Closing || hasMain(b) {
		return
	}
	if b.open != nil || !b.Old && db.Epoch != b.Epoch {
		// the session's start has not opened replication, or its replicator
		// died with its agent: nothing certifies (rev 7.11; seed 2663 /
		// 30,000 certified position 0 over two unshipped writes)
		db.Closing, db.CloseReq = false, nil
		w.stats["clean close refused: start unfinished or agent gone"]++
		return
	}
	if len(b.Procs) > 0 {
		// nothing of the host may run when the session certifies: a
		// descendant that outlived the unit stop (an escapee outside the
		// cgroup) can still acknowledge writes the certificate would not
		// cover, which the next admission's purge would lose silently (rev
		// 7.10; seed 17 / 300). The close is refused, not delayed: the
		// session stays uncertified, the next wake is cold and declares
		// (seed 97950 / 100,000: a close that waited for an escapee held
		// the revert's probe until it stuck)
		db.Closing, db.CloseReq = false, nil
		w.stats["clean close refused: a process still runs"]++
		w.logf("box %d: clean close refused, a descendant of the host still runs", b.ID)
		return
	}
	// the certificate is the live generation's: once an operation has
	// taken the writer over (a push, a move, a delete), or the claim has
	// moved on, the generation is dead or being replaced and the close is
	// dropped. A writer running inside an operation (a landing's probe or
	// its committed target, a rollback's reopened source, a push's probe)
	// certifies like any other (claim rev 6.8 §6.4a; seed 84087 / 100,000:
	// rev 7.9's first build dropped every close outside serving, so a
	// rollback's source was admitted cold at every wake and the rollback
	// never released)
	c := w.claim
	inside := c != nil && (c.Op.Kind == "serving" || c.Op.Kind == "moving" || c.Op.Kind == "landing" || c.Op.Kind == "restoring" || c.Op.Kind == "reverting")
	if c == nil || c.Holder == nil || !inside || c.Op.Exec != b.ID || c.Holder.Inc != db.Inc || db.Gen != c.Holder.Gen {
		// a push, a move's cut, a delete or a purge is taking this writer
		// over: its generation is being replaced or parked, and a
		// certificate written into it now would outlive the park (seed
		// 4493 / 30,000: a certificate landed in a generation a push had
		// superseded)
		db.Closing, db.CloseReq = false, nil
		return
	}
	w.ship(b)
	if !db.settled(w.now) && on("close-waits") {
		if db.Uncertain && on("close-evidence") {
			// a request of this writer timed out and was retried: its outcome
			// is unknown for good, so this session never certifies
			db.Closing = false
			w.stats["clean close refused: uncertain"]++
			w.logf("box %d: clean close refused, the session is uncertain", b.ID)
			return
		}
		inflight := db.InFlight != nil && !db.InFlight.done
		w.logf("box %d: clean close waits (queue %d, upload in flight %v, %d requests tracked)", b.ID, len(db.Queue), inflight, len(db.Issued))
		return
	}
	if db.CloseReq == nil {
		db.CloseReq = &request{kind: rPut, region: b.Region, key: closedKey(db.Loc, db.Gen, db.Session), ifSet: true,
			b: &blob{Kind: "closed", Gen: db.Gen, Inc: db.Inc, Stamp: db.Stamp, Seq: b.ID, Session: db.Session, TX: db.Pos}}
		w.issue(b.ID, db.CloseReq)
		w.stats["clean closes"]++

	}
	if db.CloseReq.done {
		db.Closing, db.CloseReq = false, nil
	} else if w.now-db.CloseReq.at >= uploadTimeout {
		// the certificate PUT itself was not acknowledged within the
		// client's wait: the close ends without a certificate the box can
		// rely on; the PUT may still land, and a later wake that finds it
		// may continue warm, since its content is right (the drain was
		// complete); meanwhile the next wake is cold (seed 97950 / 100,000:
		// a certificate PUT late by three horizons held the revert's probe
		// until it stuck)
		db.Closing, db.CloseReq = false, nil
		w.stats["clean close: certificate not acknowledged"]++
		w.logf("box %d: the certificate PUT was not acknowledged; the close ends uncertified", b.ID)
	}
}

// uploadTimeout is the client's patience with one level-0 upload, in
// steps, before litestream retries it.
const uploadTimeout = 8

// ship is the replica's upload loop: the local level-0 files go up in
// TXID order, and the next is never sent before the one ahead of it has
// completed (litestream does not advance past a failed upload, moves spec
// §5.1 step 6). An upload the client has waited uploadTimeout steps for is
// retried with the same bytes to the same key; the original may still land
// later, as an identical version. So a gap in a generation's current
// listing is always a file that has not landed yet, never a file that was
// skipped.
func (w *world) ship(b *box) {
	db := b.DB
	for len(db.Queue) > 0 {
		if r := db.InFlight; r != nil {
			if !r.done {
				if w.now-db.InFlightAt < uploadTimeout {
					return
				}
				// the client gives up and retries the same bytes; the
				// original's outcome is unknown for good (R8-5)
				db.Uncertain = true
				w.stats["ltx upload retries"]++
			} else {
				db.RemotePos = db.Queue[0].Max
				db.InFlight = nil
				db.Queue = db.Queue[1:]
				continue
			}
		}
		bl := db.Queue[0]
		// litestream's upload loop (replica.go Sync): the replica reads the
		// remote position once, at its first sync, caches it and advances it
		// with each upload it completes (R8-5); a local file at or below that
		// position is taken as already uploaded and skipped, and the writer
		// believes it replicated (R7-1, the initialization/remote-position
		// race: another writer's file at this TXID, present at that first
		// read, makes this one vanish)
		if db.RemotePos < 0 {
			db.RemotePos = w.maxL0(w.buckets[b.Region], dataKey(db.Loc, db.Gen, ""))
		}
		if db.RemotePos >= bl.Max {
			w.markDurable(ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), bl)
			w.stats["uploads skipped at the remote position"]++
			db.InFlight = nil
			db.Queue = db.Queue[1:]
			continue
		}
		r := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), b: bl}
		if v := w.issue(b.ID, r); v > 0 {
			db.RemotePos = bl.Max
			db.InFlight = nil
			db.Queue = db.Queue[1:]
			continue
		}
		db.track(r, 0)
		db.InFlight, db.InFlightAt = r, w.now
		return
	}
}

// currentLTX lists the current LTX files of a generation at one level, by
// range.
func (w *world) currentLTX(bk *bucket, prefix string, level int) []*version {
	var out []*version
	for _, k := range bk.keys(prefix) {
		l, _, _, ok := ltxLevel(k)
		if !ok || l != level || genPrefixOf(k) != prefix {
			continue
		}
		if v := bk.current(k); v != nil {
			out = append(out, v)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].B.Min < out[j].B.Min })
	return out
}

// compact merges the current level-0 files of b's generation into one
// level-1 file when there are at least three, deletes the inputs bar the
// newest, then
// writes a new snapshot and enforces retention (litestream's store after
// a compaction at the snapshot level): only the newest snapshot stays,
// and level-1 and level-0 files whose range ends below the previous
// snapshot's TXID are deleted. Every delete is a key delete: the versions
// stay noncurrent, which is what keeps the pinned snapshot version
// reachable (R5-3). Contiguity is checked as ltx.Compactor does.
func (w *world) compact(b *box) {
	db := b.DB
	bk := w.buckets[b.Region]
	prefix := dataKey(db.Loc, db.Gen, "")
	inputs := w.currentLTX(bk, prefix, 0)
	if len(inputs) < 3 {
		return
	}
	out := &blob{Kind: "ltx", Level: 1, Min: inputs[0].B.Min, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}, Full: inputs[0].B.Full}
	max := 0
	var keys []string
	for _, v := range inputs {
		if max != 0 && !(v.B.Min <= max+1 && v.B.Max > max) {
			return // non-contiguous inputs: litestream refuses to compact them
		}
		if v.B.Full {
			// a full input anywhere makes the image complete from there: the
			// output carries every page and a key the full file lacked is
			// gone (rev 7.10: a warm admission's first upload is full, fact
			// R5-1c, so full files appear mid-generation; seed 27371 /
			// 30,000 compacted one into a delta and a deletion was lost)
			out.Full = true
		}
		apply(out.Pages, v.B)
		for k, val := range v.B.Pages {
			if val == 0 {
				out.Pages[k] = 0 // a delete stays a delete in the compacted image
			}
		}
		out.TXs = append(out.TXs, v.B.TXs...)
		if v.B.Max > max {
			max = v.B.Max
		}
		keys = append(keys, ltxKey(db.Loc, db.Gen, 0, v.B.Min, v.B.Max))
	}
	out.Max = max
	l1 := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 1, out.Min, out.Max), b: out}
	if v := w.issue(b.ID, l1); v <= 0 {
		db.track(l1, v) // a late or lost compaction upload is part of the writer's close evidence (R8-5)
		return
	}
	// the inputs are deleted bar the newest: litestream's level-0 retention
	// never removes the newest level-0 file, which is where a fresh box
	// reads its position (facts R5-1f to h; R6 review). The mutation of
	// newest-l0 is rev 7.6's: every input, and below every L0 below the
	// previous snapshot
	keep := keys[len(keys)-1]
	for _, k := range keys {
		if k == keep && on("newest-l0") {
			continue
		}
		dr := &request{kind: rDelKey, region: b.Region, key: k}
		db.track(dr, w.issue(b.ID, dr))
	}
	w.stats["ltx compactions"]++
	// snapshot rotation: a new snapshot at the position, then retention
	prev := 0
	for _, s := range w.currentLTX(bk, prefix, 9) {
		if s.B.Max > prev {
			prev = s.B.Max
		}
	}
	snap := snapshotOf(db, db.Inc)
	snap.Kind = "ltx"
	if snap.Max <= prev {
		return
	}
	sr := &request{kind: rPut, region: b.Region, key: snapKey(db.Loc, db.Gen, snap.Max), b: snap}
	if v := w.issue(b.ID, sr); v <= 0 {
		db.track(sr, v)
		return
	}
	w.stats["snapshot rotations"]++
	// keep the newest snapshot; minSnapshotTXID is the previous one's
	// MaxTXID (litestream db.EnforceSnapshotRetention); expire levels 1
	// and 0 below it
	for _, s := range w.currentLTX(bk, prefix, 9) {
		if s.B.Max < snap.Max {
			dr := &request{kind: rDelKey, region: b.Region, key: snapKey(db.Loc, db.Gen, s.B.Max)}
			db.track(dr, w.issue(b.ID, dr))
		}
	}
	newest := w.maxL0(bk, prefix)
	for _, level := range []int{1, 0} {
		for _, f := range w.currentLTX(bk, prefix, level) {
			if level == 0 && f.B.Max == newest && on("newest-l0") {
				continue // the newest level-0 file survives every retention path (R7-3)
			}
			if f.B.Max < prev {
				dr := &request{kind: rDelKey, region: b.Region, key: ltxKey(db.Loc, db.Gen, level, f.B.Min, f.B.Max)}
				db.track(dr, w.issue(b.ID, dr))
			}
		}
	}
}

// remoteMax is the generation's remote maximum TXID over every level.
func (w *world) remoteMax(bk *bucket, prefix string) int {
	max := 0
	for _, level := range []int{0, 1, 9} {
		for _, f := range w.currentLTX(bk, prefix, level) {
			if f.B.Max > max {
				max = f.B.Max
			}
		}
	}
	return max
}

// plan is litestream's restore planner (replica.go CalcRestorePlan,
// v0.5.14) over the current files under prefix, from TXID cur: take the
// file that extends the range, overlap allowed (ltx.IsContiguous:
// min <= cur+1 && max > cur), preferring the highest max, then the lowest
// min, then the higher level; stop when nothing extends it. The mutation
// of restore-overlap is rev 7.4's rule, min == cur+1, which discards a
// compacted continuation. With restore-gap-closed, a current file that
// starts beyond cur+1 once nothing extends the range is a gap and the
// restore refuses (litestream: "non-contiguous ltx files"); the mutation
// returns what it has.
func (w *world) plan(bk *bucket, prefix string, cur int) (files []*blob, gap bool) {
	var cands []*blob
	for _, k := range bk.keys(prefix) {
		level, _, _, ok := ltxLevel(k)
		if !ok || level == 9 || genPrefixOf(k) != prefix {
			continue // snapshots are the root, never planned
		}
		if v := bk.current(k); v != nil {
			cands = append(cands, v.B)
		}
	}
	for {
		var best *blob
		for _, f := range cands {
			ext := f.Min <= cur+1 && f.Max > cur
			if !on("restore-overlap") {
				ext = f.Min == cur+1
			}
			if !ext {
				continue
			}
			if best == nil || f.Max > best.Max || f.Max == best.Max && (f.Min < best.Min || f.Min == best.Min && f.Level > best.Level) {
				best = f
			}
		}
		if best == nil {
			break
		}
		files = append(files, best)
		cur = best.Max
	}
	for _, f := range cands {
		if f.Min > cur+1 {
			gap = true
		}
	}
	return files, gap
}

// maxL0 is the newest current level-0 file's TXID of a generation: what
// litestream's open takes as a fresh box's position (R5-1).
func (w *world) maxL0(bk *bucket, prefix string) int {
	max := 0
	for _, f := range w.currentLTX(bk, prefix, 0) {
		if f.B.Max > max {
			max = f.B.Max
		}
	}
	return max
}

// expectState replays the incarnation's acknowledged, non-exempt writes,
// in acknowledgement order, into the database a correct restore must
// equal. With upto >= 0 it replays the transactions with lineage TXID 1
// to upto, what a cold restore that reached upto must hold; with upto <
// 0 it replays every acknowledged write, a write that never reached a
// database counting as the insert the application believed it made, which
// is what the serving box's own database must hold.
func (w *world) expectState(inc int, upto int) map[string]int {
	state := map[string]int{}
	for _, a := range w.acks {
		if a.Inc != inc || a.Exempt || a.Kind != "write" {
			continue
		}
		t := w.txs[a.W]
		if upto >= 0 && (t.TX == 0 || t.TX > upto) {
			continue
		}
		if t.Key == "" {
			t.Key, t.Val = fmt.Sprintf("k%d", a.W), a.W
		}
		if t.Val == 0 {
			delete(state, t.Key)
		} else {
			state[t.Key] = t.Val
		}
	}
	return state
}

// txsOf is the union, in order, of the writes the files cover: the oracle's
// log of a restored database.
func txsOf(files ...*blob) []int {
	seen := map[int]bool{}
	var out []int
	for _, f := range files {
		for _, wr := range f.TXs {
			if !seen[wr] {
				seen[wr] = true
				out = append(out, wr)
			}
		}
	}
	return out
}

// lineageTX is the oracle's lineage TXID of a restored state: the highest
// transaction id among the writes the applied files cover.
func (w *world) lineageTX(files ...*blob) int {
	max := 0
	for _, f := range files {
		for _, wr := range f.TXs {
			if t := w.txs[wr].TX; t > max {
				max = t
			}
		}
	}
	return max
}

var _ = strings.HasPrefix