package host

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

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// The files live set (J2/F4, Paul's #236 ruling; M10): every moves park
// excludes the objects a live generation's files manifest names, in its own
// region, and the source's are protected on the source until the record
// commits. Plan 5 Task 9a.

// filesManifest writes a hostfiles manifest at <location>/files/g-<gen>/
// naming objects, each "path=key".
func filesManifest(t *testing.T, put func(key string, body []byte), location, gen string, objects ...string) {
	t.Helper()
	m := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: int64(len(objects)), Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}
	for i, o := range objects {
		path, key, _ := strings.Cut(o, "=")
		m.Entries[path] = hostfiles.Entry{Object: key, SHA256: fmt.Sprintf("%064x", i+1), Size: 4, Mode: 0o644, Seq: int64(i + 1)}
	}
	body, err := json.Marshal(m)
	if err != nil {
		t.Fatal(err)
	}
	put(hostfiles.ManifestKey(location+"/files/g-"+gen+"/"), body)
}

// TestParkKeepsTheFilesLiveSet (M10): in the authoritative region the view
// pins the exact versions a live generation's files manifest names, also
// under an older generation's files prefix, so the sweep never makes that
// generation a candidate and its objects stay. Without a manifest nothing
// more is pinned and the older generation is parked. Seen from another
// region nothing is read or pinned.
func TestParkKeepsTheFilesLiveSet(t *testing.T) {
	setup := func(t *testing.T, manifest bool) (*evFixture, park.Engine, []string) {
		f := newEvFixture(t)
		body, err := json.Marshal(map[string]any{"op": evMoveID, "incarnation": f.inc, "host": evHost, "box": "box", "installed_at": evCreated, "manifest_sha256": "x", "snapshot_txid": 1})
		if err != nil {
			t.Fatal(err)
		}
		f.bucket.Put("data", f.location+"/generation.json", body)
		old := []string{f.location + "/files/g-old/objects/0000000000000001-aa", f.genKey("old")}
		for _, k := range old {
			f.bucket.Put("data", k, []byte("old"))
		}
		f.bucket.Put("data", f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb", []byte("cur"))
		if manifest {
			filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID,
				"kept.txt="+old[0], "new.txt="+f.location+"/files/g-"+evMoveID+"/objects/0000000000000002-bb")
		}
		seq := 0
		e := park.Engine{Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
			NewID:     func() string { seq++; return fmt.Sprintf("id%d", seq) },
			View:      movesParkView(f.provider().Records, f.bucket, "data", "ie"),
			Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
		return f, e, old
	}
	t.Run("a live manifest names an older generation's object", func(t *testing.T) {
		f, e, old := setup(t, true)
		v, err := e.View(t.Context(), f.observe(t))
		if err != nil {
			t.Fatal(err)
		}
		ver := f.bucket.LatestID("data", old[0])
		if !v.Pinned[park.Ref{Key: old[0], VersionID: ver}] {
			t.Fatalf("pins %v; want %s@%s", v.Pinned, old[0], ver)
		}
		d, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit)
		if err != nil {
			t.Fatal(err)
		}
		for _, c := range d.Candidates {
			if c.Generation == "old" {
				t.Fatalf("the generation a live manifest names is a candidate: %+v", d.Candidates)
			}
		}
		for _, k := range old {
			if _, ok := f.bucket.Current("data", k); !ok {
				t.Fatalf("%s was parked", k)
			}
		}
		// Seen from a region that is not authoritative, nothing is read.
		other, err := movesParkView(f.provider().Records, f.bucket, "data", "uk")(t.Context(), f.observe(t))
		if err != nil || len(other.Pinned) != 0 || other.MarkerKnown {
			t.Fatalf("the uk view %+v, %v", other, err)
		}
	})
	t.Run("no manifest", func(t *testing.T) {
		f, e, old := setup(t, false)
		if _, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit); err != nil && !errors.Is(err, park.ErrPending) {
			t.Fatal(err)
		}
		for _, k := range old {
			if _, ok := f.bucket.Current("data", k); ok {
				t.Fatalf("%s was not parked", k)
			}
		}
	})
	t.Run("a manifest naming what the location does not hold", func(t *testing.T) {
		for name, key := range map[string]string{"outside the location": "instances/" + evHost + "/elsewhere/x", "missing": "LOC/files/g-old/objects/0000000000000009-zz"} {
			f, e, _ := setup(t, false)
			filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, evMoveID, "x="+strings.Replace(key, "LOC", f.location, 1))
			if _, err := e.View(t.Context(), f.observe(t)); !errors.Is(err, park.ErrUnsafe) {
				t.Fatalf("%s: %v", name, err)
			}
		}
	})
}

// TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit (J2/F4): once the
// target's probe is admitted the target's region is authoritative, but seen
// from the source's region the source stays authoritative until the record
// commits, with its own live manifests pinned; after the commit the source
// pins nothing (P6-D3: the target's manifests name no source object), and
// its location park may run.
func TestFilesLiveSetIsProtectedOnTheSourceUntilTheCommit(t *testing.T) {
	f := newVwFixture(t)
	f.marker(t, "cur", f.inc)
	src := f.location + "/files/g-cur/objects/0000000000000001-aa"
	ver := f.bucket.Put("data", src, []byte("source file"))
	filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "cur", "a.txt="+src)
	landing := func(phase string) hostclaim.Observed {
		return vwObs(&hostclaim.Claim{Holder: f.holder(vwMove, "cur@ie", vwMove+"@uk"), Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: vwMove, Phase: phase, Exec: "uk-box"}})
	}
	f.move(provision.MoveLanding)
	v, err := f.view("ie")(t.Context(), landing(provision.LandProbed))
	if err != nil {
		t.Fatal(err)
	}
	if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: src, VersionID: ver}] {
		t.Fatalf("before the commit the source's view %+v", v)
	}
	if v, err := f.view("uk")(t.Context(), landing(provision.LandProbed)); err != nil || v.AuthoritativeRegion != "uk" {
		t.Fatalf("before the commit the target's view %+v, %v", v, err)
	}
	f.move(provision.MoveCommitted)
	v, err = f.view("ie")(t.Context(), landing(hostclaim.PhaseCommitted))
	if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 {
		t.Fatalf("after the commit the source's view %+v, %v", v, err)
	}
}

// TestLandingResidueParkKeepsTheFilesLiveSet (M10, the landing install's
// park): after the target's probe makes its generation live, a residue
// chore's acknowledgement re-lists the location under the real view. An
// object of the residue generation that the live generation's manifest names
// refuses the acknowledgement and stays; without that reference the same
// object is parked.
func TestLandingResidueParkKeepsTheFilesLiveSet(t *testing.T) {
	t.Parallel()
	for _, named := range []bool{true, false} {
		t.Run(fmt.Sprintf("named %v", named), func(t *testing.T) {
			ctx := context.Background()
			h := newLandingHarness(t)
			h.flip(t)
			h.residue(t)
			h.land(t)
			h.installed(t)
			// The probe's admission: g-<move id> live in uk, its marker
			// published, a files manifest that names an object under the
			// earlier stay's generation (written back after the parking).
			late := cutRoot + "/files/g-old/objects/0000000000000003-cc"
			if _, err := h.vs.PutIfAbsent(ctx, "data-uk", late, []byte("kept")); err != nil {
				t.Fatal(err)
			}
			if named {
				// A new version over the prepared manifest the landing's
				// install built (moves Task 6), as the probe's
				// establishment writes one.
				filesManifest(t, func(k string, b []byte) { putNewVersion(h.f, "data-uk", k, b) }, cutRoot, p1ID, "kept.txt="+late)
			}
			marker := cutRoot + "/generation.json"
			markerVer := h.f.nextVersion()
			body, _ := json.Marshal(map[string]any{"op": p1ID, "incarnation": h.inc, "host": p1Host, "box": "uk-box", "installed_at": h.clock.Now(), "manifest_sha256": "x", "snapshot_txid": 1})
			h.f.store("data-uk", marker, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + markerVer, Size: int64(len(body))}, string(body)})
			h.f.setVersion("data-uk", marker, markerVer)
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
				c.Holder.Generation = p1ID
				c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Session: c.Holder.Session + 1, Box: "uk-box", Generation: p1ID, Predecessor: "cur", Mode: "cold"})
			})
			releaseOwnLease(h.lw, p1Host)
			h.clock.Advance(park.SignatureAge + landingRequestLimit + time.Minute)
			l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
			eng := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
				View: movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk"), Authorize: movesArmAuthorize(allowParkAuthorize, l.authorizeObligation)}
			var runErr error
			for _, o := range h.claim(t).Obligations {
				if o.Kind == "park-generation" && o.Generation == "old" {
					for i := 0; i < 3; i++ {
						runErr = eng.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID})
					}
				}
			}
			_, kept := h.f.buckets["data-uk"][late]
			if named {
				if !errors.Is(runErr, park.ErrUnsafe) || !kept {
					t.Fatalf("a named object: run %v, kept %v", runErr, kept)
				}
				return
			}
			if kept {
				t.Fatalf("an unnamed residue object stayed (run %v)", runErr)
			}
		})
	}
}

