package console

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
	"github.com/carlosframework/platform/internal/svcrole"
)

// The offline operator move's doors, driven in-process against the same
// console fixture the move doors' tests use. The boxes' side is played by
// reports written at control/offline-moves/ exactly where the real boxes
// write them (moves.OfflineReportKey); the record moves only through
// provision's offline transitions.

const offlineHost = "acme.shop.example.com"

// offlineManifest is the enabled manifest of the app shop, whose instances
// live under shop.example.com.
func offlineManifest() provision.Manifest {
	return provision.Manifest{Schema: 1, App: "shop", Domain: "shop.example.com",
		Pool: "shop", Channel: "stable", Regions: []string{"ie"}, Seed: []string{"seed"}}
}

// offlineClaimFor is claimFor for an instance of the app shop: it writes the
// host's serving claim for the record's own incarnation.
func offlineClaimFor(t *testing.T, fx *consoleFixture, host, box, region string) {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), fx.rel, "bab", "shop", host)
	if err != nil {
		t.Fatal(err)
	}
	inc, err := hostclaim.Incarnation("bab", "shop", host, rec.CreatedAt)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(context.Background(), fx.rel, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
		c.Holder = &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "shop", host), Account: "bab", App: "shop",
			Incarnation: inc, CreatedAt: rec.CreatedAt, Location: "instances/" + host, Stamp: 1}
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: box}
		c.Touched = hostclaim.Touched{Complete: true, Regions: []string{region}, Boxes: []string{box}}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
}

// offlineBox is one box the fleet publishes: an edges file and a status
// snapshot.
type offlineBox struct {
	Box, Placement, Bucket string
	NoCap                  bool     // no offline-move-1 (nor offline-stage-1)
	NoStageCap             bool     // no offline-stage-1
	ClaimsOn               bool     // its status report says claims are on
	NoMetrics              bool     // its status report carries no metrics
	NotSharedPool          bool     // serves_shared_pool false
	NoHibernation          bool     // hibernation_configured false
	Hosts                  []string // hosts its edges file lists
	FP                     string   // secrets_key_fp; testKeyFP when empty
}

func ieOff(box string) offlineBox { return offlineBox{Box: box, Placement: "ie", Bucket: "data-ie"} }
func ukOff(box string) offlineBox { return offlineBox{Box: box, Placement: "uk", Bucket: "data-uk"} }

type offlineWorld struct {
	t     *testing.T
	fx    *consoleFixture
	token string
	base  string // .../instances/<host>/move/offline
	id    string // the started move's id
}

// newOfflineWorld is an operator console with one hibernating instance in
// region ie and the fleet published.
func newOfflineWorld(t *testing.T, boxes ...offlineBox) *offlineWorld {
	t.Helper()
	return newOfflineWorldIn(t, "", "ie", boxes...)
}

// newOfflineWorldIn is newOfflineWorld with the console's home named and the
// instance created in the given region.
func newOfflineWorldIn(t *testing.T, home, region string, boxes ...offlineBox) *offlineWorld {
	t.Helper()
	fx, token := instanceFixture(t, "code-offline-"+strings.ReplaceAll(t.Name(), "/", "-"))
	if err := fx.st.SetOperator("alice@keymail.dev", true); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	fx.srv.cfg.HomeRegion = home
	m := offlineManifest()
	m.Regions = nil
	m.Pool = ""
	enableInstances(t, fx, "shop", m)
	u := instancesURL(fx, "bab", "shop")
	if c, _, out := rawReq(t, http.MethodPost, u, token, `{"host":"`+offlineHost+`","region":"`+region+`"}`); c != http.StatusCreated {
		t.Fatalf("create = %d %s", c, out)
	}
	w := &offlineWorld{t: t, fx: fx, token: token, base: u + "/" + offlineHost + "/move/offline"}
	w.fleet(boxes...)
	return w
}

// putRaw writes a bucket object as a fixture, replacing what is there.
func offlinePut(t *testing.T, rel release.Store, key string, body []byte) {
	t.Helper()
	_, etag, err := rel.Get(context.Background(), key)
	if err != nil {
		etag = ""
	}
	if _, err := rel.Put(context.Background(), key, body, etag); err != nil {
		t.Fatal(err)
	}
}

// fleet publishes the boxes, replacing any earlier publication of the same
// box.
func (w *offlineWorld) fleet(boxes ...offlineBox) {
	w.t.Helper()
	w.fx.srv.cfg.SecretsPub = testKeyPub
	for _, b := range boxes {
		aws := "eu-west-1"
		if b.Placement == "uk" {
			aws = "eu-west-2"
		}
		caps := []string{"region-ha-1"}
		if !b.NoCap {
			caps = append(caps, offlineMoveCapability)
			if !b.NoStageCap {
				caps = append(caps, offlineStageCapability)
			}
		}
		hosts := map[string]any{}
		for _, h := range b.Hosts {
			hosts[h] = map[string]any{"role": "home"}
		}
		fp := b.FP
		if fp == "" {
			fp = testKeyFP
		}
		body, err := json.Marshal(map[string]any{
			"v": 1, "box": b.Box, "box_id": b.Box, "eip": "1.2.3.4", "region": aws, "placement": b.Placement,
			"data_bucket": b.Bucket, "data_bucket_region": aws, "hibernation_configured": !b.NoHibernation, "serves_shared_pool": !b.NotSharedPool,
			"accepts_inbound": true, "secrets_key_fp": fp, "capabilities": caps, "hosts": hosts,
		})
		if err != nil {
			w.t.Fatal(err)
		}
		offlinePut(w.t, w.fx.rel, edgesFilePrefix+b.Box+".json", body)
		sb := status.Box{Label: b.Box, GeneratedAt: time.Now()}
		if !b.NoMetrics {
			sb.Metrics = &status.BoxMetrics{HostClaimEnabled: b.ClaimsOn}
		}
		if err := status.WriteBox(context.Background(), w.fx.rel, sb); err != nil {
			w.t.Fatal(err)
		}
	}
}

func (w *offlineWorld) req(method, path, body string) (int, string) {
	w.t.Helper()
	code, _, out := rawReq(w.t, method, w.base+path, w.token, body)
	return code, out
}

// bucketState is every object in the bucket, for the "a refusal writes
// nothing" check.
func bucketState(t *testing.T, rel release.Store) map[string]string {
	t.Helper()
	keys, err := rel.List(context.Background(), "")
	if err != nil {
		t.Fatal(err)
	}
	out := map[string]string{}
	for _, k := range keys {
		b, _, err := rel.Get(context.Background(), k)
		if err != nil {
			t.Fatal(err)
		}
		out[k] = string(b)
	}
	return out
}

// refused posts and wants exactly this refusal, with nothing written.
func (w *offlineWorld) refused(method, path, body string, wantCode int, wantMsg string) {
	w.t.Helper()
	before := bucketState(w.t, w.fx.rel)
	code, out := w.req(method, path, body)
	if code != wantCode || errorText(out) != wantMsg {
		w.t.Fatalf("%s %s = %d %s\nwant %d %q", method, path, code, out, wantCode, wantMsg)
	}
	after := bucketState(w.t, w.fx.rel)
	if len(before) != len(after) {
		w.t.Fatalf("%s %s: a refusal changed the bucket: %d objects before, %d after", method, path, len(before), len(after))
	}
	for k, v := range before {
		if after[k] != v {
			w.t.Fatalf("%s %s: a refusal changed %s", method, path, k)
		}
	}
}

func (w *offlineWorld) view(code int, out string) offlineView {
	w.t.Helper()
	var v offlineView
	if err := json.Unmarshal([]byte(out), &v); err != nil {
		w.t.Fatalf("%v in %s", err, out)
	}
	_ = code
	return v
}

// ok posts and wants a 202 whose view has the state.
func (w *offlineWorld) ok(path, body, wantState string) offlineView {
	w.t.Helper()
	code, out := w.req(http.MethodPost, path, body)
	if code != http.StatusAccepted {
		w.t.Fatalf("POST %s = %d %s", path, code, out)
	}
	v := w.view(code, out)
	if v.State != wantState {
		w.t.Fatalf("POST %s: state %q, want %q: %s", path, v.State, wantState, out)
	}
	return v
}

func (w *offlineWorld) status() offlineView {
	w.t.Helper()
	code, out := w.req(http.MethodGet, "", "")
	if code != http.StatusOK {
		w.t.Fatalf("GET status = %d %s", code, out)
	}
	return w.view(code, out)
}

