package genselect

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/superfly/ltx"
)

type fixture struct {
	t             *testing.T
	b             *leasetest.VersionedBucket
	req           Request
	marker        Marker
	markerVersion string
	pin           File
	store         Reader
}

func newFixture(t *testing.T) *fixture {
	b := leasetest.NewVersionedBucket()
	b.AddBucket("data")
	b.PageSize = 1
	f := &fixture{t: t, b: b, store: b, req: Request{Bucket: "data", Region: "ie", Host: "h",
		Holder: hostclaim.Holder{Incarnation: "inc", Location: "instances/h/inc", Generation: "g0"}}}
	f.pin = f.file("g0", 9, 1, 11)
	f.file("g0", 0, 11, 11)
	f.marker = Marker{Op: "g0", Incarnation: "inc", Host: "h", Box: "box", InstalledAt: time.Unix(1, 0).UTC(),
		ManifestSHA256: strings.Repeat("a", 64), Pin: Pin{SnapshotKey: f.pin.Key, SnapshotVersion: f.pin.VersionID, SnapshotTXID: 11}}
	f.publish()
	return f
}

func header(t *testing.T, min, max uint64) []byte {
	t.Helper()
	h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096,
		MinTXID: ltx.TXID(min), MaxTXID: ltx.TXID(max)}
	b, err := h.MarshalBinary()
	if err != nil {
		t.Fatal(err)
	}
	return b
}

func (f *fixture) file(gen string, level int, min, max uint64) File {
	f.t.Helper()
	key := fmt.Sprintf("%s%04x/%016x-%016x.ltx", prefix(strings.TrimSuffix(f.req.Holder.Location, "/")+"/", origin(gen)), level, min, max)
	b := header(f.t, min, max)
	id := f.b.Put("data", key, b)
	return File{Key: key, VersionID: id, Level: level, MinTXID: min, MaxTXID: max, Size: int64(len(b))}
}

func (f *fixture) publish() {
	f.t.Helper()
	b, err := json.Marshal(f.marker)
	if err != nil {
		f.t.Fatal(err)
	}
	f.markerVersion = f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
}

func (f *fixture) chain(base string) {
	f.req.Holder.Generation = "g2"
	f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
		{Generation: "g1", Predecessor: base, Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
		{Generation: "g2", Predecessor: "g1", Region: "ie", PredecessorRegion: "ie", Mode: "cold"},
	}
}

func (f *fixture) origin() {
	f.req.Holder.Generation = ""
	if err := f.b.DeleteVersion(f.t.Context(), "data", f.req.Holder.Location+"/generation.json", f.markerVersion); err != nil {
		f.t.Fatal(err)
	}
	f.markerVersion = ""
}