// TestReturnResidueParkKeepsTheFilesLiveSet (M10, the rollback return's
// park): once the source's probe admits the return generation the source is
// authoritative again, and its live manifest's objects are pinned, also
// under a pre-move generation; before that admission the target is
// authoritative and the source pins nothing (the dirty return parks the
// pre-move generations, D6).
func TestReturnResidueParkKeepsTheFilesLiveSet(t *testing.T) {
	f := newVwFixture(t)
	f.rec.Move = &provision.MoveState{ID: vwMove, Incarnation: f.inc, From: "ie", To: "uk", State: provision.MoveRollingBack,
		Rollback: &provision.MoveRollback{ID: vwRb, Dirty: true}}
	body, _ := json.Marshal(map[string]any{"op": vwRb, "incarnation": f.inc, "host": vwHost, "box": "ie-box", "installed_at": vwCreated, "manifest_sha256": "x", "snapshot_txid": 1})
	f.bucket.Put("data", f.location+"/generation.json", body)
	pre := f.location + "/files/g-cur/objects/0000000000000001-aa"
	ver := f.bucket.Put("data", pre, []byte("pre-move"))
	filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, vwRb, "a.txt="+pre)
	restoring := func(gen string, admissions ...string) hostclaim.Observed {
		return vwObs(&hostclaim.Claim{Holder: f.holder(gen, admissions...), Op: hostclaim.Op{Kind: hostclaim.KindRestoring, ID: vwMove, Phase: hostclaim.PhaseProbing, Exec: "ie-box"}})
	}
	v, err := f.view("ie")(t.Context(), restoring(vwRb, "cur@ie", vwMove+"@uk", vwRb+"@ie"))
	if err != nil {
		t.Fatal(err)
	}
	if v.AuthoritativeRegion != "ie" || !v.Pinned[park.Ref{Key: pre, VersionID: ver}] {
		t.Fatalf("after the source's admission %+v", v)
	}
	v, err = f.view("ie")(t.Context(), restoring(vwMove, "cur@ie", vwMove+"@uk"))
	if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 {
		t.Fatalf("before the source's admission %+v, %v", v, err)
	}
}

