package provision

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

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

const (
	offID     = "mv_abcdefghijklmnopqrstuvwxyz"
	otherOff  = "mv_zyxwvutsrqponmlkjihgfedcba"
	offSealer = "box-a"
)

var offBoxes = []string{"box-a", "box-b"}

// offlineRecord creates a record that carries everything an offline move must
// leave alone: an acquisition marker, a pool, an environment, a pinned
// channel, a hibernate flag and a steering rule. Its region is "" (home).
func offlineRecord(t *testing.T, s release.Store) Record {
	t.Helper()
	hib := true
	rec := Record{
		Schema: 1, Host: "a.go.example.com", Slug: "s", App: "app", Account: "bab",
		Pool: "main", Region: "", Channel: "stable", Environment: "shop",
		Acquisition: "acq_0123456789abcdef", Hibernate: &hib, Steering: "nearest",
		CreatedAt: time.Date(2026, 9, 1, 8, 0, 0, 0, time.UTC), CreatedBy: "ops@example.com",
	}
	if err := CreateRecord(context.Background(), s, rec); err != nil {
		t.Fatal(err)
	}
	return rec
}

// offlineStep is one transition by name, so a test can walk the path and try
// every step from every state.
type offlineStep struct {
	name string
	mut  func() func(*Record) error
	from string // the one state it leaves
}

func offlineSteps(now time.Time) []offlineStep {
	src := MoveSource{Replication: "cold", NextDue: 99, FinalTXID: 42, Release: "v0.28.4", EnvDigest: "d1", Files: &MoveFilesDigest{Plain: "p", Sealed: "s", Bytes: 7, Entries: 2}}
	return []offlineStep{
		{"start", func() func(*Record) error { return OfflineStart(offID, "", "uk", "ops@example.com", now) }, ""},
		{"seal", func() func(*Record) error { return OfflineSeal(offID, offSealer, offBoxes) }, OfflineStopping},
		{"sealed", func() func(*Record) error { return OfflineSealedBy(offID, offSealer, src) }, OfflineSealing},
		{"flip", func() func(*Record) error { return OfflineFlip(offID) }, OfflineSealed},
		{"landed", func() func(*Record) error { return OfflineLandedBy(offID, "uk") }, OfflineLanding},
		{"open", func() func(*Record) error { return OfflineOpenTransition(offID, now.Add(30*time.Minute)) }, OfflineLanded},
		{"finishing", func() func(*Record) error { return OfflineFinishingTransition(offID) }, OfflineOpen},
		{"finish", func() func(*Record) error { return OfflineFinish(offID, now.Add(time.Hour)) }, OfflineFinishing},
	}
}

func offlineWalk(t *testing.T, s release.Store, rec Record, steps []offlineStep) *Record {
	t.Helper()
	var got *Record
	for _, st := range steps {
		var err error
		got, err = SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, st.mut())
		if err != nil {
			t.Fatalf("%s: %v", st.name, err)
		}
	}
	return got
}

func offlineGet(t *testing.T, s release.Store, rec Record) Record {
	t.Helper()
	got, err := GetRecord(context.Background(), s, rec.Account, rec.App, rec.Host)
	if err != nil {
		t.Fatal(err)
	}
	return *got
}

// asMap is a record as the generic JSON object every other reader sees.
func asMap(t *testing.T, rec Record) map[string]any {
	t.Helper()
	raw, err := json.Marshal(rec)
	if err != nil {
		t.Fatal(err)
	}
	var m map[string]any
	if err := json.Unmarshal(raw, &m); err != nil {
		t.Fatal(err)
	}
	return m
}