// These are pure-selection ports, not simulations of wake, park or replication.
// Paths refer to platform-moves-design/internal/sim/hostlife at rev 7.13.
func TestModelWitnesses(t *testing.T) {
	tests := []struct {
		name, source string
		setup        func(*fixture) []File
		gen          string
		wantErr      error
	}{
		{"wRestoreOverlap", "witness_test.go:1204", func(f *fixture) []File {
			return []File{f.pin, f.file("g0", 1, 10, 13)}
		}, "g0", nil},
		{"wRestoreGap", "witness_test.go:1233", func(f *fixture) []File {
			f.file("g0", 0, 13, 13)
			return nil
		}, "", ErrGap},
		{"wRootAdvance", "witness_test.go:1295", func(f *fixture) []File {
			f.file("g0", 0, 13, 13)
			root := f.file("g0", 9, 1, 20)
			f.b.BareDelete("data", f.pin.Key)
			return []File{root, f.file("g0", 0, 21, 21)}
		}, "g0", nil},
		{"wOriginSwept/origin", "witness_test.go:2373", func(f *fixture) []File {
			f.origin()
			return []File{f.file("", 0, 1, 1), f.file("", 1, 2, 5)}
		}, hostclaim.RootGeneration, nil},
		{"wOriginSwept/unfinished_chain", "witness_test.go:2373; protocol_test.go:1188", func(f *fixture) []File {
			f.origin()
			f.chain(hostclaim.RootGeneration)
			return []File{f.file("", 0, 1, 3)}
		}, hostclaim.RootGeneration, nil},
		{"wBaselineParked/all_empty", "witness_test.go:2168; protocol_test.go:1154", func(f *fixture) []File {
			f.chain("g0")
			return []File{f.pin}
		}, "g0", nil},
		{"wBaselineParked/newest_nonempty", "witness_test.go:2168; protocol_test.go:1172", func(f *fixture) []File {
			f.chain("g0")
			return []File{f.file("g1", 0, 1, 1)}
		}, "g1", nil},
		{"wBaselineParked/holder_nonempty", "protocol_test.go:1161", func(f *fixture) []File {
			f.chain("g0")
			f.file("g1", 0, 1, 9)
			return []File{f.file("g2", 0, 1, 1)}
		}, "g2", nil},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Log("port of", tt.source)
			f := newFixture(t)
			want := tt.setup(f)
			before := f.b.Snapshot()
			f.b.FailOn = func(op, key string) error {
				if op != "list" && op != "get" {
					t.Fatalf("mutation %s %s", op, key)
				}
				return nil
			}
			got, err := Select(t.Context(), f.store, f.req)
			if !errors.Is(err, tt.wantErr) {
				t.Fatalf("got %v, want %v", err, tt.wantErr)
			}
			if f.b.Snapshot() != before {
				t.Fatal("selection mutated storage")
			}
			if tt.wantErr != nil {
				if got != nil {
					t.Fatal("partial plan")
				}
				return
			}
			if got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) {
				t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
			}
			if got.MarkerVersion != f.markerVersion || got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID {
				t.Fatalf("root, marker or final TXID: %+v", got)
			}
		})
	}
}

type altered struct {
	Reader
	list func([]lease.ObjectVersion) ([]lease.ObjectVersion, error)
	open func(context.Context, string, string, string) (io.ReadCloser, error)
}

func (s altered) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
	vs, err := s.Reader.ListVersions(ctx, bucket, base)
	if err != nil {
		return nil, err
	}
	if s.list != nil {
		return s.list(vs)
	}
	return vs, nil
}
func (s altered) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
	if s.open != nil {
		return s.open(ctx, bucket, key, id)
	}
	return s.Reader.OpenVersion(ctx, bucket, key, id)
}