// TestRestoringKeepsTheSourcesFilesLiveSetUntilTheReturn (J2/F4; L18 joint
// round, finding 1): a park of an older source generation is queued in the
// source's region before the target's probe is admitted; then the target is
// admitted and the landing is rolled back. While the target releases
// (restoring/releasing, released) the newest admission is the target's, but
// a clean return restores the source's retained generation, so seen from the
// source's region the source stays authoritative: its marker and files pins
// hold, the chore is refused and nothing the retained manifest names is
// retired. A dirty release is protected the same way (a declared loss turns
// it clean). The target's region still names the target. From the return CAS
// on the newest admission decides, as before: a clean return is admitted in
// the source, and a dirty one names the target until the source's probe is
// admitted, so the source pins nothing and the chore parks the pre-move
// generation (D6).
func TestRestoringKeepsTheSourcesFilesLiveSetUntilTheReturn(t *testing.T) {
	for _, dirty := range []bool{false, true} {
		t.Run(fmt.Sprintf("dirty %v", dirty), func(t *testing.T) {
			ctx := t.Context()
			f := newVwFixture(t)
			markerVer, snapshot := f.marker(t, "cur", f.inc)
			snapKey := f.location + "/g-cur/0009/0000000000000001-0000000000000005.ltx"
			old := map[string]string{}
			for _, k := range []string{f.location + "/files/g-old/objects/0000000000000001-aa", f.location + "/g-old/0000/db"} {
				old[k] = f.bucket.Put("data", k, []byte("old"))
			}
			named := f.location + "/files/g-old/objects/0000000000000001-aa"
			filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "cur", "a.txt="+named)
			f.move(provision.MoveLanding)

			// The chore, queued in the source's region at landing, before
			// the target's probe is admitted.
			claims := release.NewDirStore(t.TempDir())
			chore := hostclaim.Obligation{ID: "park-old", Op: vwMove, Kind: "park-generation", Region: "ie", Owner: "region:ie", Generation: "old", Retirement: "ret-1",
				Park: &hostclaim.ParkScope{Location: f.location, Incarnation: f.inc, EvidenceAt: evAt, RequestLimit: hostbox.RequestLimit}}
			update := func(step func(*hostclaim.Claim) error) {
				t.Helper()
				obs, err := hostclaim.Read(ctx, claims, vwHost)
				if err != nil {
					t.Fatal(err)
				}
				if _, err := hostclaim.Update(ctx, claims, obs, step); err != nil {
					t.Fatal(err)
				}
			}
			update(func(c *hostclaim.Claim) error {
				c.Host, c.Holder = vwHost, f.holder("cur", "cur@ie")
				c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: vwMove, Phase: provision.LandProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"},
					Intent:  map[string]string{"source": "ie-box", "target": "uk-box", "from": "ie", "to": "uk"},
					Journal: map[string]string{hostclaim.JournalSourceStamp: "3", hostclaim.JournalSrcGen: "cur"}}
				c.Obligations = []hostclaim.Obligation{chore}
				return nil
			})
			// The target's probe's admission, then the rollback.
			update(func(c *hostclaim.Claim) error {
				c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 2, Box: "uk-box", Generation: vwMove, Predecessor: "cur", Mode: "cold", Region: "uk", PredecessorRegion: "ie"})
				c.Holder.Generation = vwMove
				return nil
			})
			f.rec.Move.State = provision.MoveRollingBack
			f.rec.Move.Rollback = &provision.MoveRollback{ID: vwRb, TargetPhase: provision.RbTargetReleasing}
			update(hostclaim.RollBack(vwMove, vwRb))

			seq := 0
			engine := park.Engine{Store: f.bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
				NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) }, View: f.view("ie"), Authorize: allowParkAuthorize}
			observe := func() hostclaim.Observed {
				t.Helper()
				obs, err := hostclaim.Read(ctx, claims, vwHost)
				if err != nil {
					t.Fatal(err)
				}
				return obs
			}
			intact := func() bool {
				for k, ver := range old {
					if _, ok := f.bucket.Current("data", k); !ok || f.bucket.LatestID("data", k) != ver {
						return false
					}
				}
				return true
			}
			protectedOnSource := func(when string) {
				t.Helper()
				if err := engine.Run(ctx, park.Work{Host: vwHost, ID: chore.ID, Op: vwMove}); !errors.Is(err, park.ErrUnsafe) {
					t.Fatalf("%s: the source's park of a generation its retained manifest names: %v", when, err)
				}
				if !intact() {
					t.Fatalf("%s: an object the retained manifest names was retired", when)
				}
				v, err := f.view("ie")(ctx, observe())
				if err != nil {
					t.Fatalf("%s: the source's view: %v", when, err)
				}
				if v.AuthoritativeRegion != "ie" || v.MarkerGeneration != "cur" || v.Marker != (park.Ref{Key: f.location + "/generation.json", VersionID: markerVer}) ||
					!v.Pinned[park.Ref{Key: named, VersionID: old[named]}] || !v.Pinned[park.Ref{Key: snapKey, VersionID: snapshot}] {
					t.Fatalf("%s: the source's view %+v", when, v)
				}
			}
			targetView := func(when, want string) {
				t.Helper()
				if v, err := f.view("uk")(ctx, observe()); err != nil || v.AuthoritativeRegion != want {
					t.Fatalf("%s: the target's view %+v, %v; want %s", when, v, err, want)
				}
			}

			protectedOnSource("releasing")
			targetView("releasing", "uk")

			f.rec.Move.Rollback.TargetPhase, f.rec.Move.Rollback.Dirty = provision.RbTargetReleased, dirty
			update(hostclaim.TargetReleased(vwMove, "uk-box", "uk", "stopped-target", hostclaim.ParkScope{Location: f.location, Incarnation: f.inc, EvidenceAt: evAt, RequestLimit: hostbox.RequestLimit},
				func() string { seq++; return fmt.Sprintf("id%d", seq) }))
			if op := observe().Claim.Op; op.Phase != hostclaim.PhaseReleased || op.Exec != "" {
				t.Fatalf("the release: %+v", op)
			}
			protectedOnSource("released")
			targetView("released", "uk")

			// The source's return CAS.
			update(hostclaim.ReturnToSource(vwMove, "ie-box", !dirty, hostclaim.ReturnFence{}))
			if !dirty {
				protectedOnSource("after the clean return")
				targetView("after the clean return", "ie")
				return
			}
			v, err := f.view("ie")(ctx, observe())
			if err != nil || v.AuthoritativeRegion != "uk" || len(v.Pinned) != 0 || v.MarkerKnown {
				t.Fatalf("after the dirty return the source's view %+v, %v", v, err)
			}
			targetView("after the dirty return", "uk")
			if err := engine.Run(ctx, park.Work{Host: vwHost, ID: chore.ID, Op: vwMove}); err != nil && !errors.Is(err, park.ErrPending) {
				t.Fatalf("the dirty return's park of the pre-move generation: %v", err)
			}
			for k := range old {
				if _, ok := f.bucket.Current("data", k); ok {
					t.Fatalf("%s was not parked after the dirty return", k)
				}
			}
		})
	}
}