func TestOfflineFullLegalPathThroughTheWriter(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	wantState := []string{OfflineStopping, OfflineSealing, OfflineSealed, OfflineLanding, OfflineLanded, OfflineOpen, OfflineFinishing}
	wantRegion := []string{"", "", "", "uk", "uk", "uk", "uk"}

	for i, st := range steps[:len(steps)-1] {
		got, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, st.mut())
		if err != nil {
			t.Fatalf("%s: %v", st.name, err)
		}
		if got.Move == nil || !got.Move.Offline() || got.Move.State != wantState[i] || got.Region != wantRegion[i] {
			t.Fatalf("%s: move %+v region %q", st.name, got.Move, got.Region)
		}
		if got.Move.Legacy() {
			t.Fatalf("%s: an offline move reads as legacy", st.name)
		}
		if got.Move.UpdatedAt.IsZero() {
			t.Fatalf("%s: updated_at not stamped", st.name)
		}
		// What the box and the console read back is what the writer wrote.
		if again := offlineGet(t, s, rec); !reflect.DeepEqual(again.Move, got.Move) || again.Region != got.Region {
			t.Fatalf("%s: stored %+v, returned %+v", st.name, again.Move, got.Move)
		}
	}
	mid := offlineGet(t, s, rec)
	m := mid.Move
	if m.ID != offID || m.From != "" || m.To != "uk" || m.RequestedBy != "ops@example.com" || !m.RequestedAt.Equal(now) ||
		m.Sealer != offSealer || !reflect.DeepEqual(m.Boxes, offBoxes) || m.Incarnation != "" || m.Cut != "" ||
		m.Source == nil || m.Source.Release != "v0.28.4" || m.Source.EnvDigest != "d1" || m.Source.FinalTXID != 42 ||
		m.Source.Files == nil || m.Source.Files.Entries != 2 || m.OpenedAt == nil {
		t.Fatalf("the journal at finishing: %+v source %+v", m, m.Source)
	}

	got, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, steps[len(steps)-1].mut())
	if err != nil {
		t.Fatal(err)
	}
	if got.Move != nil || got.LastMove == nil || got.LastMove.ID != offID || got.LastMove.Outcome != "done" || !got.LastMove.At.Equal(now.Add(time.Hour)) || got.LastMove.OfflineS < 0 {
		t.Fatalf("finish: move %+v last %+v", got.Move, got.LastMove)
	}
	if got.Region != "uk" {
		t.Fatalf("finish: region %q", got.Region)
	}
}

// Review Focus 8 (§9.2): across a whole move the record differs only in
// region, move, last_move, record_seq (advanced) and move.updated_at. The
// record here carries an acquisition marker, a pool, an environment and the
// rest of what ordinary writers must leave alone; the comparison is of the
// whole JSON object, so a field a transition touched by accident shows.
func TestOfflineRecordDiffersOnlyWhereTheMoveSaysSo(t *testing.T) {
	for _, via := range []string{"done", "aborted after the flip", "aborted before the flip"} {
		t.Run(via, func(t *testing.T) {
			// The doors call the transitions with a plain context. A counter that
			// is already present advances without any context asking for it, so
			// the record starts with one: a first write under the sequence rule
			// puts record_seq into presence, as on any record a writer has touched.
			ctx := context.Background()
			s := release.NewDirStore(t.TempDir())
			rec := offlineRecord(t, s)
			if _, err := updateRecord(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, func(*Record) error { return nil }); err != nil {
				t.Fatal(err)
			}
			before := offlineGet(t, s, rec)
			if before.RecordSeq == 0 {
				t.Fatal("the record starts without record_seq")
			}
			beforeMap := asMap(t, before)

			now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
			steps := offlineSteps(now)
			upto := len(steps)
			switch via {
			case "aborted after the flip":
				upto = 5 // through landed
			case "aborted before the flip":
				upto = 3 // through sealed
			}
			seqs := []uint64{}
			for _, st := range steps[:upto] {
				got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, st.mut())
				if err != nil {
					t.Fatalf("%s: %v", st.name, err)
				}
				seqs = append(seqs, got.RecordSeq)
			}
			if via != "done" {
				if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
					t.Fatal(err)
				}
				if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Hour))); err != nil {
					t.Fatal(err)
				}
			}
			after := offlineGet(t, s, rec)
			if after.Move != nil || after.LastMove == nil {
				t.Fatalf("the move did not end: %+v %+v", after.Move, after.LastMove)
			}
			for i := 1; i < len(seqs); i++ {
				if seqs[i] <= seqs[i-1] {
					t.Fatalf("record_seq did not advance across transitions: %v", seqs)
				}
			}
			if after.RecordSeq <= before.RecordSeq {
				t.Fatalf("record_seq %d -> %d", before.RecordSeq, after.RecordSeq)
			}

			afterMap := asMap(t, after)
			allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true}
			keys := map[string]bool{}
			for k := range beforeMap {
				keys[k] = true
			}
			for k := range afterMap {
				keys[k] = true
			}
			for k := range keys {
				if reflect.DeepEqual(beforeMap[k], afterMap[k]) {
					continue
				}
				if !allowed[k] {
					t.Errorf("field %q changed: %v -> %v", k, beforeMap[k], afterMap[k])
				}
			}
			if _, ok := afterMap["move"]; ok {
				t.Errorf("move left on the record at rest: %v", afterMap["move"])
			}
			wantRegion := ""
			if via == "done" {
				wantRegion = "uk"
			}
			if after.Region != wantRegion {
				t.Errorf("region %q, want %q", after.Region, wantRegion)
			}
			for _, k := range []string{"acq", "pool", "environment", "channel", "hibernate", "steering", "created_at", "created_by", "slug"} {
				if _, ok := afterMap[k]; !ok {
					t.Errorf("field %q is gone", k)
				}
			}
		})
	}
}