func (w *offlineWorld) record() *provision.Record {
	w.t.Helper()
	rec, err := provision.GetRecord(context.Background(), w.fx.rel, "bab", "shop", offlineHost)
	if err != nil {
		w.t.Fatal(err)
	}
	return rec
}

// edit applies record transitions as a fixture, through the record writer.
func (w *offlineWorld) edit(muts ...func(*provision.Record) error) {
	w.t.Helper()
	for _, m := range muts {
		if _, err := provision.SetRecordMove(context.Background(), w.fx.rel, "bab", "shop", offlineHost, m); err != nil {
			w.t.Fatal(err)
		}
	}
}

func serialBytes(t *testing.T, rel release.Store) string {
	t.Helper()
	b, _, err := rel.Get(context.Background(), "control/serial")
	if err != nil {
		return ""
	}
	return string(b)
}

// start runs the start door and keeps the id.
func (w *offlineWorld) start() offlineView {
	w.t.Helper()
	before := serialBytes(w.t, w.fx.rel)
	v := w.ok("", `{"region":"uk"}`, provision.OfflineStopping)
	if v.ID == "" {
		w.t.Fatalf("start answered no id: %+v", v)
	}
	if serialBytes(w.t, w.fx.rel) == before {
		w.t.Fatal("start did not bump the serial")
	}
	w.id = v.ID
	return v
}

// report writes a box's report where the real box writes it, and returns
// its bytes.
func (w *offlineWorld) report(role, box, phase string, edit func(*moves.OfflineReport)) []byte {
	w.t.Helper()
	r := moves.OfflineReport{V: moves.OfflineReportVersion, Op: w.id, Host: offlineHost, Box: box, Role: role, Phase: phase,
		At: time.Now().UTC(), Held: true, Fenced: true, Quiesced: true}
	if edit != nil {
		edit(&r)
	}
	body, err := json.Marshal(r)
	if err != nil {
		w.t.Fatal(err)
	}
	key := moves.OfflineReportKey(offlineHost, w.id, role, box)
	if key == "" {
		w.t.Fatalf("no report key for %s/%s", role, box)
	}
	offlinePut(w.t, w.fx.rel, key, body)
	return body
}

// stopped writes every source box's report as a stopped box writes it.
func (w *offlineWorld) stopped(edit func(box string, r *moves.OfflineReport), boxes ...string) {
	w.t.Helper()
	for _, b := range boxes {
		w.report(moves.OfflineRoleSource, b, moves.OfflinePhaseStopped, func(r *moves.OfflineReport) {
			if edit != nil {
				edit(b, r)
			}
		})
	}
}

func okChecks(owner string) map[string]moves.Check {
	out := map[string]moves.Check{}
	for _, it := range moves.OfflineOwned(owner) {
		out[it.ID] = moves.Check{OK: true, Reason: "ok"}
	}
	return out
}

// sealerChecks is what a sealer's report carries when the seal went well:
// every item the source side reports.
func sealerChecks() map[string]moves.Check {
	out := map[string]moves.Check{}
	for _, it := range moves.OfflineSourced() {
		out[it.ID] = moves.Check{OK: true, Reason: "ok"}
	}
	return out
}

// putSeal writes seal.json as the sealer does: once and create-only when it
// is not there, and replaced only by a test that tampers with it.
func (w *offlineWorld) putSeal(sealer string, edit func(*moves.OfflineReport)) []byte {
	w.t.Helper()
	r := moves.OfflineReport{V: moves.OfflineReportVersion, Op: w.id, Host: offlineHost, Box: sealer, Role: moves.OfflineRoleSealer,
		Phase: moves.OfflinePhaseSealed, At: time.Now().UTC(), Items: sealerChecks(), FinalTXID: 7, TombstoneOwner: moves.TombstoneOwner(w.id)}
	if edit != nil {
		edit(&r)
	}
	body, err := json.Marshal(r)
	if err != nil {
		w.t.Fatal(err)
	}
	key := moves.OfflineSealKey(offlineHost, w.id)
	if _, _, err := w.fx.rel.Get(context.Background(), key); err != nil {
		if _, err := w.fx.rel.Put(context.Background(), key, body, ""); err != nil {
			w.t.Fatal(err)
		}
	} else {
		offlinePut(w.t, w.fx.rel, key, body)
	}
	return body
}

// sealed plays the sealer: the seal, its living report, and the transition
// that records where the seal is. It returns the seal's bytes.
func (w *offlineWorld) sealed(sealer string) []byte {
	w.t.Helper()
	seal := w.putSeal(sealer, nil)
	w.sealerFresh(sealer)
	w.edit(provision.OfflineSealedBy(w.id, sealer, provision.MoveSource{FinalTXID: 7, Evidence: moves.OfflineSealEvidence(offlineHost, w.id, seal)}))
	return seal
}

// landed plays the target: its landing report and the transition.
func (w *offlineWorld) landed(target string) {
	w.t.Helper()
	w.report(moves.OfflineRoleTarget, target, moves.OfflinePhaseLanded, func(r *moves.OfflineReport) {
		r.Items = okChecks(moves.OfflineOwnerLanding)
	})
	w.edit(provision.OfflineLandedBy(w.id, "uk"))
}

// toLanded drives a started move through seal and flip to offline-landed
// with fresh reports from the one ie box and the one uk box.
func (w *offlineWorld) toLanded() {
	w.t.Helper()
	w.start()
	w.stopped(nil, "ie-box")
	w.ok("/seal", "", provision.OfflineSealing)
	w.sealed("ie-box")
	// The sealer rewrites its living report on every pass, so by the time
	// the operator flips it is not the bytes it was at the seal. The seal is.
	w.sealerFresh("ie-box")
	w.stopped(nil, "ie-box")
	w.ok("/flip", "", provision.OfflineLanding)
	w.landed("uk-box")
	w.stopped(nil, "ie-box")
	w.sealerFresh("ie-box")
}

// sealerFresh rewrites the sealer's report as the sealer does every pass.
func (w *offlineWorld) sealerFresh(sealer string) {
	w.t.Helper()
	w.report(moves.OfflineRoleSealer, sealer, moves.OfflinePhaseSealed, func(r *moves.OfflineReport) {
		r.Items = sealerChecks()
		r.TombstoneOwner = moves.TombstoneOwner(w.id)
		r.TombstoneUntil = time.Now().Add(moves.TombstoneTTL)
	})
}

var oneEach = []offlineBox{ieOff("ie-box"), ukOff("uk-box")}

const (
	msgO20 = "offline moves are not available while host claims or Phase 1 moves are on"
	msgO29 = offlineHost + " has a host claim. offline moves are only for hosts without one"
)

func msgO21(box string) string {
	return box + " cannot take part in an offline move. it runs an older release or has host claims on"
}
func msgO25(state string) string {
	return "this move is " + state + ". run carlos move offline status --host " + offlineHost + " to see what it waits for"
}
func msgMC10(box string) string {
	return "Waiting for " + box + " in ie to confirm it has stopped " + offlineHost + ". " + offlineHost + " is offline."
}
func msgO178(box string) string {
	return "the report from " + box + " is not current. it reports again within a minute. try again then"
}
func msgO31(box string) string {
	return box + " still runs a process for " + offlineHost + ". wait for it to stop, then flip"
}

var offlineRoutes = []struct{ method, path, body string }{
	{http.MethodGet, "", ""},
	{http.MethodPost, "", `{"region":"uk"}`},
	{http.MethodPost, "/stage", `{"region":"uk"}`},
	{http.MethodPost, "/seal", ""},
	{http.MethodPost, "/flip", ""},
	{http.MethodPost, "/open", ""},
	{http.MethodPost, "/finish", ""},
	{http.MethodPost, "/abort", ""},
}

// Every door is the operator's: a member's token and an Instance-role
// service token meet the same bytes http.NotFound writes, before MV-O20 or
// anything else is said.
func TestOfflineDoorsAreOperatorOnly(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	member := w.token
	router := serviceToken(t, w.fx.st, OpsAccountID, "router", "shop", svcrole.Instance)
	if err := w.fx.st.SetOperator("alice@keymail.dev", false); err != nil {
		t.Fatal(err)
	}
	want := canonicalNotFound(t)
	for _, who := range []struct{ name, token string }{{"a member", member}, {"an instance token", router}} {
		for _, rt := range offlineRoutes {
			code, h, body := rawReq(t, rt.method, w.base+rt.path, who.token, rt.body)
			assertNotFound(t, rt.method+" "+rt.path+" as "+who.name, want, code, h, body)
		}
	}
	// The switch being on does not move the 404 behind MV-O20.
	w.fx.srv.cfg.MovesEnabled = true
	for _, rt := range offlineRoutes {
		code, h, body := rawReq(t, rt.method, w.base+rt.path, member, rt.body)
		assertNotFound(t, rt.method+" "+rt.path+" as a member with Phase 1 on", want, code, h, body)
	}
	if rec := w.record(); rec.Move != nil {
		t.Fatalf("a refused request wrote a move: %+v", rec.Move)
	}
}