func TestR13_1UnfinishedRegionalWalk(t *testing.T) {
	tests := []struct {
		name  string
		setup func(*fixture) []File
		gen   string
		want  error
	}{
		{"interrupted_clean_return", func(f *fixture) []File {
			return []File{f.pin}
		}, "g0", nil},
		{"another_interrupted_admission", func(f *fixture) []File {
			f.req.Holder.Generation = "R2"
			f.req.Holder.Admissions = append(f.req.Holder.Admissions,
				hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
			return []File{f.pin}
		}, "g0", nil},
		{"covered_foreign_link_deeper_in_run", func(f *fixture) []File {
			f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
				{Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
				{Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: true},
				{Generation: "R", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold"},
			}
			return []File{f.pin}
		}, "g0", nil},
		{"uncovered_foreign_link", func(f *fixture) []File {
			f.req.Holder.Admissions[1].Covered = false
			return []File{f.pin}
		}, "g0", nil},
		{"foreign_objects_never_selected", func(f *fixture) []File {
			f.req.Holder.Admissions[1].Covered = false
			f.file("T", 9, 1, 99)
			return []File{f.pin}
		}, "g0", nil},
		{"holder_objects_with_covered_foreign_predecessor", func(f *fixture) []File {
			return []File{f.file("R", 0, 1, 1)}
		}, "R", nil},
		{"newest_local_objects", func(f *fixture) []File {
			f.req.Holder.Generation = "R2"
			f.req.Holder.Admissions = append(f.req.Holder.Admissions,
				hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold"})
			return []File{f.file("R", 0, 1, 1)}
		}, "R", nil},
		{"unsafe_covered_local_baseline", func(f *fixture) []File {
			f.req.Holder.Generation = "R2"
			f.req.Holder.Admissions = append(f.req.Holder.Admissions,
				hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
			f.file("R", 0, 1, 1)
			return nil
		}, "", ErrCovered},
		{"unsafe_covered_empty_local_baseline", func(f *fixture) []File {
			f.req.Holder.Generation = "R2"
			f.req.Holder.Admissions = append(f.req.Holder.Admissions,
				hostclaim.AdmissionRecord{Generation: "R2", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "cold", Covered: true})
			return nil
		}, "", ErrCovered},
		{"unsafe_covered_local_marker", func(f *fixture) []File {
			f.req.Holder.Admissions[0].Covered = true
			return nil
		}, "", ErrCovered},
		{"older_covered_local_not_needed", func(f *fixture) []File {
			f.req.Holder.Admissions[0].Covered = true
			return []File{f.file("R", 0, 1, 1)}
		}, "R", nil},
		{"covered_warm_session_in_run", func(f *fixture) []File {
			f.req.Holder.Admissions = append(f.req.Holder.Admissions,
				hostclaim.AdmissionRecord{Generation: "R", Predecessor: "R", Region: "ie", PredecessorRegion: "ie", Mode: "warm", Covered: true})
			return []File{f.pin}
		}, "g0", nil},
		{"absent_marker_origin", func(f *fixture) []File {
			f.origin()
			f.req.Holder.Generation = "R"
			f.req.Holder.Admissions[0].Predecessor = hostclaim.RootGeneration
			return []File{f.file("", 0, 1, 1)}
		}, hostclaim.RootGeneration, nil},
		{"foreign_predecessor_region_marker_fallback", func(f *fixture) []File {
			f.chain("g0")
			f.req.Holder.Admissions[0].PredecessorRegion = "us"
			return []File{f.pin}
		}, "g0", nil},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFixture(t)
			f.req.Holder.Generation = "R"
			f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
				{Generation: "T", Predecessor: "g0", Region: "us", PredecessorRegion: "ie", Mode: "cold"},
				{Generation: "R", Predecessor: "T", Region: "ie", PredecessorRegion: "us", Mode: "cold", Coverage: "clean-return", Covered: true},
			}
			want := tt.setup(f)
			before := f.b.Snapshot()
			holderBefore, _ := json.Marshal(f.req.Holder)
			got, err := Select(t.Context(), f.store, f.req)
			holderAfter, _ := json.Marshal(f.req.Holder)
			if f.b.Snapshot() != before || string(holderAfter) != string(holderBefore) {
				t.Fatal("selection mutated inputs or storage")
			}
			if !errors.Is(err, tt.want) {
				t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
			}
			if tt.want != nil {
				if got != nil {
					t.Fatal("partial plan")
				}
				return
			}
			if got == nil || got.Generation != tt.gen || !reflect.DeepEqual(got.Files, want) ||
				got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
				got.MarkerVersion != f.markerVersion {
				t.Fatalf("got %+v, want %s %+v", got, tt.gen, want)
			}
			if tt.gen == f.marker.Op && (got.Pin == nil || *got.Pin != f.marker.Pin) {
				t.Fatalf("marker pin not preserved: %+v", got.Pin)
			}
		})
	}
}

func TestRefusals(t *testing.T) {
	tests := []struct {
		name  string
		setup func(*fixture)
		want  error
	}{
		{"marker_holder_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown" }, ErrDisagreement},
		{"nonempty_unexplained_disagreement", func(f *fixture) { f.req.Holder.Generation = "unknown"; f.file("unknown", 0, 1, 1) }, ErrDisagreement},
		{"covered_local_marker", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Covered = true }, ErrCovered},
		{"broken_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[1].Predecessor = "missing" }, ErrDisagreement},
		{"cyclic_chain", func(f *fixture) { f.chain("g0"); f.req.Holder.Admissions[0].Predecessor = "g2" }, ErrDisagreement},
		{"incomplete_listing", func(f *fixture) { f.b.FailListAfterPages = 1 }, ErrListing},
		{"partial_data_with_error", func(f *fixture) {
			f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) { return vs, io.ErrUnexpectedEOF }}
		}, ErrListing},
		{"incomplete_unfinished_listing", func(f *fixture) { f.chain("g0"); f.b.FailListAfterPages = 1 }, ErrListing},
		{"pinned_version_missing", func(f *fixture) {
			f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
			if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
				f.t.Fatal(err)
			}
		}, ErrPinnedMissing},
		{"pinned_version_vanished_after_listing", func(f *fixture) {
			f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
				return nil, lease.ErrMirrorNotFound
			}}
		}, ErrPinnedMissing},
		{"pinned_header_mismatch", func(f *fixture) {
			f.marker.SnapshotVersion = f.b.Put("data", f.pin.Key, header(f.t, 1, 12))
			f.publish()
		}, ErrPinnedMismatch},
		{"pinned_key_mismatch", func(f *fixture) { f.marker.SnapshotTXID = 12; f.publish() }, ErrPinnedMismatch},
		{"pinned_delete_marker", func(f *fixture) { f.marker.SnapshotVersion = f.b.BareDelete("data", f.pin.Key); f.publish() }, ErrDeleteMarker},
		{"selected_delete_marker_on_open", func(f *fixture) {
			f.store = altered{Reader: f.b, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
				return nil, lease.ErrIsDeleteMarker
			}}
		}, ErrDeleteMarker},
		{"marker_delete_marker", func(f *fixture) { f.b.BareDelete("data", f.req.Holder.Location+"/generation.json") }, ErrDeleteMarker},
		{"marker_read_error", func(f *fixture) {
			f.b.FailOn = func(op, key string) error {
				if op == "get" {
					return io.ErrUnexpectedEOF
				}
				return nil
			}
		}, ErrMarker},
		{"marker_disappears", func(f *fixture) {
			f.b.FailOn = func(op, key string) error {
				if op == "get" {
					return lease.ErrMirrorNotFound
				}
				return nil
			}
		}, ErrMarker},
		{"marker_incarnation", func(f *fixture) { f.marker.Incarnation = "previous"; f.publish() }, ErrIncarnation},
		{"marker_host", func(f *fixture) { f.marker.Host = "other"; f.publish() }, ErrMarker},
		{"malformed_marker", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{")) }, ErrMarker},
		{"trailing_marker_json", func(f *fixture) {
			b, _ := json.Marshal(f.marker)
			f.b.Put("data", f.req.Holder.Location+"/generation.json", append(b, []byte(" {}")...))
		}, ErrMarker},
		{"empty_pin_version", func(f *fixture) { f.marker.SnapshotVersion = ""; f.publish() }, ErrVersionID},
		{"null_pin_version", func(f *fixture) { f.marker.SnapshotVersion = "null"; f.publish() }, ErrVersionID},
		{"empty_current_version", func(f *fixture) { alterVersion(f, "") }, ErrVersionID},
		{"null_current_version", func(f *fixture) { alterVersion(f, "null") }, ErrVersionID},
		{"tombstoned_unfinished_is_not_empty", func(f *fixture) {
			f.chain("g0")
			dead := f.file("g2", 0, 1, 1)
			f.b.BareDelete("data", dead.Key)
		}, ErrEmpty},
		{"metadata_only_unfinished", func(f *fixture) {
			f.chain("g0")
			f.b.Put("data", f.req.Holder.Location+"/g-g2/build-intent.json", []byte("{}"))
		}, ErrEmpty},
		{"empty_published_generation", func(f *fixture) {
			f.marker.Op = "empty"
			f.marker.Pin = Pin{}
			f.req.Holder.Generation = "empty"
			f.publish()
		}, ErrEmpty},
		{"bad_successor_no_fallback", func(f *fixture) { r := f.file("g0", 9, 1, 20); f.b.Put("data", r.Key, header(f.t, 1, 19)) }, ErrRoot},
		{"missing_snapshot_txid", func(f *fixture) {
			b, _ := json.Marshal(f.marker)
			var fields map[string]any
			json.Unmarshal(b, &fields)
			delete(fields, "snapshot_txid")
			b, _ = json.Marshal(fields)
			f.b.Put("data", f.req.Holder.Location+"/generation.json", b)
		}, ErrMarker},
		{"malformed_current_ltx", func(f *fixture) { f.b.Put("data", f.req.Holder.Location+"/g-g0/0000/not-a-range.ltx", []byte("bad")) }, ErrListing},
		{"invalid_snapshot_min", func(f *fixture) { f.file("g0", 9, 2, 20) }, ErrRoot},
		{"gap_without_root", func(f *fixture) { f.origin(); f.file("", 0, 2, 2) }, ErrGap},
		{"location_escape", func(f *fixture) { f.req.Holder.Location = "instances/h/../other" }, ErrInput},
		{"holder_root_sentinel", func(f *fixture) { f.req.Holder.Generation = hostclaim.RootGeneration }, ErrInput},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFixture(t)
			tt.setup(f)
			before := f.b.Snapshot()
			got, err := Select(t.Context(), f.store, f.req)
			if got != nil || !errors.Is(err, tt.want) {
				t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
			}
			if f.b.Snapshot() != before {
				t.Fatal("mutated storage")
			}
		})
	}
}