// Every transition, tried from every state but its own, is refused with
// ErrMoveConflict and leaves the record as it was.
func TestOfflineEveryIllegalPairIsRefused(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	// Each state is reached by walking the first n steps; reverting by an
	// abort from stopping.
	type at struct {
		state string
		walk  []offlineStep
		abort bool
	}
	states := []at{
		{OfflineStopping, steps[:1], false}, {OfflineSealing, steps[:2], false}, {OfflineSealed, steps[:3], false},
		{OfflineLanding, steps[:4], false}, {OfflineLanded, steps[:5], false}, {OfflineOpen, steps[:6], false},
		{OfflineFinishing, steps[:7], false}, {OfflineReverting, steps[:1], true},
	}

	for _, at := range states {
		for _, st := range steps[1:] { // start has its own test
			legal := st.from == at.state
			s := release.NewDirStore(t.TempDir())
			rec := offlineRecord(t, s)
			offlineWalk(t, s, rec, at.walk)
			if at.abort {
				if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
					t.Fatal(err)
				}
			}
			before := offlineGet(t, s, rec)
			_, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, st.mut())
			switch {
			case legal && err != nil:
				t.Errorf("%s from %s: %v", st.name, at.state, err)
			case !legal && st.name == "finish" && at.state == OfflineReverting:
				if err != nil {
					t.Errorf("finish from reverting: %v", err)
				}
			case !legal && !errors.Is(err, ErrMoveConflict):
				t.Errorf("%s from %s: err %v, want ErrMoveConflict", st.name, at.state, err)
			}
			if err != nil {
				if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
					t.Errorf("%s from %s: a refusal changed the record", st.name, at.state)
				}
			}
		}
	}
}

func TestOfflineWrongIDIsRefusedEverywhere(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	wrong := map[string]func() func(*Record) error{
		"seal":      func() func(*Record) error { return OfflineSeal(otherOff, offSealer, offBoxes) },
		"sealed":    func() func(*Record) error { return OfflineSealedBy(otherOff, offSealer, MoveSource{}) },
		"flip":      func() func(*Record) error { return OfflineFlip(otherOff) },
		"landed":    func() func(*Record) error { return OfflineLandedBy(otherOff, "uk") },
		"open":      func() func(*Record) error { return OfflineOpenTransition(otherOff, now) },
		"finishing": func() func(*Record) error { return OfflineFinishingTransition(otherOff) },
		"finish":    func() func(*Record) error { return OfflineFinish(otherOff, now) },
		"abort":     func() func(*Record) error { return OfflineAbort(otherOff) },
	}
	// Each wrong-id step is tried in the state where its right-id twin is legal.
	for i, st := range steps[1:] {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		offlineWalk(t, s, rec, steps[:i+1])
		before := offlineGet(t, s, rec)
		if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, wrong[st.name]()); !errors.Is(err, ErrMoveConflict) {
			t.Errorf("%s with another move's id: %v", st.name, err)
		}
		if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
			t.Errorf("%s with another move's id changed the record", st.name)
		}
	}
	// Abort from a legal state with the wrong id, too.
	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:1])
	if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, wrong["abort"]()); !errors.Is(err, ErrMoveConflict) {
		t.Errorf("abort with another move's id: %v", err)
	}
	// No move at all.
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	for name, mk := range wrong {
		if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, mk()); !errors.Is(err, ErrMoveConflict) {
			t.Errorf("%s on a record with no move: %v", name, err)
		}
	}
}