// MV-O20, from the console's own configuration and from the deployment's
// backfill plan; each leaves the bucket as it found it.
func TestOfflineDoorsRefuseUnderClaims(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	each := func(label string) {
		t.Helper()
		for _, rt := range offlineRoutes {
			w.refused(rt.method, rt.path, rt.body, http.StatusConflict, msgO20)
		}
		_ = label
	}
	w.fx.srv.cfg.HostClaim = true
	each("claims, no authority")
	w.fx.srv.cfg.ClaimAuthority = &hostops.ProductionAuthority{}
	each("claims, with an authority")
	w.fx.srv.cfg.HostClaim, w.fx.srv.cfg.ClaimAuthority = false, nil
	w.fx.srv.cfg.MovesEnabled = true
	each("Phase 1")
	w.fx.srv.cfg.MovesEnabled = false

	plan := func(phase string) {
		t.Helper()
		raw, err := release.SignRootEnvelope(release.RootSigner(w.fx.priv), hostops.BackfillPlan{V: 1, ID: "plan-1", Phase: phase, CreatedAt: time.Now().UTC()})
		if err != nil {
			t.Fatal(err)
		}
		offlinePut(t, w.fx.rel, hostops.BackfillKey, raw)
	}
	plan("collecting")
	w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, msgO20)
	plan("adopting")
	w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, msgO20)
	offlinePut(t, w.fx.rel, hostops.BackfillKey, []byte("not an envelope"))
	w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, msgO20)
	// A plan at done is allowed, with any number of hosts.
	plan("done")
	w.start()
}

func TestOfflineStartRefusals(t *testing.T) {
	type tc struct {
		name  string
		boxes []offlineBox
		setup func(w *offlineWorld)
		body  string
		code  int
		msg   string
	}
	const body = `{"region":"uk"}`
	other := func(w *offlineWorld) {
		// A second instance, moving offline: started through its own door, so
		// it holds the lock.
		v, code, out := w.startOther("beta.shop.example.com")
		if code != http.StatusAccepted {
			t.Fatalf("beta start = %d %s (%+v)", code, out, v)
		}
	}
	cases := []tc{
		{name: "a replica", boxes: oneEach, body: body, code: 400, msg: "a replica does not move. delete it and create one in the new region",
			setup: func(w *offlineWorld) { w.edit(func(r *provision.Record) error { r.Role = "replica"; return nil }) }},
		{name: "replicas", boxes: oneEach, body: body, code: 400, msg: "a home with replicas does not move. remove its replicas first",
			setup: func(w *offlineWorld) {
				w.edit(func(r *provision.Record) error { r.Replicas = []string{"uk"}; return nil })
			}},
		{name: "steering", boxes: oneEach, body: body, code: 400, msg: offlineHost + " is steered or runs under a unit. offline moves cannot carry either yet",
			setup: func(w *offlineWorld) { w.edit(func(r *provision.Record) error { r.Steering = "geo"; return nil }) }},
		{name: "a unit", boxes: oneEach, body: body, code: 400, msg: offlineHost + " is steered or runs under a unit. offline moves cannot carry either yet",
			setup: func(w *offlineWorld) { w.edit(func(r *provision.Record) error { r.Unit = "app.service"; return nil }) }},
		{name: "a record acquisition", boxes: oneEach, body: body, code: 409, msg: msgO29,
			setup: func(w *offlineWorld) { w.edit(func(r *provision.Record) error { r.Acquisition = "acq-1"; return nil }) }},
		{name: "a host claim", boxes: oneEach, body: body, code: 409, msg: msgO29,
			setup: func(w *offlineWorld) { offlineClaimFor(t, w.fx, offlineHost, "ie-box", "ie") }},
		{name: "a domains claim with an acquisition", boxes: oneEach, body: body, code: 409, msg: msgO29,
			setup: func(w *offlineWorld) {
				c := domains.Claim{Schema: 1, Host: offlineHost, Account: "bab", App: "shop", Acquisition: "acq-1"}
				raw, _ := json.Marshal(c)
				offlinePut(t, w.fx.rel, domains.ClaimKey(offlineHost), raw)
			}},
		{name: "an admission", boxes: oneEach, body: body, code: 409, msg: msgO29,
			setup: func(w *offlineWorld) {
				key, err := hostops.AdmissionKey("bab", "shop")
				if err != nil {
					t.Fatal(err)
				}
				offlinePut(t, w.fx.rel, key, []byte(`{}`))
			}},
		{name: "a second move in flight", boxes: oneEach, body: body, code: 409, msg: "beta.shop.example.com is moving offline. move one instance at a time", setup: other},
		{name: "no region", boxes: oneEach, body: `{"region":""}`, code: 400, msg: "invalid region"},
		{name: "a bad region", boxes: oneEach, body: `{"region":"UK!"}`, code: 400, msg: "invalid region"},
		{name: "the current region", boxes: oneEach, body: `{"region":"ie"}`, code: 409, msg: "the instance is already in ie"},
		{name: "the manifest's regions", boxes: oneEach, body: body, code: 400, msg: "region uk is not one of this app's regions: ie",
			setup: func(w *offlineWorld) {
				m := offlineManifest()
				m.Pool = ""
				enableInstances(t, w.fx, "shop", m)
			}},
		{name: "an unreadable edges file", boxes: oneEach, body: body, code: 409, msg: "the fleet's edges files could not all be read. try again",
			setup: func(w *offlineWorld) { offlinePut(t, w.fx.rel, edgesFilePrefix+"broken.json", []byte("{")) }},
		{name: "no source box", boxes: []offlineBox{ukOff("uk-box")}, body: body, code: 400, msg: "no eligible box in ie publishes an edges file"},
		{name: "no target box", boxes: []offlineBox{ieOff("ie-box")}, body: body, code: 409, msg: "uk has 0 boxes. an offline move needs exactly one box in the region it moves to"},
		{name: "two target boxes", boxes: []offlineBox{ieOff("ie-box"), ukOff("uk-a"), ukOff("uk-b")}, body: body, code: 409,
			msg: "uk has 2 boxes. an offline move needs exactly one box in the region it moves to"},
		{name: "a source box without the capability", boxes: []offlineBox{func() offlineBox { b := ieOff("ie-box"); b.NoCap = true; return b }(), ukOff("uk-box")},
			body: body, code: 409, msg: msgO21("ie-box")},
		{name: "the second source box without the capability", boxes: []offlineBox{ieOff("ie-a"), func() offlineBox { b := ieOff("ie-b"); b.NoCap = true; return b }(), ukOff("uk-box")},
			body: body, code: 409, msg: msgO21("ie-b")},
		{name: "a target box without the capability", boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.NoCap = true; return b }()},
			body: body, code: 409, msg: msgO21("uk-box")},
		{name: "a target that does not hold the secrets key", boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.FP = "not-it"; return b }()},
			body: body, code: 409, msg: "uk-box does not hold this deployment's secrets key"},
		{name: "a target that lists the host", boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.Hosts = []string{offlineHost}; return b }()},
			body: body, code: 409, msg: "uk-box already serves " + offlineHost + ". an offline move needs the region it moves to to be empty of it"},
		{name: "a target with no data bucket", boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.Bucket = ""; return b }()},
			body: body, code: 400, msg: "region uk cannot take a move in this deployment. it publishes no staging bucket. Regions that can: ie"},
		// Rule 3: a claims-on box in either region, judged from its published state.
		{name: "a claims-on box in the source region that serves the shared pool", body: body, code: 409, msg: msgO21("ie-claims"),
			boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ieOff("ie-claims"); b.ClaimsOn = true; b.NoCap = true; return b }(), ukOff("uk-box")}},
		{name: "a claims-on box in the target region whose pools do not match (no census evidence is published)", body: body, code: 409, msg: msgO21("uk-claims"),
			boxes: []offlineBox{ieOff("ie-box"), ukOff("uk-box"), func() offlineBox {
				b := ukOff("uk-claims")
				b.ClaimsOn = true
				b.NoCap = true
				b.NotSharedPool = true
				return b
			}()}},
		{name: "a claims-on box that publishes a row for the host", body: body, code: 409, msg: msgO21("ie-claims"),
			boxes: []offlineBox{ieOff("ie-box"), func() offlineBox {
				b := ieOff("ie-claims")
				b.ClaimsOn, b.NoCap, b.NotSharedPool, b.Hosts = true, true, true, []string{offlineHost}
				return b
			}(), ukOff("uk-box")}},
		{name: "a box whose report says claims are on although it carries the capability", body: body, code: 409, msg: msgO21("uk-box"),
			boxes: []offlineBox{ieOff("ie-box"), func() offlineBox { b := ukOff("uk-box"); b.ClaimsOn = true; return b }()}},
		{name: "a box outside both regions that runs claims is not consulted", body: "", code: 202,
			boxes: []offlineBox{ieOff("ie-box"), ukOff("uk-box"), func() offlineBox {
				b := offlineBox{Box: "se-box", Placement: "se", Bucket: "data-se"}
				b.ClaimsOn, b.NoCap = true, true
				return b
			}()}},
		{name: "a box with no metrics and no capability in the region cannot be judged", body: body, code: 409, msg: msgO21("ie-old"),
			boxes: []offlineBox{ieOff("ie-box"), func() offlineBox {
				b := ieOff("ie-old")
				b.NoMetrics, b.NoCap, b.NotSharedPool = true, true, true
				return b
			}(), ukOff("uk-box")}},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			w := newOfflineWorld(t, c.boxes...)
			if c.setup != nil {
				c.setup(w)
			}
			if c.code == 202 {
				w.start()
				return
			}
			w.refused(http.MethodPost, "", c.body, c.code, c.msg)
			if rec := w.record(); rec.Move != nil && c.name != "a second move in flight" {
				t.Fatalf("a refused start left a move: %+v", rec.Move)
			}
		})
	}
}