func TestContinuationRootRefusals(t *testing.T) {
	for _, gen := range []string{"", "fresh"} {
		name := gen
		if name == "" {
			name = "origin"
		}
		for level := 0; level < 9; level++ {
			for _, corruption := range []string{"malformed_header", "truncated_file", "header_range_mismatch"} {
				t.Run(fmt.Sprintf("%s/level_%d/%s", name, level, corruption), func(t *testing.T) {
					f := newFixture(t)
					if gen == "" {
						f.origin()
					} else {
						f.marker.Op = gen
						f.marker.Pin = Pin{}
						f.req.Holder.Generation = gen
						f.publish()
					}
					root := f.file(gen, level, 1, 1)
					body := header(t, 1, 1)
					switch corruption {
					case "malformed_header":
						body[0] ^= 0xff
					case "truncated_file":
						body = body[:len(body)-1]
					case "header_range_mismatch":
						body = header(t, 1, 2)
					}
					f.b.Put("data", root.Key, body)
					before := f.b.Snapshot()
					got, err := Select(t.Context(), f.store, f.req)
					if got != nil || !errors.Is(err, ErrRoot) {
						t.Fatalf("got %+v / %v, want nil / %v", got, err, ErrRoot)
					}
					if f.b.Snapshot() != before {
						t.Fatal("selection mutated storage")
					}
				})
			}
		}
	}
}