func TestOfflineStartRefusals(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)

	start := func(t *testing.T, s release.Store, rec Record, id, from, to string) (*Record, error) {
		t.Helper()
		return SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(id, from, to, "ops@example.com", now))
	}
	t.Run("stamps the head and leaves the region", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		got, err := start(t, s, rec, offID, "", "uk")
		if err != nil {
			t.Fatal(err)
		}
		m := got.Move
		if m.Kind != MoveKindOffline || m.ID != offID || m.From != "" || m.To != "uk" || m.State != OfflineStopping || m.RequestedBy != "ops@example.com" || !m.RequestedAt.Equal(now) || got.Region != "" {
			t.Fatalf("head %+v region %q", m, got.Region)
		}
		if m.Cut != "" || m.Incarnation != "" || m.Inputs != nil || m.Source != nil || m.CutPhase != "" || m.Sealer != "" || m.Boxes != nil {
			t.Fatalf("start stamped more than the head: %+v", m)
		}
	})
	t.Run("in flight", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		if _, err := start(t, s, rec, offID, "", "uk"); err != nil {
			t.Fatal(err)
		}
		if _, err := start(t, s, rec, otherOff, "", "us"); !errors.Is(err, ErrMoveConflict) {
			t.Fatalf("second start: %v", err)
		}
		if got := offlineGet(t, s, rec); got.Move.ID != offID || got.Move.To != "uk" {
			t.Fatalf("a refused start replaced the move: %+v", got.Move)
		}
	})
	t.Run("a legacy or Phase 1 move in flight", func(t *testing.T) {
		for name, mut := range map[string]func(*Record) error{
			"legacy": MoveRequest("uk", now),
			"phase1": MoveStartTransition(MoveStart{ID: offID, Incarnation: strings.Repeat("ab", 32), From: "", To: "uk", Cut: "auto", Now: now}),
		} {
			s := release.NewDirStore(t.TempDir())
			rec := offlineRecord(t, s)
			if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil {
				t.Fatalf("%s: %v", name, err)
			}
			if _, err := start(t, s, rec, otherOff, "", "us"); !errors.Is(err, ErrMoveConflict) {
				t.Fatalf("%s: offline start over it: %v", name, err)
			}
		}
	})
	t.Run("same region, wrong from, bad id", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		for name, c := range map[string][3]string{
			"same region": {offID, "", ""},
			"wrong from":  {offID, "us", "uk"},
			"short id":    {"mv_short", "", "uk"},
			"empty id":    {"", "", "uk"},
			"rollback id": {"rb_abcdefghijklmnopqrstuvwxyz", "", "uk"},
		} {
			if _, err := start(t, s, rec, c[0], c[1], c[2]); !errors.Is(err, ErrMoveConflict) {
				t.Errorf("%s: %v", name, err)
			}
		}
		if got := offlineGet(t, s, rec); got.Move != nil {
			t.Fatalf("a refused start left a journal: %+v", got.Move)
		}
	})
	t.Run("replica", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		r := Record{Schema: 1, Host: "r.go.example.com", Slug: "s", App: "app", Account: "bab", Region: "", Channel: "stable", CreatedAt: now, Role: "replica"}
		if err := CreateRecord(ctx, s, r); err != nil {
			t.Fatal(err)
		}
		if _, err := start(t, s, r, offID, "", "uk"); !errors.Is(err, ErrMoveConflict) {
			t.Fatalf("replica: %v", err)
		}
	})
	t.Run("seed", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		// A seeded record that is not placed: the writer under test refuses
		// before any journal exists, as MoveRequest does.
		unplaced := rec
		unplaced.Seed = &SeedSource{ID: "sd_1", Kind: seeds.KindReplica, State: seeds.StateReady, By: "ops@example.com", At: now}
		err := OfflineStart(offID, "", "uk", "ops@example.com", now)(&unplaced)
		var refusal *SeedMoveError
		if !errors.Is(err, ErrMoveConflict) || !errors.As(err, &refusal) {
			t.Fatalf("seed: %v", err)
		}
		if unplaced.Move != nil {
			t.Fatal("a refused start left a journal")
		}
	})
	t.Run("open run", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		running := rec
		running.ActiveRun = &RunState{ID: "run_1", State: RunPicked, Deadline: now.Add(time.Hour)}
		err := OfflineStart(offID, "", "uk", "ops@example.com", now)(&running)
		var open *MoveRunOpenError
		if !errors.As(err, &open) {
			t.Fatalf("open run: %v", err)
		}
		if running.Move != nil {
			t.Fatal("a refused start left a journal")
		}
	})
}