// A done backfill plan does not refuse, and the claims-on truth table is
// all three conditions: one missing is a refusal.
func TestOfflineClaimsOnBoxNeedsAllThree(t *testing.T) {
	yes, no := true, false
	cases := []struct {
		name string
		box  offlineClaimsOnBox
		ok   bool
	}{
		{"all three", offlineClaimsOnBox{Census: &yes, SharedPool: &no}, true},
		{"a row for the host", offlineClaimsOnBox{HostRow: true, Census: &yes, SharedPool: &no}, false},
		{"census not present", offlineClaimsOnBox{Census: &no, SharedPool: &no}, false},
		{"census not said", offlineClaimsOnBox{SharedPool: &no}, false},
		{"pools match", offlineClaimsOnBox{Census: &yes, SharedPool: &yes}, false},
		{"pools not said", offlineClaimsOnBox{Census: &yes}, false},
	}
	for _, c := range cases {
		if got := c.box.ok(); got != c.ok {
			t.Errorf("%s: ok = %v, want %v", c.name, got, c.ok)
		}
	}
	// No report field carries the census today, so no claims-on box passes.
	if offlineCensusReported(&status.BoxMetrics{HostClaimEnabled: true}) != nil {
		t.Error("a box report says something about census.json this build does not know how to read")
	}
}

func TestOfflineStartWritesTheMoveAndBumpsTheSerial(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	if v := w.status(); v.State != "none" || v.Host != offlineHost {
		t.Fatalf("status before any move = %+v", v)
	}
	v := w.start()
	if v.From != "ie" || v.To != "uk" || v.Host != offlineHost || !v.CanAbort || v.RequestedBy != "alice@keymail.dev" {
		t.Fatalf("start view = %+v", v)
	}
	rec := w.record()
	if rec.Move == nil || rec.Move.Kind != provision.MoveKindOffline || rec.Move.State != provision.OfflineStopping ||
		rec.Move.From != "ie" || rec.Move.To != "uk" || rec.Move.ID != v.ID || rec.Region != "ie" {
		t.Fatalf("record after start = %+v region %q", rec.Move, rec.Region)
	}
	// Waiting for the boxes, in the move's own words; no verb is next.
	if v.Next != "" || v.Waiting != msgMC10("ie-box") {
		t.Fatalf("view waits for %q (next %q)", v.Waiting, v.Next)
	}
	// A second start is refused by the record.
	w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, "a move to uk is already offline-stopping")
	// The status now carries the manifest with the door-owned class row ok.
	got := w.status()
	if got.State != provision.OfflineStopping || len(got.Items) != len(moves.OfflineManifest) {
		t.Fatalf("status = %+v", got)
	}
	for _, it := range got.Items {
		switch it.Owner {
		case moves.OfflineOwnerDoor:
			if it.Status != "ok" {
				t.Errorf("%s: %s", it.ID, it.Status)
			}
		case moves.OfflineOwnerNone:
			if it.Status != "unchecked" {
				t.Errorf("%s: %s", it.ID, it.Status)
			}
		default:
			if it.Status != "waiting" {
				t.Errorf("%s: %s", it.ID, it.Status)
			}
		}
	}
}

func TestOfflineSealDoor(t *testing.T) {
	w := newOfflineWorld(t, ieOff("ie-a"), ieOff("ie-b"), ukOff("uk-box"))
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, offlineHost+" has no offline move in flight")
	w.start()
	// No report yet: the first box is named.
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgMC10("ie-a"))
	// A box that has not held, fenced and quiesced is named.
	for _, lost := range []string{"held", "fenced", "quiesced"} {
		w.stopped(func(box string, r *moves.OfflineReport) {
			if box == "ie-b" {
				switch lost {
				case "held":
					r.Held = false
				case "fenced":
					r.Fenced = false
				case "quiesced":
					r.Quiesced = false
				}
			}
		}, "ie-a", "ie-b")
		w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgMC10("ie-b"))
	}
	// A stale report is not a report.
	w.stopped(func(box string, r *moves.OfflineReport) {
		if box == "ie-b" {
			r.At = time.Now().Add(-5 * time.Minute)
		}
	}, "ie-a", "ie-b")
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgO178("ie-b"))
	// A box still listing the host has not stopped serving it.
	w.stopped(nil, "ie-a", "ie-b")
	still := ieOff("ie-b")
	still.Hosts = []string{offlineHost}
	w.fleet(still)
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgMC10("ie-b"))
	w.fleet(ieOff("ie-b"))
	// A sidecar anywhere.
	w.stopped(func(box string, r *moves.OfflineReport) { r.Sidecar = box == "ie-b" }, "ie-a", "ie-b")
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, "ie-b keeps a sidecar file for "+offlineHost+". instances with sidecars cannot move yet")
	// Local data on two boxes.
	w.stopped(func(box string, r *moves.OfflineReport) { r.LocalDB = true; r.Files = box == "ie-b" }, "ie-a", "ie-b")
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, offlineHost+" keeps local data on more than one box: ie-a, ie-b. remove the stale copy first")
	// One box with data is the sealer, even when it is not the first.
	before := serialBytes(t, w.fx.rel)
	w.stopped(func(box string, r *moves.OfflineReport) { r.Env = box == "ie-b" }, "ie-a", "ie-b")
	v := w.ok("/seal", "", provision.OfflineSealing)
	if v.Sealer != "ie-b" || strings.Join(v.Boxes, ",") != "ie-a,ie-b" {
		t.Fatalf("sealer %q boxes %v", v.Sealer, v.Boxes)
	}
	if serialBytes(t, w.fx.rel) == before {
		t.Fatal("seal did not bump the serial")
	}
	rec := w.record()
	if rec.Move.State != provision.OfflineSealing || rec.Move.Sealer != "ie-b" || len(rec.Move.Boxes) != 2 {
		t.Fatalf("record = %+v", rec.Move)
	}
	// seal is the verb of stopping only.
	w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgO25(provision.OfflineSealing))
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealing))
}

func TestOfflineSealWithNoDataTakesTheFirstBox(t *testing.T) {
	w := newOfflineWorld(t, ieOff("ie-a"), ieOff("ie-b"), ukOff("uk-box"))
	w.start()
	w.stopped(func(box string, r *moves.OfflineReport) { r.RowPresent = false }, "ie-a", "ie-b")
	if v := w.ok("/seal", "", provision.OfflineSealing); v.Sealer != "ie-a" {
		t.Fatalf("sealer = %q", v.Sealer)
	}
}