func TestMaxLevel0TXID(t *testing.T) {
	tests := []struct {
		name         string
		setup        func(*fixture)
		wantLevel0   uint64
		wantRestored uint64
	}{
		{"current_anchor", func(f *fixture) {}, 11, 11},
		{"current_continuation", func(f *fixture) {
			f.file("g0", 0, 12, 12)
			f.file("g0", 0, 13, 13)
		}, 13, 13},
		{"noncurrent_data", func(f *fixture) {
			old := f.file("g0", 0, 30, 30)
			f.b.BareDelete("data", old.Key)
		}, 11, 11},
		{"delete_marker_only", func(f *fixture) {
			old := f.file("g0", 0, 40, 40)
			f.b.BareDelete("data", old.Key)
			if err := f.b.DeleteVersion(f.t.Context(), "data", old.Key, old.VersionID); err != nil {
				f.t.Fatal(err)
			}
		}, 11, 11},
		{"newer_restored_snapshot", func(f *fixture) {
			f.file("g0", 9, 1, 30)
		}, 11, 30},
		{"no_current_level0", func(f *fixture) {
			anchor := f.file("g0", 0, 11, 11)
			f.b.BareDelete("data", anchor.Key)
		}, 0, 11},
	}
	for _, entry := range []string{"ordinary", "prepared"} {
		for _, tt := range tests {
			t.Run(entry+"/"+tt.name, func(t *testing.T) {
				f := newFixture(t)
				tt.setup(f)
				var got *Plan
				var err error
				if entry == "ordinary" {
					got, err = Select(t.Context(), f.store, f.req)
				} else {
					p := f.marker.Pin
					got, err = SelectPrepared(t.Context(), f.store, f.req, PreparedPin{
						Generation: "g0", SnapshotKey: p.SnapshotKey, SnapshotVersion: p.SnapshotVersion,
						SnapshotTXID: p.SnapshotTXID, SnapshotCRC64: p.SnapshotCRC64,
					})
				}
				if err != nil || got == nil {
					t.Fatalf("got %+v / %v, want a plan", got, err)
				}
				if got.MaxLevel0TXID != tt.wantLevel0 || got.MaxTXID != tt.wantRestored {
					t.Fatalf("MaxLevel0TXID=%d, MaxTXID=%d; want %d, %d",
						got.MaxLevel0TXID, got.MaxTXID, tt.wantLevel0, tt.wantRestored)
				}
			})
		}
	}
}