func TestOfflineSealAndSealedByChecks(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	ctx := context.Background()

	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:1])
	for name, mut := range map[string]func(*Record) error{
		"no sealer": OfflineSeal(offID, "", offBoxes),
		"no boxes":  OfflineSeal(offID, offSealer, nil),
	} {
		if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); !errors.Is(err, ErrMoveConflict) {
			t.Errorf("%s: %v", name, err)
		}
	}
	// The sealer is one of the boxes.
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, "box-z", offBoxes)); !errors.Is(err, ErrMoveConflict) || !strings.Contains(err.Error(), refusedOfflineSeal) {
		t.Errorf("a sealer outside the boxes: %v", err)
	}
	// The seal copies the caller's slice.
	boxes := []string{"box-a", "box-b"}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSeal(offID, offSealer, boxes)); err != nil {
		t.Fatal(err)
	}
	boxes[0] = "changed"
	if got := offlineGet(t, s, rec); got.Move.Boxes[0] != "box-a" {
		t.Fatalf("the journal aliases the caller's slice: %v", got.Move.Boxes)
	}

	// Only the named sealer records what it sealed, and the source is copied.
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSealedBy(offID, "box-b", MoveSource{})); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("another box sealing: %v", err)
	}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSealedBy(offID, "", MoveSource{})); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("no sealer: %v", err)
	}
	files := &MoveFilesDigest{Plain: "p", Bytes: 1}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineSealedBy(offID, offSealer, MoveSource{Files: files, Release: "v1"})); err != nil {
		t.Fatal(err)
	}
	files.Plain = "changed"
	if got := offlineGet(t, s, rec); got.Move.Source.Files.Plain != "p" || got.Move.Source.Release != "v1" {
		t.Fatalf("the journal aliases the caller's source: %+v", got.Move.Source)
	}
}

func TestOfflineFlipAndLandedGuardTheRegion(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	ctx := context.Background()

	// A record whose region moved under the move is not flipped.
	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:3])
	if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
		t.Fatal(err)
	}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFlip(offID)); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("flip over a changed region: %v", err)
	}

	// landed is guarded by the record's region == to.
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:4]) // through flip: region uk, landing
	for _, to := range []string{"", "us"} {
		if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineLandedBy(offID, to)); !errors.Is(err, ErrMoveConflict) {
			t.Errorf("landed by %q: %v", to, err)
		}
	}
	// An abort puts the region back; a landing that raced it is refused.
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
		t.Fatal(err)
	}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineLandedBy(offID, "uk")); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("landed after abort: %v", err)
	}
}

func TestOfflineAbort(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)

	// Legal from stopping (1 step walked) through landed (5), region back
	// from landing and landed only.
	for walked, c := range map[int]struct {
		state  string
		region string
	}{
		1: {OfflineStopping, ""}, 2: {OfflineSealing, ""}, 3: {OfflineSealed, ""},
		4: {OfflineLanding, "uk"}, 5: {OfflineLanded, "uk"},
	} {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		got := offlineWalk(t, s, rec, steps[:walked])
		if got.Move.State != c.state || got.Region != c.region {
			t.Fatalf("setup %s: %s %q", c.state, got.Move.State, got.Region)
		}
		got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
		if err != nil {
			t.Fatalf("abort from %s: %v", c.state, err)
		}
		if got.Move.State != OfflineReverting || got.Region != "" || !got.Move.Offline() || got.Move.ID != offID {
			t.Fatalf("abort from %s: %+v region %q", c.state, got.Move, got.Region)
		}
		// A second abort is a success that writes nothing: the stored record
		// is the same bytes, record_seq and updated_at included.
		key := RecordKey(rec.Account, rec.App, rec.Host)
		rawBefore, etagBefore, err := s.Get(ctx, key)
		if err != nil {
			t.Fatal(err)
		}
		again, err := SetRecordMove(WithRecordSequence(ctx), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))
		rawAfter, etagAfter, gerr := s.Get(ctx, key)
		if gerr != nil || !bytes.Equal(rawBefore, rawAfter) || etagBefore != etagAfter {
			t.Fatalf("second abort from %s wrote the record: %v", c.state, gerr)
		}
		if err != nil || again.Move.State != OfflineReverting || again.Region != "" {
			t.Fatalf("second abort from %s: %v %+v", c.state, err, again)
		}
		// The way back to rest is finish.
		done, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now.Add(time.Minute)))
		if err != nil || done.Move != nil || done.LastMove == nil || done.LastMove.Outcome != "aborted" || done.LastMove.ID != offID || done.Region != "" {
			t.Fatalf("finish after abort from %s: %v %+v", c.state, err, done)
		}
	}

	// Refused from open and finishing, and nothing changes.
	for walked, state := range map[int]string{6: OfflineOpen, 7: OfflineFinishing} {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		offlineWalk(t, s, rec, steps[:walked])
		before := offlineGet(t, s, rec)
		if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); !errors.Is(err, ErrMoveConflict) {
			t.Fatalf("abort from %s: %v", state, err)
		}
		if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
			t.Fatalf("abort from %s changed the record", state)
		}
	}
}