// TestFilesLiveSetFollowsTheLiveChain (fix round 1, I2): with the marker
// lagging, park.protected treats the whole unfinished chain g3 -> g2 -> g1
// as live, and the view reads every one of those generations' manifests. An
// object only g2's manifest names, under the older generation g0, is pinned
// and survives the sweep's park.
func TestFilesLiveSetFollowsTheLiveChain(t *testing.T) {
	f := newEvFixture(t)
	f.setClaim(t, func(c *hostclaim.Claim) {
		c.Holder.Generation = "g3"
		c.Holder.Admissions = []hostclaim.AdmissionRecord{
			{Session: 1, Box: "box", Generation: "g1", Mode: "cold", Region: "ie"},
			{Session: 2, Box: "box", Generation: "g2", Predecessor: "g1", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
			{Session: 3, Box: "box", Generation: "g3", Predecessor: "g2", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
		}
		c.Holder.Stamp, c.Holder.Session = 3, 3
	})
	body, err := json.Marshal(map[string]any{"op": "g1", "incarnation": f.inc, "host": evHost, "box": "box", "installed_at": evCreated, "manifest_sha256": "x", "snapshot_txid": 1})
	if err != nil {
		t.Fatal(err)
	}
	f.bucket.Put("data", f.location+"/generation.json", body)
	only := f.location + "/files/g-g0/objects/0000000000000001-aa"
	ver := f.bucket.Put("data", only, []byte("named by g2 alone"))
	other := f.location + "/files/g-g0/objects/0000000000000002-bb"
	f.bucket.Put("data", other, []byte("named by nobody"))
	filesManifest(t, func(k string, b []byte) { f.bucket.Put("data", k, b) }, f.location, "g2", "a.txt="+only)
	seq := 0
	e := park.Engine{Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{}, Now: func() time.Time { return evNow },
		NewID:     func() string { seq++; return fmt.Sprintf("id%d", seq) },
		View:      movesParkView(f.provider().Records, f.bucket, "data", "ie"),
		Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
	v, err := e.View(t.Context(), f.observe(t))
	if err != nil {
		t.Fatal(err)
	}
	if !v.Pinned[park.Ref{Key: only, VersionID: ver}] {
		t.Fatalf("pins %v; want g2's object %s", v.Pinned, only)
	}
	if _, err := e.Sweep(t.Context(), evHost, f.inc, evAt, hostbox.RequestLimit); err != nil && !errors.Is(err, park.ErrPending) {
		t.Fatal(err)
	}
	if got, ok := f.bucket.Current("data", only); !ok || string(got) != "named by g2 alone" {
		t.Fatalf("g2's object did not survive the park: %q %v", got, ok)
	}
}

// putFilesGeneration writes, in bucket, a files object of gen at location
// and gen's manifest naming it, returning the object's key.
func putFilesGeneration(t *testing.T, vs *fakeVersioned, bucket, location, gen string) string {
	t.Helper()
	ctx := context.Background()
	obj := location + "/files/g-" + gen + "/objects/0000000000000001-aa"
	if _, err := vs.PutIfAbsent(ctx, bucket, obj, []byte("a live file")); err != nil {
		t.Fatal(err)
	}
	// gen's manifest is a new version: a landing's or a return's install
	// may already have built gen's prepared manifest (moves Task 6), and the
	// probe's establishment writes the live one over it.
	filesManifest(t, func(k string, b []byte) { putNewVersion(vs.f, bucket, k, b) }, location, gen, "a.txt="+obj)
	return obj
}

// putNewVersion writes body as a new current version of key in bucket,
// whether or not the key holds one.
func putNewVersion(f *fakeMoveMirror, bucket, key string, body []byte) {
	ver := f.nextVersion()
	f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
	f.setVersion(bucket, key, ver)
}

// archivedUnder reports whether bucket holds a parked copy of key, a key
// under location (park's destination: parked/<inc>/<host>/<retirement>/
// instances/<key below the location>@<version>).
func archivedUnder(f *fakeMoveMirror, bucket, inc, location, key string) bool {
	tail := strings.TrimPrefix(key, location+"/")
	for k := range f.buckets[bucket] {
		if strings.HasPrefix(k, "parked/"+inc+"/") && strings.Contains(k, "/instances/"+tail+"@") {
			return true
		}
	}
	return false
}

// admitTarget is the target's probe admission on the claim (as Plan 3's
// probe makes it): holder.generation is the move's, admitted in uk.
func admitTarget(t *testing.T, h *landingHarness) {
	t.Helper()
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Session: c.Holder.Session + 1, Box: "uk-box", Generation: p1ID, Predecessor: c.Holder.Generation, Mode: "cold"})
		c.Holder.Generation = p1ID
	})
}