func TestOfflineFlipDoor(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.start()
	w.stopped(nil, "ie-box")
	w.ok("/seal", "", provision.OfflineSealing)
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealing))
	seal := w.sealed("ie-box")
	w.stopped(nil, "ie-box")
	sealKey := moves.OfflineSealKey(offlineHost, w.id)
	// seal.json is create-only: a second write does not replace it.
	if _, err := w.fx.rel.Put(context.Background(), sealKey, []byte("{}"), ""); err == nil {
		t.Fatal("the seal was written twice")
	}
	// The sealer's living report is rewritten on every pass and the seal is
	// not: nothing the living report does moves the hash flip checks.
	w.sealerFresh("ie-box")
	w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseSealed, func(r *moves.OfflineReport) {
		r.Items = map[string]moves.Check{} // the living report carries no items at all now
		r.TombstoneOwner = moves.TombstoneOwner(w.id)
		r.TombstoneUntil = time.Now().Add(time.Hour)
		r.FinalTXID = 99
	})
	// A seal that is not the one the record names.
	tampered := w.putSeal("ie-box", func(r *moves.OfflineReport) { r.FinalTXID = 99 })
	if string(tampered) == string(seal) {
		t.Fatal("the tampered seal is the seal")
	}
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealed))
	offlinePut(t, w.fx.rel, sealKey, seal)
	// No seal at all.
	if err := w.fx.rel.Delete(context.Background(), sealKey); err != nil {
		t.Fatal(err)
	}
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealed))
	offlinePut(t, w.fx.rel, sealKey, seal)
	// A seal item that is not OK, and a seal of another move's tombstone, each
	// with a hash the record names.
	bad := w.putSeal("ie-box", func(r *moves.OfflineReport) {
		r.Items["files"] = moves.Check{OK: false, Reason: "a file the archive refuses"}
	})
	w.rehash(bad)
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealed))
	bad = w.putSeal("ie-box", func(r *moves.OfflineReport) { r.TombstoneOwner = moves.TombstoneOwner("mv_other") })
	w.rehash(bad)
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealed))
	// A seal written by a box other than the recorded sealer.
	bad = w.putSeal("ie-x", nil)
	w.rehash(bad)
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO25(provision.OfflineSealed))
	offlinePut(t, w.fx.rel, sealKey, seal)
	w.rehash(seal)
	// A process left, a layout journal, a stale report.
	w.stopped(func(_ string, r *moves.OfflineReport) { r.Procs = 1 }, "ie-box")
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO31("ie-box"))
	w.stopped(func(_ string, r *moves.OfflineReport) { r.LayoutJournal = true }, "ie-box")
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO31("ie-box"))
	// A stale report is not a process (harness Run 9, finding 7): MV-O178,
	// never MV-O31.
	w.stopped(func(_ string, r *moves.OfflineReport) { r.At = time.Now().Add(-3 * time.Minute) }, "ie-box")
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgO178("ie-box"))
	w.stopped(func(_ string, r *moves.OfflineReport) { r.Fenced = false }, "ie-box")
	w.refused(http.MethodPost, "/flip", "", http.StatusConflict, msgMC10("ie-box"))
	// All clear.
	w.stopped(nil, "ie-box")
	before := serialBytes(t, w.fx.rel)
	v := w.ok("/flip", "", provision.OfflineLanding)
	if v.Next != "" {
		t.Fatalf("next after flip = %q", v.Next)
	}
	if serialBytes(t, w.fx.rel) == before {
		t.Fatal("flip did not bump the serial")
	}
	if rec := w.record(); rec.Region != "uk" || rec.Move.State != provision.OfflineLanding {
		t.Fatalf("after flip: region %q move %+v", rec.Region, rec.Move)
	}
}

// rehash points the record's source evidence at a seal's bytes, as the
// sealer's transition records its own seal.
func (w *offlineWorld) rehash(seal []byte) {
	w.t.Helper()
	w.edit(func(r *provision.Record) error {
		r.Move.Source.Evidence = moves.OfflineSealEvidence(offlineHost, w.id, seal)
		return nil
	})
}

func TestOfflineOpenDoorRechecksTheStop(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.toLanded()
	// The listing check of a source box is not the door's: the door reads its report.
	reset := func() {
		w.stopped(nil, "ie-box")
		w.sealerFresh("ie-box")
		w.report(moves.OfflineRoleTarget, "uk-box", moves.OfflinePhaseLanded, func(r *moves.OfflineReport) { r.Items = okChecks(moves.OfflineOwnerLanding) })
	}
	reset()
	// A stale sealer report.
	w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseSealed, func(r *moves.OfflineReport) {
		r.Items = sealerChecks()
		r.At = time.Now().Add(-3 * time.Minute)
		r.TombstoneOwner = moves.TombstoneOwner(w.id)
		r.TombstoneUntil = time.Now().Add(time.Hour)
	})
	w.refused(http.MethodPost, "/open", "", http.StatusConflict, msgO178("ie-box"))
	reset()
	// A tombstone that is another move's, or none, or run out.
	for name, edit := range map[string]func(*moves.OfflineReport){
		"none":     func(r *moves.OfflineReport) { r.TombstoneOwner = "" },
		"another":  func(r *moves.OfflineReport) { r.TombstoneOwner = moves.TombstoneOwner("mv_other") },
		"expired":  func(r *moves.OfflineReport) { r.TombstoneUntil = time.Now().Add(-time.Minute) },
		"unstated": func(r *moves.OfflineReport) { r.TombstoneUntil = time.Time{} },
	} {
		w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseSealed, func(r *moves.OfflineReport) {
			r.Items = sealerChecks()
			r.TombstoneOwner = moves.TombstoneOwner(w.id)
			r.TombstoneUntil = time.Now().Add(time.Hour)
			edit(r)
		})
		code, out := w.req(http.MethodPost, "/open", "")
		if code != http.StatusConflict || errorText(out) != msgO25(provision.OfflineLanded) {
			t.Fatalf("tombstone %s: %d %s", name, code, out)
		}
		reset()
	}
	// A source report that lost Fenced, or went stale.
	w.stopped(func(_ string, r *moves.OfflineReport) { r.Fenced = false }, "ie-box")
	w.refused(http.MethodPost, "/open", "", http.StatusConflict, msgMC10("ie-box"))
	w.stopped(func(_ string, r *moves.OfflineReport) { r.At = time.Now().Add(-10 * time.Minute) }, "ie-box")
	w.refused(http.MethodPost, "/open", "", http.StatusConflict, msgO178("ie-box"))
	reset()
	// The seal changed under the move.
	sealKey := moves.OfflineSealKey(offlineHost, w.id)
	sealRaw, _, err := w.fx.rel.Get(context.Background(), sealKey)
	if err != nil {
		t.Fatal(err)
	}
	offlinePut(t, w.fx.rel, sealKey, append(append([]byte{}, sealRaw...), ' '))
	w.refused(http.MethodPost, "/open", "", http.StatusConflict, msgO25(provision.OfflineLanded))
	offlinePut(t, w.fx.rel, sealKey, sealRaw)
	// A landing item that is not OK.
	w.report(moves.OfflineRoleTarget, "uk-box", moves.OfflinePhaseLanded, func(r *moves.OfflineReport) {
		r.Items = okChecks(moves.OfflineOwnerLanding)
		r.Items["config"] = moves.Check{OK: false, Reason: "the render waits"}
	})
	w.refused(http.MethodPost, "/open", "", http.StatusConflict, msgO25(provision.OfflineLanded))
	// The status view says the same, and names no verb.
	if v := w.status(); v.Next != "" || v.Waiting != msgO25(provision.OfflineLanded) {
		t.Fatalf("status = next %q waiting %q", v.Next, v.Waiting)
	} else {
		for _, it := range v.Items {
			if it.ID == "config" && (it.Status != "waiting" || it.Detail != "the render waits") {
				t.Fatalf("config item = %+v", it)
			}
		}
	}
	reset()
	if v := w.status(); v.Next != offlineVerbOpen {
		t.Fatalf("status next = %q (%s)", v.Next, v.Waiting)
	}
	before := serialBytes(t, w.fx.rel)
	w.ok("/open", "", provision.OfflineOpen)
	if serialBytes(t, w.fx.rel) == before {
		t.Fatal("open did not bump the serial")
	}
	if rec := w.record(); rec.Move.State != provision.OfflineOpen || rec.Region != "uk" || rec.Move.OpenedAt == nil {
		t.Fatalf("after open: %+v region %q", rec.Move, rec.Region)
	}
}