func TestOfflineFinishFromBothAndTheRegionGuard(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now) // open at now+30m, finish at now+1h

	// Done: offline time runs from the request to the open.
	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	got := offlineWalk(t, s, rec, steps)
	if got.LastMove.Outcome != "done" {
		t.Fatalf("outcome %q", got.LastMove.Outcome)
	}
	if got.LastMove.OfflineS != 1800 || got.LastMove.At != now.Add(time.Hour) {
		t.Fatalf("offline_s %v at %v, want 1800 (request to open) at the finish", got.LastMove.OfflineS, got.LastMove.At)
	}

	// finish refuses a region that is not where the outcome says.
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:7])
	if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
		t.Fatal(err)
	}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now)); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("done finish with the wrong region: %v", err)
	}
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:5])
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
		t.Fatal(err)
	}
	if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "uk"; return nil }); err != nil {
		t.Fatal(err)
	}
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineFinish(offID, now)); !errors.Is(err, ErrMoveConflict) {
		t.Fatalf("aborted finish with the target's region: %v", err)
	}
}

// A record written before this change must parse and re-encode byte for byte.
// The bytes are written out by hand, in the order the struct has always
// encoded them: a record at rest, one with a legacy journal, and one with a
// Phase 1 journal carrying the source baselines.
func TestOfflineFieldsAreAdditiveOnTheWire(t *testing.T) {
	const atRest = `{"record_seq":4,"acq":"acq_0123456789abcdef","schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","pool":"main","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","created_by":"ops@example.com","last_move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","outcome":"done","at":"2026-10-06T10:00:00Z","offline_s":61.5},"hibernate":true,"environment":"shop","steering":"nearest"}`
	const legacy = `{"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"to":"uk","state":"ready","requested_at":"2026-10-06T09:00:00Z","ready_at":"2026-10-06T09:01:00Z","updated_at":"2026-10-06T09:01:00Z"}}`
	const phase1 = `{"record_seq":9,"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","incarnation":"abababababababababababababababababababababababababababababababab","from":"ie","to":"uk","cut":"auto","state":"cutting","requested_by":"ops@example.com","requested_at":"2026-10-06T09:00:00Z","updated_at":"2026-10-06T09:05:00Z","cut_phase":"stopped","cut_at":"2026-10-06T09:04:00Z","cut_budget_s":600,"source":{"replication":"cold","next_due":99,"final_txid":42,"db_sha256":"abc","files":{"plain":"p","sealed":"s","bytes":7,"entries":2},"timers_sha256":"t","evidence":"e"}}}`
	for name, raw := range map[string]string{"at rest": atRest, "legacy move": legacy, "phase 1 move": phase1} {
		rec, err := DecodeRecord([]byte(raw))
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		if rec.Move.Offline() {
			t.Fatalf("%s: a move written before offline moves reads as offline", name)
		}
		if rec.Move != nil && (rec.Move.Kind != "" || rec.Move.Sealer != "" || rec.Move.Boxes != nil) {
			t.Fatalf("%s: the new fields are not empty: %+v", name, rec.Move)
		}
		if rec.Move != nil && rec.Move.Source != nil && (rec.Move.Source.Release != "" || rec.Move.Source.EnvDigest != "") {
			t.Fatalf("%s: the new source fields are not empty", name)
		}
		out, err := json.Marshal(rec)
		if err != nil {
			t.Fatal(err)
		}
		if string(out) != raw {
			t.Errorf("%s did not round-trip byte for byte:\n got %s\nwant %s", name, out, raw)
		}
	}

	// And an offline journal round-trips too, with its fields in the journal.
	const offline = `{"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"uk","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","to":"uk","state":"offline-landing","requested_by":"ops@example.com","requested_at":"2026-10-21T09:00:00Z","updated_at":"2026-10-21T09:05:00Z","source":{"final_txid":42,"release":"v0.28.4","env_digest":"d1"},"kind":"offline","sealer":"box-a","boxes":["box-a","box-b"]}}`
	rec, err := DecodeRecord([]byte(offline))
	if err != nil {
		t.Fatal(err)
	}
	if !rec.Move.Offline() || rec.Move.Sealer != "box-a" || len(rec.Move.Boxes) != 2 || rec.Move.Source.Release != "v0.28.4" {
		t.Fatalf("offline journal: %+v", rec.Move)
	}
	if out, _ := json.Marshal(rec); string(out) != offline {
		t.Errorf("offline journal did not round-trip:\n got %s\nwant %s", out, offline)
	}
}