// TestSourceCleanupKeepsTheFilesLiveSetUntilTheCommit (fix round 1, I3; the
// source cleanup, J2/F4): the source's location park runs through the real
// view and engine. Before the record commits, with the target's probe
// admitted, a park of the source's location in its own region is refused
// and the source's files stay byte for byte. After the commit the source's
// finalization runs that park and the files are retired, archived under
// parked/.
func TestSourceCleanupKeepsTheFilesLiveSetUntilTheCommit(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newFinalHarness(t)
	h.advanceToProbed(t)
	admitTarget(t, h.landingHarness)
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	obj := putFilesGeneration(t, h.vs, "data-ie", location, "cur")
	view := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")

	// Before the commit: the same scope the commit queues, run by the engine.
	rec := h.read(t, p1Host)
	scope, err := sourceLocationParkScope(rec, *h.claim(t).Holder)
	if err != nil {
		t.Fatal(err)
	}
	early := hostclaim.Obligation{ID: obligationID(), Op: p1ID, Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: obligationID(), Park: &scope}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Obligations = append(c.Obligations, early) })
	e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-ie", Region: "ie", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
		View: view, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
	if err := e.Run(ctx, park.Work{Host: p1Host, ID: early.ID, Op: p1ID}); !errors.Is(err, park.ErrUnsafe) {
		t.Fatalf("a park of the source before the commit: %v", err)
	}
	if body, ok := h.f.buckets["data-ie"][obj]; !ok || body.body != "a live file" {
		t.Fatalf("the source's file before the commit: %+v %v", body, ok)
	}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		kept := c.Obligations[:0]
		for _, o := range c.Obligations {
			if o.ID != early.ID {
				kept = append(kept, o)
			}
		}
		c.Obligations = kept
	})

	// The commit, then the source's finalization through the real view.
	h.commitRecord(t)
	h.commitClaim(t)
	h.cfg.ParkView = view
	h.sourcePass(t)
	if m := h.read(t, p1Host).Move; m.SourceParkedAt == nil {
		t.Fatalf("the source did not finalize: %+v", m)
	}
	if _, ok := h.f.buckets["data-ie"][obj]; ok {
		t.Fatal("the source's file was not retired after the commit")
	}
	if !archivedUnder(h.f, "data-ie", h.inc, location, obj) {
		t.Fatal("the source's file was not archived")
	}
}