func alterVersion(f *fixture, id string) {
	f.file("g0", 0, 12, 12)
	f.store = altered{Reader: f.b, list: func(vs []lease.ObjectVersion) ([]lease.ObjectVersion, error) {
		for i := range vs {
			if strings.Contains(vs[i].Key, "0000/000000000000000c-") {
				vs[i].VersionID = id
			}
		}
		return vs, nil
	}}
}

func TestSelectionDetails(t *testing.T) {
	tests := []struct {
		name  string
		setup func(*fixture) []File
	}{
		{"pinned_noncurrent_duplicate", func(f *fixture) []File {
			f.b.Put("data", f.pin.Key, header(f.t, 1, 11))
			return []File{f.pin}
		}},
		{"pinned_under_retention_tombstone", func(f *fixture) []File {
			f.b.BareDelete("data", f.pin.Key)
			return []File{f.pin}
		}},
		{"retention_tombstones_ignored", func(f *fixture) []File {
			f.b.BareDelete("data", f.file("g0", 0, 13, 13).Key)
			return []File{f.pin}
		}},
		{"max_then_min_then_level", func(f *fixture) []File {
			f.file("g0", 8, 12, 19)
			f.file("g0", 8, 12, 20)
			f.file("g0", 1, 10, 20)
			return []File{f.pin, f.file("g0", 2, 10, 20)}
		}},
		{"origin_snapshot_and_isolation", func(f *fixture) []File {
			f.origin()
			f.file("other", 9, 1, 999)
			f.b.Put("data", f.req.Holder.Location+"/child/0009/0000000000000001-000000000000ffff.ltx", header(f.t, 1, 65535))
			return []File{f.file("", 9, 1, 20), f.file("", 0, 21, 21)}
		}},
		{"legacy_location", func(f *fixture) []File {
			f.req.Holder.Location = "instances/h"
			f.req.Holder.Generation = ""
			return []File{f.file("", 0, 1, 1)}
		}},
		{"fresh_zero_pin_marker", func(f *fixture) []File {
			f.marker.Op = "fresh"
			f.marker.Pin = Pin{}
			f.req.Holder.Generation = "fresh"
			f.publish()
			return []File{f.file("fresh", 0, 1, 1)}
		}},
		{"warm_record_in_unfinished_run", func(f *fixture) []File {
			f.chain("g0")
			f.req.Holder.Admissions = append(f.req.Holder.Admissions, hostclaim.AdmissionRecord{Generation: "g2", Predecessor: "g2", Region: "ie", PredecessorRegion: "ie", Mode: "warm"})
			return []File{f.pin}
		}},
		{"root_advance_past_missing_old_pin", func(f *fixture) []File {
			if err := f.b.DeleteVersion(f.t.Context(), "data", f.pin.Key, f.pin.VersionID); err != nil {
				f.t.Fatal(err)
			}
			return []File{f.file("g0", 9, 1, 30)}
		}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFixture(t)
			want := tt.setup(f)
			got, err := Select(t.Context(), f.store, f.req)
			if err != nil || !reflect.DeepEqual(got.Files, want) {
				t.Fatalf("got %+v / %v, want %+v", got, err, want)
			}
		})
	}
}