func TestOfflineFinishDoor(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.toLanded()
	w.ok("/open", "", provision.OfflineOpen)
	// open to finishing: the target's opened and every source's parked.
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineOpen))
	w.report(moves.OfflineRoleTarget, "uk-box", moves.OfflinePhaseOpened, func(r *moves.OfflineReport) { r.Items = okChecks(moves.OfflineOwnerLanding) })
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineOpen))
	// A parked report whose inventory is not settled is not parked.
	for name, retired := range map[string][]moves.RetiredPath{
		"an unknown action": {{Path: "/data/acme.shop.example.com", Action: "deleted"}},
		"left, no reason":   {{Path: "/opt/carlos/live/acme.shop.example.com", Action: moves.RetiredLeft}},
		"no path":           {{Action: moves.RetiredParked}},
	} {
		w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseParked, func(r *moves.OfflineReport) { r.Retired = retired })
		code, out := w.req(http.MethodPost, "/finish", "")
		if code != http.StatusConflict || errorText(out) != msgO25(provision.OfflineOpen) {
			t.Fatalf("%s: %d %s", name, code, out)
		}
	}
	// A path the box could not park is terminal when it says why, and the
	// status view shows it.
	w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseParked, func(r *moves.OfflineReport) {
		r.Retired = []moves.RetiredPath{
			{Path: "/data/acme.shop.example.com", Action: moves.RetiredParked},
			{Path: "/data/acme.shop.example.com.env", Action: moves.RetiredAbsent},
			{Path: "/opt/carlos/live/acme.shop.example.com", Action: moves.RetiredLeft, Reason: "root-owned"},
		}
	})
	if v := w.status(); v.Next != offlineVerbFinish || len(v.Left) != 1 || v.Left[0] != (offlineLeftView{Box: "ie-box", Path: "/opt/carlos/live/acme.shop.example.com", Reason: "root-owned"}) {
		t.Fatalf("status = next %q left %+v waiting %q", v.Next, v.Left, v.Waiting)
	}
	before := serialBytes(t, w.fx.rel)
	w.ok("/finish", "", provision.OfflineFinishing)
	if serialBytes(t, w.fx.rel) == before {
		t.Fatal("finish did not bump the serial")
	}
	// finishing to cleared: the sealer lifted and every source gone.
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineFinishing))
	w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseLifted, nil)
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineFinishing))
	w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseGone, nil)
	v := w.ok("/finish", "", "none")
	if v.Last == nil || v.Last.Outcome != provision.OfflineOutcomeDone || v.Last.ID != w.id {
		t.Fatalf("last = %+v", v.Last)
	}
	if rec := w.record(); rec.Move != nil || rec.Region != "uk" || rec.LastMove == nil {
		t.Fatalf("after finish: move %+v region %q", rec.Move, rec.Region)
	}
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, offlineHost+" has no offline move in flight")
	// The status of a host with no move is the outcome of the last.
	if s := w.status(); s.State != "none" || s.Last == nil {
		t.Fatalf("status = %+v", s)
	}
}

func TestOfflineAbortDoor(t *testing.T) {
	// Before the flip: the region is untouched.
	w := newOfflineWorld(t, oneEach...)
	w.refused(http.MethodPost, "/abort", "", http.StatusConflict, offlineHost+" has no offline move in flight")
	w.start()
	before := serialBytes(t, w.fx.rel)
	w.ok("/abort", "", provision.OfflineReverting)
	if serialBytes(t, w.fx.rel) == before {
		t.Fatal("abort did not bump the serial")
	}
	if rec := w.record(); rec.Region != "ie" || rec.Move.State != provision.OfflineReverting {
		t.Fatalf("after abort: %q %+v", rec.Region, rec.Move)
	}
	// A second abort writes nothing.
	state := bucketState(t, w.fx.rel)
	w.ok("/abort", "", provision.OfflineReverting)
	for k, v := range bucketState(t, w.fx.rel) {
		if state[k] != v {
			t.Fatalf("a second abort changed %s", k)
		}
	}
	// reverting to cleared: every source box's reverted. Aborted before the
	// seal there is no sealer to ask, and no target either.
	w.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineReverting))
	w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseReverted, nil)
	if v := w.ok("/finish", "", "none"); v.Last == nil || v.Last.Outcome != provision.OfflineOutcomeAborted {
		t.Fatalf("last = %+v", v.Last)
	}

	// After the flip: the record goes back to the source.
	w2 := newOfflineWorld(t, oneEach...)
	w2.start()
	w2.stopped(nil, "ie-box")
	w2.ok("/seal", "", provision.OfflineSealing)
	w2.sealed("ie-box")
	w2.stopped(nil, "ie-box")
	w2.ok("/flip", "", provision.OfflineLanding)
	if rec := w2.record(); rec.Region != "uk" {
		t.Fatalf("region after flip = %q", rec.Region)
	}
	w2.ok("/abort", "", provision.OfflineReverting)
	if rec := w2.record(); rec.Region != "ie" {
		t.Fatalf("region after abort = %q", rec.Region)
	}
	// Sealed and then aborted: the sealer's reverted is asked for too, and
	// the target's, which may be a box that died, is not.
	w2.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseReverted, nil)
	w2.refused(http.MethodPost, "/finish", "", http.StatusConflict, msgO25(provision.OfflineReverting))
	w2.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseReverted, nil)
	if v := w2.ok("/finish", "", "none"); v.Last == nil || v.Last.Outcome != provision.OfflineOutcomeAborted {
		t.Fatalf("last = %+v", v.Last)
	}
	if rec := w2.record(); rec.Region != "ie" || rec.Move != nil {
		t.Fatalf("after finish: %q %+v", rec.Region, rec.Move)
	}

	// From open on, abort is MV-O27 and writes nothing.
	w3 := newOfflineWorld(t, oneEach...)
	w3.toLanded()
	w3.ok("/open", "", provision.OfflineOpen)
	w3.refused(http.MethodPost, "/abort", "", http.StatusConflict, offlineHost+" is already served from uk. move it back with a new move")
	w3.edit(provision.OfflineFinishingTransition(w3.id))
	w3.refused(http.MethodPost, "/abort", "", http.StatusConflict, offlineHost+" is already served from uk. move it back with a new move")
}

// The status view is the move, every report, the manifest with each item's
// latest check and the next legal verb.
func TestOfflineStatusView(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.toLanded()
	v := w.status()
	if v.State != provision.OfflineLanded || v.ID != w.id || v.Sealer != "ie-box" || v.From != "ie" || v.To != "uk" || !v.CanAbort || v.Next != offlineVerbOpen {
		t.Fatalf("status = %+v", v)
	}
	if v.Failed {
		t.Error("a healthy move is marked failed")
	}
	roles := map[string]bool{}
	for _, r := range v.Reports {
		roles[r.Role+"/"+r.Box] = true
		if !r.Fresh {
			t.Errorf("report %s/%s is not fresh", r.Role, r.Box)
		}
	}
	for _, want := range []string{"source/ie-box", "sealer/ie-box", "target/uk-box"} {
		if !roles[want] {
			t.Errorf("no %s report in the view: %v", want, roles)
		}
	}
	if len(v.Items) != len(moves.OfflineManifest) {
		t.Fatalf("%d items, want %d", len(v.Items), len(moves.OfflineManifest))
	}
	for _, it := range v.Items {
		if (it.Owner == moves.OfflineOwnerSeal || it.Owner == moves.OfflineOwnerLanding) && it.Status != "ok" {
			t.Errorf("item %s is %s", it.ID, it.Status)
		}
	}
	// The step that owns an item decides it: a landing item is the target's
	// to prove, a seal item the sealer's. A failed phase marks a non-OK
	// item failed; a landing phase marks it waiting.
	w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseFailed, func(r *moves.OfflineReport) {
		r.Items = sealerChecks()
		r.Items["db-tail"] = moves.Check{OK: false, Reason: "the replica is behind"}
		r.Items["files"] = moves.Check{OK: false, Reason: "the sealer's own view does not decide it"}
	})
	w.report(moves.OfflineRoleTarget, "uk-box", moves.OfflinePhaseLanded, func(r *moves.OfflineReport) {
		r.Items = okChecks(moves.OfflineOwnerLanding)
		delete(r.Items, "release")
		r.Items["files"] = moves.Check{OK: false, Reason: "the digest differs"}
	})
	if !w.status().Failed {
		t.Error("a sealer that reported failed is not marked failed in the view")
	}
	seen := map[string]offlineItemView{}
	for _, it := range w.status().Items {
		seen[it.ID] = it
	}
	if it := seen["db-tail"]; it.Status != "failed" || it.Detail != "the replica is behind" {
		t.Errorf("db-tail = %+v", it)
	}
	if it := seen["files"]; it.Status != "waiting" || it.Detail != "the digest differs" || it.Owner != moves.OfflineOwnerLanding || it.Source != moves.OfflineSourceSeal {
		t.Errorf("files = %+v", it)
	}
	if it := seen["release"]; it.Status != "waiting" {
		t.Errorf("release = %+v", it)
	}
	// A report that is not this move's, or not at its own key, is not in the view.
	other := moves.OfflineReport{V: 1, Op: "mv_other", Host: offlineHost, Box: "ie-box", Role: moves.OfflineRoleSource, Phase: moves.OfflinePhaseStopped, At: time.Now()}
	body, _ := json.Marshal(other)
	offlinePut(t, w.fx.rel, moves.OfflineReportKey(offlineHost, w.id, moves.OfflineRoleSource, "ie-x"), body)
	for _, r := range w.status().Reports {
		if r.Box == "ie-x" {
			t.Errorf("a report at the wrong key is in the view: %+v", r)
		}
	}
}