// Kinds never cross: no legacy and no Phase 1 transition accepts an offline
// move, in any of its states, and none changes the record.
func TestEveryLegacyAndPhase1TransitionRefusesAnOfflineMove(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	// An empty incarnation and the offline move's own id: the guard that would
	// match if the kind were not checked.
	g := MoveGuard{ID: offID}
	gInc := MoveGuard{ID: offID, Incarnation: strings.Repeat("ab", 32)}
	in := MoveInputs{}
	refused := map[string]func() func(*Record) error{
		"MoveRequest": func() func(*Record) error { return MoveRequest("us", now) },
		"MoveStartTransition": func() func(*Record) error {
			return MoveStartTransition(MoveStart{ID: otherOff, Incarnation: strings.Repeat("ab", 32), From: "", To: "us", Cut: "auto", Now: now})
		},
		"MoveAbort":                 func() func(*Record) error { return MoveAbort() },
		"MoveMarkCutting":           func() func(*Record) error { return MoveMarkCutting() },
		"MoveComplete":              func() func(*Record) error { return MoveComplete() },
		"MoveAbortComplete":         func() func(*Record) error { return MoveAbortComplete() },
		"MovePromoteReady":          func() func(*Record) error { return MovePromoteReady(now) },
		"MoveGuardedPromoteReady":   func() func(*Record) error { return MoveGuardedPromoteReady(g, now, in) },
		"MoveGuardedCut":            func() func(*Record) error { return MoveGuardedCut(g) },
		"MoveGuardedRepin":          func() func(*Record) error { return MoveGuardedRepin(g, in) },
		"MoveGuardedAbort":          func() func(*Record) error { return MoveGuardedAbort(g) },
		"MoveGuardedAbortComplete":  func() func(*Record) error { return MoveGuardedAbortComplete(g, now) },
		"MoveAdmit":                 func() func(*Record) error { return MoveAdmit(g, now, 600, now.Add(time.Hour), 3) },
		"MoveBackToReady":           func() func(*Record) error { return MoveBackToReady(g) },
		"MoveRecordSource":          func() func(*Record) error { return MoveRecordSource(g, MoveSource{}) },
		"MoveFlip":                  func() func(*Record) error { return MoveFlip(g, now, now.Add(time.Hour)) },
		"MoveCommit":                func() func(*Record) error { return MoveCommit(g, now) },
		"MoveClose":                 func() func(*Record) error { return MoveClose(g, now) },
		"MoveRevert":                func() func(*Record) error { return MoveRevert(g, "r") },
		"MoveMarkStuck":             func() func(*Record) error { return MoveMarkStuck(g, StuckBaselineMismatch) },
		"MoveRetry":                 func() func(*Record) error { return MoveRetry(g) },
		"MoveRollBack":              func() func(*Record) error { return MoveRollBack(g, "rb_abcdefghijklmnopqrstuvwxyz", "r", now) },
		"guarded, real incarnation": func() func(*Record) error { return MoveGuardedAbort(gInc) },
	}
	// The guard itself refuses, whatever the transition would say after it:
	// this pins MoveGuard.check, not each transition's own state check.
	for _, st := range []string{OfflineStopping, OfflineLanding, OfflineOpen, OfflineReverting} {
		offRec := Record{Move: &MoveState{Kind: MoveKindOffline, ID: offID, State: st}}
		err := g.check(&offRec)
		if !errors.Is(err, ErrMoveConflict) || !strings.Contains(err.Error(), refusedOfflineKind) {
			t.Errorf("MoveGuard.check on an offline move in %s: %v", st, err)
		}
	}
	// Walk one record through every offline state and try everything on it.
	s := release.NewDirStore(t.TempDir())
	rec := offlineRecord(t, s)
	check := func(label string) {
		t.Helper()
		before := offlineGet(t, s, rec)
		for name, mk := range refused {
			if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, mk()); !errors.Is(err, ErrMoveConflict) {
				t.Errorf("%s: %s: err %v, want ErrMoveConflict", label, name, err)
			}
		}
		if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
			t.Errorf("%s: a refused transition changed the record: %+v -> %+v", label, before.Move, after.Move)
		}
	}
	for _, st := range steps[:len(steps)-1] {
		offlineWalk(t, s, rec, []offlineStep{st})
		check(st.name)
	}
	// The reverting state, reached by abort from sealed.
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	offlineWalk(t, s, rec, steps[:3])
	if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); err != nil {
		t.Fatal(err)
	}
	check("reverting")
}