// TestRollbackReturnCleanupKeepsTheFilesLiveSet (fix round 1, I3; the
// rollback return's cleanup): the target's region location park, queued
// when the target released, runs through the real view and engine. While
// the target's admitted generation is still the holder's (before the return
// names the source's again) the target region is authoritative and the park
// is refused, the target's files kept; once the source has taken the host
// back, the target's cleaning runs it and the files are retired.
func TestRollbackReturnCleanupKeepsTheFilesLiveSet(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	admitTarget(t, h.landingHarness)
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	body, err := json.Marshal(map[string]any{"op": p1ID, "incarnation": h.inc, "host": p1Host, "box": "uk-box", "installed_at": h.clock.Now(), "manifest_sha256": "x", "snapshot_txid": 1})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := h.vs.PutIfAbsent(ctx, "data-uk", location+"/generation.json", body); err != nil {
		t.Fatal(err)
	}
	obj := putFilesGeneration(t, h.vs, "data-uk", location, p1ID)
	view := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-uk", "uk")
	h.parkView = view
	h.rollBack(t)
	h.rollbackPass(t)
	var chore *hostclaim.Obligation
	for _, o := range h.claim(t).Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" && o.Park != nil && o.Park.Location == location {
			chore = &o
		}
	}
	if chore == nil {
		t.Fatalf("the release queued no target location park: %+v", h.claim(t).Obligations)
	}
	e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
		View: view, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
	if err := e.Run(ctx, park.Work{Host: p1Host, ID: chore.ID, Op: p1ID}); !errors.Is(err, park.ErrUnsafe) {
		t.Fatalf("a park of the target while its generation is live: %v", err)
	}
	if b, ok := h.f.buckets["data-uk"][obj]; !ok || b.body != "a live file" {
		t.Fatal("the target's file was taken before the return")
	}

	// The source takes the host back: the return names its generation again
	// and it recovers; the target's cleaning then runs the park.
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = "cur" })
	h.recovered(t)
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned {
		t.Fatalf("the target did not clean: %+v", rb)
	}
	if _, ok := h.f.buckets["data-uk"][obj]; ok {
		t.Fatal("the target's file was not retired by the cleaning")
	}
	if !archivedUnder(h.f, "data-uk", h.inc, location, obj) {
		t.Fatal("the target's file was not archived")
	}
}

// returnPreMove lists the pre-move generation's objects at the source's
// location, with their bodies.
func returnPreMove(t *testing.T, h *returnHarness, location string) map[string]string {
	t.Helper()
	out := map[string]string{}
	for k, o := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, location+"/g-cur/") {
			out[k] = o.body
		}
	}
	if len(out) == 0 {
		t.Fatal("fixture: no pre-move generation at the source's location")
	}
	return out
}

// TestReturnResidueParkRunsThroughTheInstall (fix round 2, I3; the source's
// return-residue park, movesinstall.go parking): with the record's evidence
// that the target admitted this move (its probe's admission of the move's
// prepared generation), a dirty return's parking runs its residue chores
// through the installer's own engine with the real view: the pre-move
// generation is archived under parked/ and retired from the source's
// location, and the return moves on.
func TestReturnResidueParkRunsThroughTheInstall(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, false)
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	pre := returnPreMove(t, h, location)
	h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
	if _, err := h.install(t); err != nil {
		t.Fatalf("the return with the target's admission: %v", err)
	}
	if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase == provision.RbSourceParking {
		t.Fatalf("the return is still parking: %+v", rb)
	}
	for k := range pre {
		if _, ok := h.f.buckets["data-ie"][k]; ok {
			t.Fatalf("%s was not retired", k)
		}
		if !archivedUnder(h.f, "data-ie", h.inc, location, k) {
			t.Fatalf("%s was not archived", k)
		}
	}
	if j := h.returnJournal(t); j.Steps[installStepResidueKept] != "" {
		t.Fatalf("a return with admission evidence journalled the no-op: %q", j.Steps[installStepResidueKept])
	}
}