// Evidence matching takes the report's hash bare or after its key.
func TestOfflineEvidenceMatches(t *testing.T) {
	raw := []byte(`{"x":1}`)
	sum := sha256.Sum256(raw)
	h := hex.EncodeToString(sum[:])
	key := moves.OfflineReportKey(offlineHost, "mv_00000000000000000000000000", moves.OfflineRoleSealer, "ie-box")
	for _, ok := range []string{h, "sha256:" + h, key + "@" + h, key + ":" + h, key + " " + h, key + "#sha256:" + h} {
		if !offlineEvidenceMatches(ok, key, raw) {
			t.Errorf("%q should match", ok)
		}
	}
	for _, bad := range []string{"", key, key + "@", key + "@" + strings.Repeat("0", 64), "other/key@" + h, h + "0", key + h} {
		if offlineEvidenceMatches(bad, key, raw) {
			t.Errorf("%q should not match", bad)
		}
	}
}

// A transition's refusal reaches the operator as a recorded sentence and
// never as the error's own text: the offline transitions' rows pass as they
// are (the values in them are the record's own), and anything else is
// MV-N46.
func TestOfflineConflictMapsOntoRecordedSentences(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.start()
	apply := func(mut func(*provision.Record) error) error {
		_, err := provision.SetRecordMove(context.Background(), w.fx.rel, "bab", "shop", offlineHost, mut)
		return err
	}
	const other = "mv_ZZZZZZZZZZZZZZZZZZZZZZZZZZ"
	cases := []struct {
		name string
		mut  func(*provision.Record) error
		want string
	}{
		{"wrong state", provision.OfflineFlip(w.id), "this offline move is offline-stopping. this step follows offline-sealed"},
		{"another move's id", provision.OfflineSeal(other, "ie-box", []string{"ie-box"}), "the record carries a different offline move"},
		{"a seal without its sealer among the boxes", provision.OfflineSeal(w.id, "ie-x", []string{"ie-box"}),
			"a seal names the box that seals and the boxes that stopped the instance. the box that seals is one of them"},
		{"an abort already past", provision.OfflineFinish(w.id, time.Now()), "this offline move is offline-stopping. it finishes from finishing or reverting"},
	}
	for _, c := range cases {
		err := apply(c.mut)
		if err == nil {
			t.Fatalf("%s: accepted", c.name)
		}
		if got := offlineConflict(err); got != c.want {
			t.Errorf("%s: %q, want %q", c.name, got, c.want)
		}
	}
	// A refusal this table does not know is the fixed sentence.
	if got := offlineConflict(fmt.Errorf("%w: something new and unreviewed", provision.ErrMoveConflict)); got != "this move's state does not allow that. read its state and try again" {
		t.Errorf("unknown refusal = %q", got)
	}
	// Once open, an abort is refused with its recorded sentence.
	w.stopped(nil, "ie-box")
	w.ok("/seal", "", provision.OfflineSealing)
	w.sealed("ie-box")
	w.stopped(nil, "ie-box")
	w.ok("/flip", "", provision.OfflineLanding)
	w.landed("uk-box")
	w.edit(provision.OfflineOpenTransition(w.id, time.Now()))
	err := apply(provision.OfflineAbort(w.id))
	if got := offlineConflict(err); got != "this offline move is offline-open. it cannot be aborted once it is open" {
		t.Errorf("abort after open = %q", got)
	}
	// A legacy move meets "this is not an offline move".
	if got := offlineConflict(fmt.Errorf("%w: %s", provision.ErrMoveConflict, "this is not an offline move")); got != "this is not an offline move" {
		t.Errorf("not offline = %q", got)
	}
}

// The flagship's shape: the home placement is the record's empty region and
// the console names it. A move out of home reads the placementless boxes, a
// move into it the same, and the operator reads the home by its name.
func TestOfflineMovesOutOfAndIntoHome(t *testing.T) {
	homeBox := offlineBox{Box: "home-box", Bucket: "data-ie"}
	w := newOfflineWorldIn(t, "ie", "ie", homeBox, ukOff("uk-box"))
	if rec := w.record(); rec.Region != "" {
		t.Fatalf("the record's region = %q, want the home's empty code", rec.Region)
	}
	w.refused(http.MethodPost, "", `{"region":"ie"}`, http.StatusConflict, "the instance is already in ie")
	v := w.start()
	if v.From != "ie" || v.To != "uk" || v.Waiting != "Waiting for home-box in ie to confirm it has stopped "+offlineHost+". "+offlineHost+" is offline." {
		t.Fatalf("view = %+v", v)
	}
	if rec := w.record(); rec.Move.From != "" || rec.Move.To != "uk" || rec.Region != "" {
		t.Fatalf("record = %+v region %q", rec.Move, rec.Region)
	}
	w.report(moves.OfflineRoleSource, "home-box", moves.OfflinePhaseStopped, nil)
	if v := w.ok("/seal", "", provision.OfflineSealing); v.Sealer != "home-box" {
		t.Fatalf("sealer = %q", v.Sealer)
	}

	// Back the other way: from uk into home.
	b := newOfflineWorldIn(t, "ie", "uk", homeBox, ukOff("uk-box"))
	bv := b.ok("", `{"region":"ie"}`, provision.OfflineStopping)
	if bv.From != "uk" || bv.To != "ie" {
		t.Fatalf("view = %+v", bv)
	}
	if rec := b.record(); rec.Move.From != "uk" || rec.Move.To != "" || rec.Region != "uk" {
		t.Fatalf("record = %+v region %q", rec.Move, rec.Region)
	}

	// A box that sets CARLOS_REGION to the home's name is home to the console
	// and not to itself: it cannot take a move into home.
	named := offlineBox{Box: "ie-named", Placement: "ie", Bucket: "data-ie"}
	c := newOfflineWorldIn(t, "ie", "uk", homeBox, ukOff("uk-box"), named)
	c.refused(http.MethodPost, "", `{"region":"ie"}`, http.StatusConflict, "ie-named sets CARLOS_REGION, so it cannot take a move into the home region")

	// And out of home, where its published placement is not the record's.
	d := newOfflineWorldIn(t, "ie", "ie", homeBox, ukOff("uk-box"), named)
	d.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, "ie-named sets CARLOS_REGION, so it cannot take a move into the home region")
}

// startOther creates a second instance and starts it through its own door.
func (w *offlineWorld) startOther(host string) (offlineView, int, string) {
	w.t.Helper()
	u := instancesURL(w.fx, "bab", "shop")
	if c, _, out := rawReq(w.t, http.MethodPost, u, w.token, `{"host":"`+host+`","region":"ie"}`); c != http.StatusCreated && c != http.StatusConflict {
		w.t.Fatalf("create %s = %d %s", host, c, out)
	}
	code, _, out := rawReq(w.t, http.MethodPost, u+"/"+host+"/move/offline", w.token, `{"region":"uk"}`)
	var v offlineView
	_ = json.Unmarshal([]byte(out), &v)
	return v, code, out
}

func (w *offlineWorld) lock() (moves.OfflineActive, bool) {
	w.t.Helper()
	raw, _, err := w.fx.rel.Get(context.Background(), moves.OfflineActiveKey)
	if err != nil {
		return moves.OfflineActive{}, false
	}
	var a moves.OfflineActive
	if err := json.Unmarshal(raw, &a); err != nil {
		w.t.Fatal(err)
	}
	return a, true
}