// Port of resumeOpen's prepared branch (protocol_test.go:658) and
// restorePinned/restoreWith (:183/:187), moves §5.1 / §7.14.
func TestPreparedRestore(t *testing.T) {
	tests := []struct {
		name            string
		setup           func(*fixture, *PreparedPin, File) []File
		want            error
		ordinaryRefuses bool
	}{
		{"absent_target_marker_covered_source", func(f *fixture, p *PreparedPin, root File) []File {
			f.origin()
			f.req.Holder.Generation = "g2"
			return []File{root, f.file("g2", 1, 29, 33)}
		}, nil, true},
		{"absent_target_marker_trimmed_source", func(f *fixture, p *PreparedPin, root File) []File {
			f.origin()
			f.req.Holder.Generation = "g2"
			f.req.Holder.Admissions = f.req.Holder.Admissions[1:]
			return []File{root}
		}, nil, true},
		{"unpublished_historical_pin", func(f *fixture, p *PreparedPin, root File) []File {
			f.b.Put("data", root.Key, header(f.t, 1, 30))
			return []File{root}
		}, nil, false},
		{"marker_pin_disagreement_ignored", func(f *fixture, p *PreparedPin, root File) []File {
			f.marker.Op = "g2" // Its old g0 pin cannot override the authorized pin.
			f.publish()
			return []File{root}
		}, nil, false},
		{"malformed_marker_ignored", func(f *fixture, p *PreparedPin, root File) []File {
			f.b.Put("data", f.req.Holder.Location+"/generation.json", []byte("{"))
			return []File{root}
		}, nil, false},
		{"marker_tombstone_ignored", func(f *fixture, p *PreparedPin, root File) []File {
			f.b.BareDelete("data", f.req.Holder.Location+"/generation.json")
			return []File{root}
		}, nil, false},
		{"missing_prepared_root_no_current_key_fallback", func(f *fixture, p *PreparedPin, root File) []File {
			f.b.Put("data", root.Key, header(f.t, 1, 30))
			deletePreparedRoot(f, root)
			return nil
		}, ErrPinnedMissing, false},
		{"empty_prepared_generation_never_older_baseline", func(f *fixture, p *PreparedPin, root File) []File {
			f.chain("g0")
			deletePreparedRoot(f, root)
			versions, err := f.b.ListVersions(f.t.Context(), "data", prefix(f.req.Holder.Location+"/", "g2"))
			if err != nil {
				f.t.Fatal(err)
			}
			for _, v := range versions {
				if err := f.b.DeleteVersion(f.t.Context(), "data", v.Key, v.VersionID); err != nil {
					f.t.Fatal(err)
				}
			}
			baseline, err := Select(f.t.Context(), f.b, f.req)
			if err != nil || baseline.Generation != "g0" {
				f.t.Fatalf("ordinary baseline: %+v / %v", baseline, err)
			}
			return nil
		}, ErrPinnedMissing, false},
		{"zero_txid_prepared_pin_refuses", func(f *fixture, p *PreparedPin, root File) []File {
			*p = PreparedPin{Generation: "empty"}
			return nil
		}, ErrPinnedMismatch, false},
		{"prepared_pin_key_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
			p.SnapshotTXID++
			return nil
		}, ErrPinnedMismatch, false},
		{"prepared_pin_other_generation", func(f *fixture, p *PreparedPin, root File) []File {
			p.SnapshotKey = f.pin.Key
			p.SnapshotVersion = f.pin.VersionID
			p.SnapshotTXID = f.pin.MaxTXID
			return nil
		}, ErrPinnedMismatch, false},
		{"prepared_pin_header_mismatch", func(f *fixture, p *PreparedPin, root File) []File {
			p.SnapshotVersion = f.b.Put("data", root.Key, header(f.t, 1, 29))
			return nil
		}, ErrPinnedMismatch, false},
		{"prepared_pin_empty_version", func(f *fixture, p *PreparedPin, root File) []File {
			p.SnapshotVersion = ""
			return nil
		}, ErrVersionID, false},
		{"prepared_pin_deleted_version", func(f *fixture, p *PreparedPin, root File) []File {
			p.SnapshotVersion = f.b.BareDelete("data", root.Key)
			return nil
		}, ErrDeleteMarker, false},
		{"prepared_root_advance_past_missing_pin", func(f *fixture, p *PreparedPin, root File) []File {
			deletePreparedRoot(f, root)
			f.file("other", 9, 1, 999)
			return []File{f.file("g2", 9, 1, 40), f.file("g2", 0, 41, 41)}
		}, nil, false},
		{"prepared_bad_successor_no_fallback", func(f *fixture, p *PreparedPin, root File) []File {
			r := f.file("g2", 9, 1, 40)
			f.b.Put("data", r.Key, header(f.t, 1, 39))
			return nil
		}, ErrRoot, false},
		{"prepared_incomplete_listing", func(f *fixture, p *PreparedPin, root File) []File {
			f.b.FailListAfterPages = 1
			return nil
		}, ErrListing, false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			f := newFixture(t)
			f.req.Holder.Generation = "g2"
			f.req.Holder.Admissions = []hostclaim.AdmissionRecord{
				{Generation: "source", Predecessor: "older", Region: "us", PredecessorRegion: "us", Mode: "cold", Covered: true},
				{Generation: "g2", Predecessor: "source", Region: "ie", PredecessorRegion: "us", Mode: "cold"},
			}
			root := f.file("g2", 9, 1, 30)
			f.file("g2", 0, 30, 30)
			prepared := PreparedPin{Generation: "g2", SnapshotKey: root.Key, SnapshotVersion: root.VersionID, SnapshotTXID: 30, SnapshotCRC64: 123}
			want := tt.setup(f, &prepared, root)
			if tt.ordinaryRefuses {
				got, err := Select(t.Context(), f.store, f.req)
				if got != nil || !errors.Is(err, ErrDisagreement) {
					t.Fatalf("ordinary selection: %+v / %v", got, err)
				}
			}
			before := f.b.Snapshot()
			holderBefore, _ := json.Marshal(f.req.Holder)
			f.b.FailOn = func(op, key string) error {
				if op != "list" && op != "get" {
					t.Fatalf("mutation %s %s", op, key)
				}
				return nil
			}
			reader := preparedReader{Reader: f.store, t: t, base: prefix(f.req.Holder.Location+"/", prepared.Generation)}
			got, err := SelectPrepared(t.Context(), reader, f.req, prepared)
			holderAfter, _ := json.Marshal(f.req.Holder)
			if f.b.Snapshot() != before || string(holderBefore) != string(holderAfter) {
				t.Fatal("selection mutated inputs or storage")
			}
			if !errors.Is(err, tt.want) {
				t.Fatalf("got %+v / %v, want %v", got, err, tt.want)
			}
			if tt.want != nil {
				if got != nil {
					t.Fatal("partial plan")
				}
				return
			}
			wantPin := Pin{SnapshotKey: prepared.SnapshotKey, SnapshotVersion: prepared.SnapshotVersion, SnapshotTXID: prepared.SnapshotTXID, SnapshotCRC64: prepared.SnapshotCRC64}
			if got == nil || got.Generation != prepared.Generation || !reflect.DeepEqual(got.Files, want) ||
				got.Root == nil || *got.Root != want[0] || got.MaxTXID != want[len(want)-1].MaxTXID ||
				got.Marker != nil || got.MarkerVersion != "" || got.Pin == nil || *got.Pin != wantPin {
				t.Fatalf("got %+v, want prepared %s %+v", got, prepared.Generation, want)
			}
		})
	}
}

func deletePreparedRoot(f *fixture, root File) {
	f.t.Helper()
	if err := f.b.DeleteVersion(f.t.Context(), "data", root.Key, root.VersionID); err != nil {
		f.t.Fatal(err)
	}
}

type preparedReader struct {
	Reader
	t    *testing.T
	base string
}

func (s preparedReader) ListVersions(ctx context.Context, bucket, base string) ([]lease.ObjectVersion, error) {
	if base != s.base {
		s.t.Fatalf("prepared listing escaped generation: %s", base)
	}
	return s.Reader.ListVersions(ctx, bucket, base)
}
func (s preparedReader) GetVersion(context.Context, string, string, string) ([]byte, error) {
	s.t.Fatal("prepared restore must not read a marker")
	return nil, ErrMarker
}
func (s preparedReader) OpenVersion(ctx context.Context, bucket, key, id string) (io.ReadCloser, error) {
	if !strings.HasPrefix(key, s.base) {
		s.t.Fatalf("prepared open escaped generation: %s", key)
	}
	return s.Reader.OpenVersion(ctx, bucket, key, id)
}