// And the other way: no offline transition accepts a legacy or Phase 1 move.
func TestOfflineTransitionsRefuseOtherKinds(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	for name, start := range map[string]func(*Record) error{
		"legacy": MoveRequest("uk", now),
		"phase1": MoveStartTransition(MoveStart{ID: offID, Incarnation: strings.Repeat("ab", 32), From: "", To: "uk", Cut: "auto", Now: now}),
	} {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, start); err != nil {
			t.Fatal(err)
		}
		before := offlineGet(t, s, rec)
		for _, st := range offlineSteps(now)[1:] {
			if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, st.mut()); !errors.Is(err, ErrMoveConflict) {
				t.Errorf("%s move: offline %s: %v", name, st.name, err)
			}
		}
		if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineAbort(offID)); !errors.Is(err, ErrMoveConflict) {
			t.Errorf("%s move: offline abort: %v", name, err)
		}
		if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {
			t.Errorf("%s move: an offline transition changed it", name)
		}
	}
}

func TestMoveStateOfflineIsNilSafe(t *testing.T) {
	var m *MoveState
	if m.Offline() {
		t.Fatal("nil move is offline")
	}
	if (&MoveState{}).Offline() || (&MoveState{Kind: "other"}).Offline() || !(&MoveState{Kind: MoveKindOffline}).Offline() {
		t.Fatal("Offline reads the wrong field")
	}
}

// The refusals an operator can read are the recorded sentences (copy rows
// MV-O34 to MV-O43 and the existing MV-N rows), not free text, and none has an
// em dash. A door maps them; this holds the provision side to its rows.
func TestOfflineRefusalsAreTheRecordedSentences(t *testing.T) {
	now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)
	steps := offlineSteps(now)
	ctx := context.Background()
	refuse := func(t *testing.T, s release.Store, rec Record, mut func(*Record) error, want string) {
		t.Helper()
		_, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut)
		if !errors.Is(err, ErrMoveConflict) {
			t.Fatalf("err %v, want ErrMoveConflict", err)
		}
		if got := strings.TrimPrefix(err.Error(), ErrMoveConflict.Error()+": "); got != want {
			t.Errorf("refusal %q, want %q", got, want)
		}
		if strings.ContainsRune(err.Error(), '\u2014') {
			t.Errorf("em dash in %q", err)
		}
	}
	fresh := func(t *testing.T, walked int) (release.Store, Record) {
		s := release.NewDirStore(t.TempDir())
		rec := offlineRecord(t, s)
		offlineWalk(t, s, rec, steps[:walked])
		return s, rec
	}

	s, rec := fresh(t, 0)
	refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes), refusedNoMove)
	refuse(t, s, rec, OfflineStart(offID, "us", "uk", "ops", now), "the instance is in home, not us")
	refuse(t, s, rec, OfflineStart(offID, "", "", "ops", now), "the instance is already in home")
	refuse(t, s, rec, OfflineStart("mv_short", "", "uk", "ops", now), refusedMalformedStart)

	s, rec = fresh(t, 1)
	refuse(t, s, rec, OfflineSeal(otherOff, offSealer, offBoxes), refusedOtherOfflineMove)
	refuse(t, s, rec, OfflineFlip(offID), "this offline move is offline-stopping. this step follows offline-sealed")
	refuse(t, s, rec, OfflineSeal(offID, "", offBoxes), refusedOfflineSeal)
	refuse(t, s, rec, OfflineFinish(offID, now), "this offline move is offline-stopping. it finishes from finishing or reverting")

	s, rec = fresh(t, 2)
	refuse(t, s, rec, OfflineSealedBy(offID, "box-b", MoveSource{}), "this offline move is sealed by box-a")

	s, rec = fresh(t, 6)
	refuse(t, s, rec, OfflineAbort(offID), "this offline move is offline-open. it cannot be aborted once it is open")

	s, rec = fresh(t, 3)
	if _, err := updateRecord(ctx, s, rec.Account, rec.App, rec.Host, func(r *Record) error { r.Region = "us"; return nil }); err != nil {
		t.Fatal(err)
	}
	refuse(t, s, rec, OfflineFlip(offID), "the record's region is us, which this step does not expect. the instance changed under the move")

	// A legacy move meets an offline step.
	s = release.NewDirStore(t.TempDir())
	rec = offlineRecord(t, s)
	if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, MoveRequest("uk", now)); err != nil {
		t.Fatal(err)
	}
	refuse(t, s, rec, OfflineSeal(offID, offSealer, offBoxes), refusedNotOffline)

	// And the other way: the Phase 1 guard on an offline move.
	s, rec = fresh(t, 1)
	refuse(t, s, rec, MoveGuardedCut(MoveGuard{ID: offID}), refusedOfflineKind)
}