// TestDirtyReturnWithoutAdmissionKeepsThePreMoveGeneration (fix round 3,
// ruling (a)): the record carries no evidence that the target admitted this
// move (the landing never reached probing, and the claim holds no admission
// of the move's generation). The pre-move generation is still the source's
// live state, so the return's parking is a journalled no-op: through the
// real view and engine the return completes, the pre-move bytes are intact
// and no residue park is queued.
func TestDirtyReturnWithoutAdmissionKeepsThePreMoveGeneration(t *testing.T) {
	t.Parallel()
	h := newReturnHarnessWith(t, false, false)
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	pre := returnPreMove(t, h, location)
	h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
	if _, err := h.install(t); err != nil {
		t.Fatalf("the return without admission evidence: %v", err)
	}
	if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceRestored {
		t.Fatalf("the return did not complete its install: %+v", rb)
	}
	for k, body := range pre {
		if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
			t.Fatalf("%s changed", k)
		}
	}
	j := h.returnJournal(t)
	if got := j.Steps[installStepResidueKept]; !strings.Contains(got, "never admitted") {
		t.Fatalf("the journal's no-op line: %q", got)
	}
	if got := j.Steps[landStepResidue]; got != "[]" {
		t.Fatalf("the residue list %q, want none", got)
	}
	for _, o := range h.claim(t).Obligations {
		if o.Kind == "park-generation" {
			t.Fatalf("a residue park was queued: %+v", o)
		}
	}
}

// TestReturnResidueParkWaitsOnATransientRefusal (fix round 3): with the
// target's admission recorded, a view refusal never turns the parking into
// the no-op. The return waits at parking with the pre-move bytes intact and
// no no-op line, and parks once the view answers.
func TestReturnResidueParkWaitsOnATransientRefusal(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, false)
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	pre := returnPreMove(t, h, location)
	real := movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
	down := true
	h.rcfg.ParkView = func(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
		if down {
			return park.View{}, fmt.Errorf("%w: the record read timed out", park.ErrUnsafe)
		}
		return real(ctx, obs)
	}
	if _, err := h.install(t); err == nil {
		t.Fatal("the return went on through a view refusal")
	}
	if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceParking {
		t.Fatalf("the return left parking: %+v", rb)
	}
	if j := h.returnJournal(t); j.Steps[installStepResidueKept] != "" {
		t.Fatalf("a view refusal became the no-op: %q", j.Steps[installStepResidueKept])
	}
	for k, body := range pre {
		if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
			t.Fatalf("%s changed during the refusal", k)
		}
	}
	down = false
	if _, err := h.install(t); err != nil {
		t.Fatalf("after the refusal: %v", err)
	}
	for k := range pre {
		if _, ok := h.f.buckets["data-ie"][k]; ok {
			t.Fatalf("%s was not retired once the view answered", k)
		}
	}
}

// TestDirtyReturnAtProbingWithoutAdmissionKeepsTheResidue (fix round 4): the
// record enters probing before the probe runs (movesland.go), so a landing
// that crashed or failed at probing before its admission and was then
// rolled back dirty is not evidence that the target admitted. With the
// holder holding no admission of the move's generation, the return keeps
// the pre-move generation and completes.
func TestDirtyReturnAtProbingWithoutAdmissionKeepsTheResidue(t *testing.T) {
	t.Parallel()
	h := newReturnHarnessWith(t, false, false)
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.LandingPhase = provision.LandProbing
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	location := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
	pre := returnPreMove(t, h, location)
	h.rcfg.ParkView = movesParkView(MovesRecordsFrom(h.s), h.vs, "data-ie", "ie")
	if _, err := h.install(t); err != nil {
		t.Fatalf("the return at probing without an admission: %v", err)
	}
	if rb := h.read(t, p1Host).Move.Rollback; rb.SourcePhase != provision.RbSourceRestored {
		t.Fatalf("the return did not complete its install: %+v", rb)
	}
	for k, body := range pre {
		if got, ok := h.f.buckets["data-ie"][k]; !ok || got.body != body {
			t.Fatalf("%s changed", k)
		}
	}
	if got := h.returnJournal(t).Steps[installStepResidueKept]; !strings.Contains(got, "never admitted") {
		t.Fatalf("the journal's no-op line: %q", got)
	}
}