// Two starts at once for different hosts: exactly one wins the lock.
func TestOfflineConcurrentStartsOneWins(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	u := instancesURL(w.fx, "bab", "shop")
	if c, _, out := rawReq(t, http.MethodPost, u, w.token, `{"host":"beta.shop.example.com","region":"ie"}`); c != http.StatusCreated {
		t.Fatalf("create = %d %s", c, out)
	}
	hosts := []string{offlineHost, "beta.shop.example.com"}
	type answer struct {
		code int
		out  string
	}
	got := make([]answer, len(hosts))
	var wg sync.WaitGroup
	gate := make(chan struct{})
	for i, h := range hosts {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-gate
			c, _, out := rawReq(t, http.MethodPost, u+"/"+h+"/move/offline", w.token, `{"region":"uk"}`)
			got[i] = answer{c, out}
		}()
	}
	close(gate)
	wg.Wait()
	won, lost := -1, -1
	for i, a := range got {
		switch a.code {
		case http.StatusAccepted:
			won = i
		case http.StatusConflict:
			lost = i
		default:
			t.Fatalf("start %s = %d %s", hosts[i], a.code, a.out)
		}
	}
	if won < 0 || lost < 0 {
		t.Fatalf("want one winner and one loser: %+v", got)
	}
	if want := hosts[won] + " is moving offline. move one instance at a time"; errorText(got[lost].out) != want {
		t.Fatalf("the loser was told %q, want %q", errorText(got[lost].out), want)
	}
	if a, ok := w.lock(); !ok || a.Host != hosts[won] {
		t.Fatalf("lock = %+v %v, want the winner %s", a, ok, hosts[won])
	}
	loser, err := provision.GetRecord(context.Background(), w.fx.rel, "bab", "shop", hosts[lost])
	if err != nil || loser.Move != nil {
		t.Fatalf("the loser's record = %+v %v", loser.Move, err)
	}
}

// The lock is released when the move ends, by finish and by the finish that
// ends an abort, and a stale lock does not stop the next start.
func TestOfflineLockLifecycle(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	w.toLanded()
	if a, ok := w.lock(); !ok || a.Host != offlineHost || a.Op != w.id || a.Account != "bab" || a.App != "shop" {
		t.Fatalf("lock = %+v %v", a, ok)
	}
	w.ok("/open", "", provision.OfflineOpen)
	w.report(moves.OfflineRoleTarget, "uk-box", moves.OfflinePhaseOpened, func(r *moves.OfflineReport) { r.Items = okChecks(moves.OfflineOwnerLanding) })
	w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseParked, nil)
	w.ok("/finish", "", provision.OfflineFinishing)
	if _, ok := w.lock(); !ok {
		t.Fatal("the lock was released before the move ended")
	}
	w.report(moves.OfflineRoleSealer, "ie-box", moves.OfflinePhaseLifted, nil)
	w.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseGone, nil)
	w.ok("/finish", "", "none")
	if _, ok := w.lock(); ok {
		t.Fatal("finish left the lock")
	}

	// An abort ends at finish too.
	b := newOfflineWorld(t, oneEach...)
	b.start()
	b.ok("/abort", "", provision.OfflineReverting)
	if _, ok := b.lock(); !ok {
		t.Fatal("abort released the lock before the finish")
	}
	b.report(moves.OfflineRoleSource, "ie-box", moves.OfflinePhaseReverted, nil)
	b.ok("/finish", "", "none")
	if _, ok := b.lock(); ok {
		t.Fatal("the finish after an abort left the lock")
	}
	// Another start now takes it.
	b.start()
	if a, ok := b.lock(); !ok || a.Op != b.id {
		t.Fatalf("lock = %+v %v", a, ok)
	}
}

func TestOfflineStaleLockIsReplaced(t *testing.T) {
	put := func(w *offlineWorld, a moves.OfflineActive) {
		body, _ := json.Marshal(a)
		offlinePut(t, w.fx.rel, moves.OfflineActiveKey, body)
	}
	// A lock naming a host with no record.
	w := newOfflineWorld(t, oneEach...)
	put(w, moves.OfflineActive{Host: "gone.shop.example.com", Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now().Add(-time.Hour)})
	w.start()
	if a, _ := w.lock(); a.Host != offlineHost || a.Op != w.id {
		t.Fatalf("lock = %+v", a)
	}
	// A lock naming a host whose record carries no such move (a failed release).
	x := newOfflineWorld(t, oneEach...)
	if c, _, out := rawReq(t, http.MethodPost, instancesURL(x.fx, "bab", "shop"), x.token, `{"host":"beta.shop.example.com","region":"ie"}`); c != http.StatusCreated {
		t.Fatalf("create = %d %s", c, out)
	}
	put(x, moves.OfflineActive{Host: "beta.shop.example.com", Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now().Add(-time.Hour)})
	x.start()
	// A lock whose record carries another move of that host is not this one's.
	y := newOfflineWorld(t, oneEach...)
	put(y, moves.OfflineActive{Host: offlineHost, Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now().Add(-time.Hour)})
	y.start()
	// A young lock is believed even before its record carries the move: the
	// start door takes the lock, then writes the record.
	f := newOfflineWorld(t, oneEach...)
	if c, _, out := rawReq(t, http.MethodPost, instancesURL(f.fx, "bab", "shop"), f.token, `{"host":"beta.shop.example.com","region":"ie"}`); c != http.StatusCreated {
		t.Fatalf("create = %d %s", c, out)
	}
	put(f, moves.OfflineActive{Host: "beta.shop.example.com", Op: "mv_aaaaaaaaaaaaaaaaaaaaaaaaaa", Account: "bab", App: "shop", At: time.Now()})
	f.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, "beta.shop.example.com is moving offline. move one instance at a time")
	// An old lock is live while its record carries the move.
	g := newOfflineWorld(t, oneEach...)
	if _, code, out := g.startOther("beta.shop.example.com"); code != http.StatusAccepted {
		t.Fatalf("beta start = %d %s", code, out)
	}
	a, _ := g.lock()
	a.At = time.Now().Add(-24 * time.Hour)
	put(g, a)
	g.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, "beta.shop.example.com is moving offline. move one instance at a time")
	// An unreadable lock is not guessed at.
	z := newOfflineWorld(t, oneEach...)
	offlinePut(t, z.fx.rel, moves.OfflineActiveKey, []byte("not json"))
	z.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusBadGateway, "could not read the instance")
}

// A record nobody can read elsewhere does not block a start: the lock is the
// gate, not a listing of every record.
func TestOfflineStartIgnoresOtherRecords(t *testing.T) {
	w := newOfflineWorld(t, oneEach...)
	offlinePut(t, w.fx.rel, provision.RecordKey("bab", "shop", "junk.shop.example.com"), []byte("{not a record"))
	w.start()
}

// A failed record write does not prove nothing was written, so start never
// releases its lock on one. A write that lands and then reports an error is a
// start that succeeded; a write that does not land leaves the lock, which
// goes stale by itself. Either way no other host starts meanwhile.
func TestOfflineStartKeepsTheLockWhenTheRecordWriteFails(t *testing.T) {
	for _, landed := range []bool{true, false} {
		t.Run(fmt.Sprintf("landed=%v", landed), func(t *testing.T) {
			w := newOfflineWorld(t, oneEach...)
			if c, _, out := rawReq(t, http.MethodPost, instancesURL(w.fx, "bab", "shop"), w.token, `{"host":"beta.shop.example.com","region":"ie"}`); c != http.StatusCreated {
				t.Fatalf("create = %d %s", c, out)
			}
			w.fx.srv.cfg.Releases = &recordWriteFault{Store: w.fx.rel, key: provision.RecordKey("bab", "shop", offlineHost), land: landed}
			code, out := w.req(http.MethodPost, "", `{"region":"uk"}`)
			if landed {
				// The record carries the move: the start succeeded.
				if code != http.StatusAccepted {
					t.Fatalf("start = %d %s", code, out)
				}
				if v := w.view(code, out); v.State != provision.OfflineStopping || v.ID == "" {
					t.Fatalf("view = %+v", v)
				}
			} else if code != http.StatusBadGateway || errorText(out) != "could not write the move. try again" {
				t.Fatalf("start = %d %s", code, out)
			}
			a, ok := w.lock()
			if !ok || a.Host != offlineHost {
				t.Fatalf("lock = %+v %v, want it kept", a, ok)
			}
			// A second host's start is refused whichever way the write went.
			bu := instancesURL(w.fx, "bab", "shop") + "/beta.shop.example.com/move/offline"
			c, _, o := rawReq(t, http.MethodPost, bu, w.token, `{"region":"uk"}`)
			if c != http.StatusConflict || errorText(o) != offlineHost+" is moving offline. move one instance at a time" {
				t.Fatalf("beta start = %d %s", c, o)
			}
		})
	}
}