package host

import (
	"context"
	"crypto/ecdh"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/carlosframework/platform/internal/park"
	"io"
	"io/fs"
	"log"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// The target's readiness report (moves spec §7.3, T1 to T12, and the cut
// guard's T13), behind
// CARLOS_MOVES_PHASE1. The named target box (the claim's op.intent.target)
// writes it every pass for every move into its placement that is preparing
// or ready; no other box writes one. Each check is its own small function
// returning a moves.Check. The reasons are rows of the copy record, pending
// review; they reach `carlos move status` through the report.
const (
	t1FloorUnreadable = "the moves protocol floor could not be read"
	t1LocalUnreadable = "this box cannot read its own copy of the moves protocol floor"
	t1NotArmed        = "this box has not armed the moves protocol floor yet"
	t2NoConsole       = "the console has not published its secrets key fingerprint yet"
	t2NoKey           = "this box has no secrets key"
	t2Differs         = "this box's secrets key is not the console's"
	t3NoSource        = "the source has not published a readiness report yet" // MV-N62's text
	t3NoCompose       = "this box could not compose the launch environment of the held row"
	// t3OldReport is MV-N513: the source's env report is version 1, whose
	// pin cannot tell an instance's box variable from the box's
	// (moves.EnvReportVersion).
	t3OldReport   = "the source reported its launch environment in an older format"
	t5NoArchive   = "the source has published no archive yet" // MV-N73
	t5Unreadable  = "this box could not read the archive the source published"
	t5Refused     = "the archive does not open with this box's key or does not match its manifest"
	t5NotPinned   = "the source staged its archive without a fixed version id"
	t6Unmeasured  = "this box could not measure its free disk space"
	t7NoRelease   = "the source has not reported its release yet" // MV-N76
	t7Waiting     = "waiting for the root stage pass"             // MV-N75
	t7Failed      = "this box could not stage the source's release"
	t7NotProven   = "this box requires a proof for the source's release and has not accepted one"
	t7NotStaged   = "the release is not staged on this box yet" // MV-N455: T7 under the claim, the version store not writable (heldReleaseStaged)
	t9NotThisMove = "the host claim is not held by this move"
	t9Incarnation = "the host claim names another incarnation of this instance"
	t9NoMoveJSON  = "the staging lane has no move.json yet"
	t9OtherMove   = "the staging lane holds another move's move.json" // MV-N9, aligned
	t10NoBucket   = "the source region publishes no data bucket"
	t10Failed     = "this box cannot write the return lane in the source's bucket"
	t11NoReplica  = "no replica has been staged yet"
	t11Unmeasured = "the database size could not be measured" // MV-N69
	t13NotArmed   = "this box cannot land the move or roll it back yet"
)

func t1Below(min int) string {
	return fmt.Sprintf("this box speaks moves protocol %d, below the floor's %d", moves.ProtocolVersion, min)
}
func t3Differs(names []string) string {
	return "the launch environment differs from the source's: " + strings.Join(names, ", ")
}
func t3Shadowed(name string) string {
	return name + " in the carried .env overrides the rendered configuration"
}
func t4Failed(prefix string) string {
	return "the store bound as " + prefix + " did not answer the probe from this box"
}
func t6NoRoom(needGiB float64) string {
	return fmt.Sprintf("this box needs %.1f GiB free for the move without filling its disk past 85%%", needGiB)
}
func t8Count(n int, to string) string {
	if n == 0 {
		return "no box publishes " + to
	}
	return fmt.Sprintf("%d boxes publish %s, and a move needs exactly one", n, to)
}
func t9Untouched(what string) string { return "the host claim has not touched " + what + " yet" } // MV-N74
func t12NotQuiet(region string, until time.Time) string {
	return "the instance left " + region + " recently and can move back after " + until.UTC().Format(time.RFC3339)
}

// moveFleet is the fleet as the edges files publish it: per placement, the
// box whose coordinates a move uses (the lexically first, as moveTargets
// picks) and every box that publishes it (T8). The home publishes no
// placement code, so it is kept under "", where a move out of the home
// finds its source bucket (T10).
type moveFleet struct {
	targets map[string]edgesFile
	boxes   map[string][]string
}

func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) {
	bodies, err := loadEdgesFileBodies(ctx, s)
	if err != nil {
		return moveFleet{}, err
	}
	f := moveFleet{targets: map[string]edgesFile{}, boxes: map[string][]string{}}
	for _, body := range bodies {
		var t edgesFile
		if err := json.Unmarshal(body, &t); err != nil || t.V < 1 {
			continue
		}
		if !containsString(f.boxes[t.Placement], t.Box) {
			f.boxes[t.Placement] = append(f.boxes[t.Placement], t.Box)
		}
		if t.DataBucket == "" {
			continue
		}
		if prev, dup := f.targets[t.Placement]; dup && prev.Box <= t.Box {
			continue
		}
		f.targets[t.Placement] = t
	}
	return f, nil
}

// targetReportPass writes the target report for every move into this box's
// placement whose claim names this box as the target.
func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, recs []provision.Record, fleet moveFleet) {
	live := map[string]bool{}
	defer w.evictArchives(live)
	defer w.retireHeldRefusals(live)
	for _, rec := range recs {
		m := rec.Move
		if m == nil || m.Legacy() || m.To != w.cfg.Region {
			continue
		}
		// A move into this box is live from preparing through the cut and
		// the landing (ruling G26): the held refusal a landing's configured
		// phase writes, and T5's verdict, stay until the move is over. The
		// report itself is written only while preparing or ready. An
		// aborting move is over for this box: it never flipped, so no
		// landing wrote a refusal, and its record stays aborting only until
		// the lane park is acknowledged (Plan 5 Task 1).
		if m.State == provision.MoveAborting {
			continue
		}
		live[m.ID] = true
		if m.State != provision.MovePreparing && m.State != provision.MoveReady {
			continue
		}
		if w.lost[rec.Host] {
			continue // the lost-box gate (movespass.go lostHere)
		}
		if m.State == provision.MovePreparing && rec.Seed.MoveBlock() != "" {
			continue // movesPass's seed rule (spec §5.8), which logged it
		}
		if err := w.targetReport(ctx, s, rec, fleet); err != nil {
			w.logTrouble(rec.Host+" (target)", err)
			continue
		}
		w.trouble.Delete(rec.Host + " (target)")
	}
}

// evictArchives drops T5's verdicts (and the carried .env they hold) for
// every move no longer preparing or ready into this box.
func (w *movesWorker) evictArchives(live map[string]bool) {
	w.mu.Lock()
	defer w.mu.Unlock()
	for id := range w.archives {
		if !live[id] {
			delete(w.archives, id)
		}
	}
}

// retireHeldRefusals retires the held-stage refusal (D3-held) of every
// move no longer preparing or ready into this box, with the rest of its
// staging state: live is the same set evictArchives keeps. A record root
// wrote cannot be removed here (the agent does not own it), which is not
// worth a line: the root stage pass retires those (retireEndedHeldRefusals).
func (w *movesWorker) retireHeldRefusals(live map[string]bool) {
	if w.cfg.DataDir == "" {
		return
	}
	retired, err := release.RetireHeldRefusals(w.cfg.DataDir, func(r release.Refusal) bool { return live[r.MoveOp] })
	for _, host := range retired {
		log.Printf("moves: %s: retired the held refusal of a move that is over", host)
	}
	if err != nil && !errors.Is(err, fs.ErrPermission) {
		w.logTrouble("held refusals", err)
	}
}

// verifiedArchive is what T5 learned from one published archive, kept per
// exact version so a pass does not open the same archive again.
type verifiedArchive struct {
	ref     moves.ArchiveRef
	check   moves.Check
	plain   string
	entries int
	bytes   int64
	env     []byte // the carried <host>.env, nil when the archive has none
}

// heldEnv is T3's composition: the environment, the names a carried .env
// shadows, and the input versions rendered.
type heldEnv struct {
	env      map[string]string
	shadowed []string
	inputs   provision.MoveInputs
}

func (w *movesWorker) targetReport(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
	cfg := w.cfg
	id := rec.Move.ID
	obs, err := hostclaim.Read(ctx, s, rec.Host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent || obs.Claim.Op.ID != id {
		w.notice(rec.Host+"#target", "moves: %s: the host claim does not name move %s; no target report", rec.Host, id)
		return nil
	}
	// T8 and T9 bind to the named target: only it writes the report.
	if target := obs.Claim.Op.Intent["target"]; cfg.Box == "" || target != cfg.Box {
		w.notice(rec.Host+"#target", "moves: %s: this box is not the move's target (%q); no target report", rec.Host, target)
		return nil
	}
	source, err := readSourceReport(ctx, s, rec)
	if err != nil {
		return err
	}
	now := cfg.Now().UTC()
	rep := moves.TargetReport{V: 1, Box: cfg.Box, Region: cfg.Region, MoveID: id, At: now,
		Checks: map[string]moves.Check{}, CutBudgetS: int(moves.CutWindowDefault)}
	rep.Checks[moves.CheckProtocol] = w.checkProtocol(ctx, s)
	rep.Checks[moves.CheckSecrets] = w.checkSecrets(ctx, s)
	art := w.checkArchive(ctx, rec, source)
	rep.Checks[moves.CheckArchive] = art.check
	held, t3, notes := w.checkEnv(ctx, s, rec, source, art)
	rep.Checks[moves.CheckEnv] = t3
	rep.Notes = notes
	rep.Checks[moves.CheckStorage] = w.checkStorage(ctx, held, id)
	// The database's logical size: the newest staged LTX header's, or the
	// source's local file when that is larger (review finding 1).
	staged, dbBytes, replicaErr := w.stagedReplica(ctx, rec)
	if source != nil && source.DBBytes > dbBytes {
		dbBytes = source.DBBytes
	}
	rep.Checks[moves.CheckDisk] = w.checkDisk(max(staged, dbBytes), replicaErr, art)
	rep.Checks[moves.CheckRelease] = w.checkRelease(ctx, s, rec, source)
	rep.Checks[moves.CheckOneOwner] = checkOneOwner(fleet, rec.Move.To, moveRegionName(obs.Claim, park.PlacementRegion(rec.Move.To)))
	rep.Checks[moves.CheckClaim] = w.checkClaim(ctx, obs, rec)
	rep.Checks[moves.CheckLane] = w.checkLane(ctx, rec, fleet)
	rep.Checks[moves.CheckQuiet] = checkQuiet(obs, park.PlacementRegion(rec.Move.To), now)
	rep.Checks[moves.CheckArmed] = checkArmed(cfg)

	// T11 from measured rates: this box's own, and the source's cross rate.
	m := w.measure(ctx, rec)
	m.Cross = source.CrossMeasurement()
	dbMiB := float64(dbBytes) / (1 << 20)
	if replicaErr != nil {
		rep.Checks[moves.CheckBudget] = moves.Check{Reason: replicaReason(replicaErr)}
	} else {
		res := moves.BudgetCheck(dbMiB, art.entries, m, cfg.ReadyDelta, moves.CutWindowDefault, now)
		rep.Checks[moves.CheckBudget] = res.Check
		rep.RecordBudget(dbMiB, art.entries, m, res)
	}

	rep.Inputs = held.inputs
	if source != nil {
		rep.Inputs.Release = source.Release
		if t3.OK {
			rep.Inputs.EnvReport = source.Env.Digest
		}
	}
	if art.check.OK {
		rep.Inputs.FilesPlain = art.plain
	}
	body, err := json.Marshal(rep)
	if err != nil {
		return err
	}
	return moves.PutReplace(ctx, s, moves.TargetReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, id)), body)
}

// checkArmed is T13, the cut guard's arming statement (Paul's ruling,
// 2026-10-11): this box can land the move and roll it back, every hook of
// both wired, judged as the report is written. The source refuses to start
// a cut unless the target's latest report states it (cutRun.refuseUnarmed).
// In production both stay unarmed until the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b). Its reason is
// copy record MV-N440.
func checkArmed(cfg moveCfg) moves.Check {
	if landingArmed(cfg) && rollbackTargetArmed(cfg) {
		return moves.Check{OK: true}
	}
	return moves.Check{Reason: t13NotArmed}
}

// checkProtocol is T1 (§11): this box speaks at least the published floor
// and has armed it locally, so its executable selection enforces it. No
// published floor is a pass: the floor is opt-in until the first
// production move (§11).
func (w *movesWorker) checkProtocol(ctx context.Context, s release.Store) moves.Check {
	fl, err := moves.ReadFloor(ctx, s)
	if err != nil {
		log.Printf("moves: T1: %v", err)
		return moves.Check{Reason: t1FloorUnreadable}
	}
	if fl == nil {
		return moves.Check{OK: true}
	}
	if moves.ProtocolVersion < fl.Min {
		return moves.Check{Reason: t1Below(fl.Min)}
	}
	local, err := readLocalFloor(w.cfg.DataDir)
	if err != nil {
		log.Printf("moves: T1: local floor: %v", err)
		return moves.Check{Reason: t1LocalUnreadable}
	}
	if local == nil || local.MinRelease != fl.MinRelease || local.Min < fl.Min {
		return moves.Check{Reason: t1NotArmed}
	}
	return moves.Check{OK: true}
}

// checkSecrets is T2 (§10): this box's secrets key fingerprint equals the
// one the console publishes.
func (w *movesWorker) checkSecrets(ctx context.Context, s release.Store) moves.Check {
	if w.cfg.SecretsPub == nil {
		return moves.Check{Reason: t2NoKey}
	}
	var rec moves.ConsoleRecord
	if ok, err := readReport(ctx, s, moves.ConsoleKey, &rec); err != nil || !ok || rec.SecretsKeyFP == "" {
		if err != nil {
			log.Printf("moves: T2: %v", err)
		}
		return moves.Check{Reason: t2NoConsole}
	}
	if appconfig.KeyFingerprint(w.cfg.SecretsPub) != rec.SecretsKeyFP {
		return moves.Check{Reason: t2Differs}
	}
	return moves.Check{OK: true}
}

// checkArchive is T5: the archive the source report names, read by exact
// version from this box's bucket, opens with this box's key and matches its
// manifest (archive.Verify: digests, every entry, the entry types, the
// limits), and is this move's. Its carried <host>.env is kept for T3. A
// verdict is kept per exact version; a read that fails is not.
func (w *movesWorker) checkArchive(ctx context.Context, rec provision.Record, source *moves.SourceReport) verifiedArchive {
	if source == nil || source.Archive == nil {
		return verifiedArchive{check: moves.Check{Reason: t5NoArchive}}
	}
	ref := *source.Archive
	// A reference by "null" (or no) version pins nothing, so it is refused
	// before the cache: a verdict kept for it would vouch for whatever the
	// key holds now (final review finding 4).
	if !ref.Pinned() {
		return verifiedArchive{ref: ref, check: moves.Check{Reason: t5NotPinned}}
	}
	w.mu.Lock()
	cached, ok := w.archives[rec.Move.ID]
	w.mu.Unlock()
	if ok && cached.ref == ref {
		return cached
	}
	v, keep := w.verifyArchive(ctx, rec, ref)
	if keep {
		w.mu.Lock()
		if w.archives == nil {
			w.archives = map[string]verifiedArchive{}
		}
		w.archives[rec.Move.ID] = v
		w.mu.Unlock()
	}
	return v
}

func (w *movesWorker) verifyArchive(ctx context.Context, rec provision.Record, ref moves.ArchiveRef) (verifiedArchive, bool) {
	cfg := w.cfg
	out := verifiedArchive{ref: ref}
	fail := func(reason string, keep bool, err error) (verifiedArchive, bool) {
		log.Printf("moves: %s: T5: %v", rec.Host, err)
		out.check = moves.Check{Reason: reason}
		return out, keep
	}
	if cfg.SecretsPriv == nil {
		out.check = moves.Check{Reason: t2NoKey}
		return out, false
	}
	if cfg.Versioned == nil {
		return fail(t5Unreadable, false, errors.New("no versioned store"))
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return fail(t5Unreadable, false, err)
	}
	raw, err := vs.GetVersion(ctx, cfg.OwnBucket, ref.ManifestKey, ref.ManifestVersion)
	if err != nil {
		return fail(t5Unreadable, false, err)
	}
	var m archive.Manifest
	if err := json.Unmarshal(raw, &m); err != nil {
		return fail(t5Refused, true, err)
	}
	if m.Op != rec.Move.ID || m.Incarnation != rec.Move.Incarnation || m.Plain != ref.Plain {
		return fail(t5Refused, true, fmt.Errorf("the manifest is for %s/%s/%s", m.Op, m.Incarnation, m.Plain))
	}
	tmp, err := os.MkdirTemp("", "carlos-move-t5-")
	if err != nil {
		return fail(t5Unreadable, false, err)
	}
	defer os.RemoveAll(tmp)
	sealedPath := filepath.Join(tmp, "archive.sealed")
	if err := downloadVersion(ctx, vs, cfg.OwnBucket, ref.SealedObjectKey, ref.SealedVersion, sealedPath); err != nil {
		return fail(t5Unreadable, false, err)
	}
	bind := appconfig.Binding{Account: rec.Account, App: rec.App}
	f, err := os.Open(sealedPath)
	if err != nil {
		return fail(t5Unreadable, false, err)
	}
	defer f.Close()
	if err := archive.Verify(f, m, cfg.SecretsPriv, bind); err != nil {
		return fail(t5Refused, true, err)
	}
	env, err := carriedEnv(f, m, cfg.SecretsPriv, bind, tmp, rec.Host)
	if err != nil {
		return fail(t5Refused, true, err)
	}
	out.check, out.plain, out.entries, out.bytes, out.env = moves.Check{OK: true}, m.Plain, len(m.Entries), m.Bytes, env
	return out, true
}

func downloadVersion(ctx context.Context, vs lease.VersionedStore, bucket, key, version, path string) error {
	rc, err := vs.OpenVersion(ctx, bucket, key, version)
	if err != nil {
		return err
	}
	defer rc.Close()
	f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	if _, err := io.Copy(f, io.LimitReader(rc, archive.MaxDecoded+archive.MaxDecoded/16)); err != nil {
		f.Close()
		return err
	}
	return f.Close()
}

// carriedEnv extracts the verified archive into a scratch directory and
// returns its <host>.env (nil when it carries none).
func carriedEnv(f *os.File, m archive.Manifest, priv *ecdh.PrivateKey, bind appconfig.Binding, tmp, host string) ([]byte, error) {
	if _, err := f.Seek(0, io.SeekStart); err != nil {
		return nil, err
	}
	pr, pw := io.Pipe()
	done := make(chan error, 1)
	go func() {
		err := archive.Open(f, m.SealedDataKey, priv, bind, pw)
		pw.CloseWithError(err)
		done <- err
	}()
	envPath := filepath.Join(tmp, host+".env")
	xerr := archive.Extract(pr, filepath.Join(tmp, "files"), envPath)
	pr.CloseWithError(errors.New("extract stopped reading"))
	if oerr := <-done; oerr != nil && xerr == nil {
		xerr = oerr
	}
	if xerr != nil {
		return nil, xerr
	}
	env, err := os.ReadFile(envPath)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	return env, err
}

// checkEnv is T3: the environment this box would launch the flat held row
// with, compared by HMAC with the source's launch report. Shadowed names
// and advisory differences are notes, never failures.
func (w *movesWorker) checkEnv(ctx context.Context, s release.Store, rec provision.Record, source *moves.SourceReport, art verifiedArchive) (heldEnv, moves.Check, []string) {
	cfg := w.cfg
	if source == nil {
		return heldEnv{}, moves.Check{Reason: t3NoSource}, nil
	}
	if !art.check.OK {
		return heldEnv{}, moves.Check{Reason: t5NoArchive}, nil
	}
	key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
	if err != nil {
		return heldEnv{}, moves.Check{Reason: t2NoKey}, nil
	}
	if cfg.ComposeHeldEnv == nil {
		return heldEnv{}, moves.Check{Reason: t3NoCompose}, nil
	}
	env, instance, shadowed, in, err := cfg.ComposeHeldEnv(ctx, s, rec, art.env)
	if err != nil {
		log.Printf("moves: %s: T3: %v", rec.Host, err)
		return heldEnv{}, moves.Check{Reason: t3NoCompose}, nil
	}
	held := heldEnv{env: env, shadowed: shadowed, inputs: in}
	var notes, failed []string
	for _, name := range shadowed {
		notes = append(notes, t3Shadowed(name))
	}
	// A v1 report cannot say which box variables the instance supplied, so
	// it is refused before any comparison (fail closed).
	if !moves.CurrentEnvReport(source.Env) {
		return held, moves.Check{Reason: t3OldReport}, notes
	}
	// A move into home names the home's code, "", on the record: T3's home
	// rule follows the record, never this box's region, so a second
	// placementless box stays T8's failure and is not T3's pass.
	targetHome := rec.Move.To == ""
	for _, d := range moves.CompareEnv(source.Env, moves.ReportLaunch(key, env, instance), true, targetHome) {
		if d.Note() {
			notes = append(notes, d.Name+" ("+d.Kind+")")
			continue
		}
		failed = append(failed, d.Name+" ("+d.Kind+")")
	}
	if len(failed) > 0 {
		return held, moves.Check{Reason: t3Differs(failed)}, notes
	}
	return held, moves.Check{OK: true}, notes
}

// checkStorage is T4: every store bound in the composed environment answers
// a list, put, get and delete of .carlos-move-probe/<move id> with its own
// credentials.
func (w *movesWorker) checkStorage(ctx context.Context, held heldEnv, id string) moves.Check {
	if held.env == nil {
		return moves.Check{Reason: t3NoCompose}
	}
	if w.cfg.ProbeStore == nil {
		return moves.Check{Reason: t4Failed("(no probe on this box)")}
	}
	if prefix, err := w.cfg.ProbeStore(ctx, held.env, id); err != nil {
		log.Printf("moves: T4: %s: %v", prefix, err)
		return moves.Check{Reason: t4Failed(prefix)}
	}
	return moves.Check{OK: true}
}

// errNoReplicaStaged is a lane with no LTX file in it yet.
var errNoReplicaStaged = errors.New("no replica has been staged")

func replicaReason(err error) string {
	if errors.Is(err, errNoReplicaStaged) {
		return t11NoReplica
	}
	return t11Unmeasured
}

// stagedReplica reads the replica staged in this box's bucket for the move:
// the bytes staged, and the database's logical size, Commit pages of
// PageSize from the header of the newest LTX file (the highest max TXID).
// The staged bytes are no measure of the database: LTX compresses its
// pages, so a compressible database stages far fewer bytes than it holds.
// A newest file whose header cannot be read is an error, never a guess.
func (w *movesWorker) stagedReplica(ctx context.Context, rec provision.Record) (staged, logical int64, err error) {
	m, err := w.mirror(ctx, w.cfg.OwnBucketRegion)
	if err != nil {
		return 0, 0, err
	}
	prefix := moves.ReplicaPrefix(rec.Host, rec.Move.ID)
	objs, err := m.List(ctx, w.cfg.OwnBucket, prefix)
	if err != nil {
		return 0, 0, err
	}
	for _, o := range objs {
		staged += o.Size
	}
	newest := newestLTX(objs, prefix, nil)
	if newest == "" {
		return staged, 0, errNoReplicaStaged
	}
	hdr, err := w.ltxHeader(ctx, newest)
	if err != nil {
		return staged, 0, fmt.Errorf("the newest staged LTX file %s: %w", newest, err)
	}
	return staged, int64(hdr.Commit) * int64(hdr.PageSize), nil
}

// newestLTX is the key in objs of the newest LTX file under prefix, the one
// with the highest max TXID (ties go to the greater key), or "" when there
// is none. keep, when non-nil, must admit a key relative to prefix.
func newestLTX(objs map[string]lease.MirrorObject, prefix string, keep func(string) bool) string {
	var newest string
	var newestTX ltx.TXID
	for k := range objs {
		rel := strings.TrimPrefix(k, prefix)
		if keep != nil && !keep(rel) {
			continue
		}
		_, name, ok := strings.Cut(rel, "/")
		if !ok {
			continue
		}
		_, maxTX, perr := ltx.ParseFilename(name)
		if perr != nil {
			continue
		}
		if newest == "" || maxTX > newestTX || (maxTX == newestTX && k > newest) {
			newest, newestTX = k, maxTX
		}
	}
	return newest
}

// replicaLogicalSize is the source's half of the size evidence (final
// review finding 2): the database's logical size as its own replica states
// it, from the header of the newest LTX file under the claim's generation
// (srcPrefix, filtered by keep: the files the lane mirrors), read as the
// target reads its staged copy. No LTX file yet is 0; a newest file whose
// header cannot be read is an error, which ends the round before its
// report, so nothing promotes on a size nobody measured.
func (w *movesWorker) replicaLogicalSize(ctx context.Context, m lease.MirrorStore, srcPrefix string, keep func(string) bool) (int64, error) {
	objs, err := m.List(ctx, w.cfg.OwnBucket, srcPrefix)
	if err != nil {
		return 0, fmt.Errorf("list the replica source: %w", err)
	}
	newest := newestLTX(objs, srcPrefix, keep)
	if newest == "" {
		return 0, nil
	}
	hdr, err := w.ltxHeader(ctx, newest)
	if err != nil {
		return 0, fmt.Errorf("the replica's newest LTX file %s: %w", newest, err)
	}
	return int64(hdr.Commit) * int64(hdr.PageSize), nil
}

// ltxHeader reads the header of the current version of key in this box's
// bucket: its first ltx.HeaderSize bytes, by exact version.
func (w *movesWorker) ltxHeader(ctx context.Context, key string) (ltx.Header, error) {
	cfg := w.cfg
	if cfg.Versioned == nil {
		return ltx.Header{}, errors.New("no versioned store")
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return ltx.Header{}, err
	}
	vers, err := vs.ListVersions(ctx, cfg.OwnBucket, key)
	if err != nil {
		return ltx.Header{}, err
	}
	var version string
	for _, v := range vers {
		if v.Key == key && v.IsLatest && !v.DeleteMarker {
			version = v.VersionID
		}
	}
	if version == "" {
		return ltx.Header{}, errors.New("no current version")
	}
	var rc io.ReadCloser
	if ro, ok := vs.(lease.VersionRangeOpener); ok {
		rc, err = ro.OpenVersionRange(ctx, cfg.OwnBucket, key, version, 0, ltx.HeaderSize)
	} else {
		rc, err = vs.OpenVersion(ctx, cfg.OwnBucket, key, version)
	}
	if err != nil {
		return ltx.Header{}, err
	}
	defer rc.Close()
	buf := make([]byte, ltx.HeaderSize)
	if _, err := io.ReadFull(rc, buf); err != nil {
		return ltx.Header{}, err
	}
	var hdr ltx.Header
	if err := hdr.UnmarshalBinary(buf); err != nil {
		return ltx.Header{}, err
	}
	if !ltx.IsValidPageSize(hdr.PageSize) || hdr.Commit == 0 {
		return ltx.Header{}, fmt.Errorf("header names %d pages of %d bytes", hdr.Commit, hdr.PageSize)
	}
	return hdr, nil
}

// checkDisk is T6: free space covers twice the replica, plus the archive,
// plus 2 GiB, and the disk stays under 85% full after it.
func (w *movesWorker) checkDisk(replicaBytes int64, replicaErr error, art verifiedArchive) moves.Check {
	if replicaErr != nil {
		log.Printf("moves: T6: %v", replicaErr)
		return moves.Check{Reason: replicaReason(replicaErr)}
	}
	if w.cfg.DiskFree == nil {
		return moves.Check{Reason: t6Unmeasured}
	}
	total, avail, err := w.cfg.DiskFree(w.cfg.DataDir)
	if err != nil || total == 0 {
		log.Printf("moves: T6: %v", err)
		return moves.Check{Reason: t6Unmeasured}
	}
	need := uint64(2*replicaBytes+art.bytes) + 2<<30
	used := total - avail
	if avail < need || float64(used+need)/float64(total) >= 0.85 {
		return moves.Check{Reason: t6NoRoom(float64(need) / (1 << 30))}
	}
	return moves.Check{OK: true}
}

// checkRelease is T7: the source's pinned release materializes for the held
// row on this box. With the claim off that is staged and published
// (heldReleaseReady); under the claim it is staged and verified, and the
// landing's configured step publishes it (heldReleaseStaged; Paul's T7
// ruling, 2026-10-12).
func (w *movesWorker) checkRelease(ctx context.Context, s release.Store, rec provision.Record, source *moves.SourceReport) moves.Check {
	if source == nil || source.Release == "" {
		return moves.Check{Reason: t7NoRelease}
	}
	if w.cfg.ReleaseReady == nil {
		return moves.Check{Reason: t7Failed}
	}
	return w.cfg.ReleaseReady(ctx, s, rec, source.Release)
}

// checkOneOwner is T8: exactly one hosting box publishes the target
// placement.
func checkOneOwner(fleet moveFleet, to string, names ...string) moves.Check {
	name := moves.RegionText(to)
	if len(names) == 1 && names[0] != "" {
		name = names[0]
	}
	if n := len(fleet.boxes[to]); n != 1 {
		return moves.Check{Reason: t8Count(n, name)}
	}
	return moves.Check{OK: true}
}

// checkClaim is T9: the claim is moving under this move, held by the
// record's incarnation, has touched this region and box, and the staging
// lane's move.json is this move's (the target owns that validation: the
// source cannot read the lane, round-2 finding 7).
func (w *movesWorker) checkClaim(ctx context.Context, obs hostclaim.Observed, rec provision.Record) moves.Check {
	c := obs.Claim
	if err := hostclaim.RequireMoveOp(rec.Move.ID, hostclaim.KindMoving, hostclaim.PhaseMoving)(c); err != nil {
		return moves.Check{Reason: t9NotThisMove}
	}
	if c.Holder == nil || c.Holder.Incarnation != rec.Move.Incarnation {
		return moves.Check{Reason: t9Incarnation}
	}
	if !slices.ContainsFunc(c.Touched.Regions, func(region string) bool {
		return park.PlacementMatches(c.Op.Intent["home_region"], rec.Move.To, region)
	}) {
		return moves.Check{Reason: t9Untouched(moveRegionName(c, park.PlacementRegion(rec.Move.To)))}
	}
	if !containsString(c.Touched.Boxes, w.cfg.Box) {
		return moves.Check{Reason: t9Untouched(w.cfg.Box)}
	}
	m, err := w.mirror(ctx, w.cfg.OwnBucketRegion)
	if err != nil {
		log.Printf("moves: %s: T9: %v", rec.Host, err)
		return moves.Check{Reason: t9NoMoveJSON}
	}
	raw, err := m.Get(ctx, w.cfg.OwnBucket, moves.MoveJSONKey(rec.Host, rec.Move.ID))
	if err != nil {
		if !errors.Is(err, lease.ErrMirrorNotFound) {
			log.Printf("moves: %s: T9: %v", rec.Host, err)
		}
		return moves.Check{Reason: t9NoMoveJSON}
	}
	var mj moves.MoveJSON
	if err := json.Unmarshal(raw, &mj); err != nil || mj.MoveID != rec.Move.ID || mj.Incarnation != c.Holder.Incarnation ||
		mj.Host != rec.Host || mj.From != rec.Move.From || mj.To != rec.Move.To {
		return moves.Check{Reason: t9OtherMove}
	}
	return moves.Check{OK: true}
}

// checkLane is T10: this box can put and delete moves/_lane/<id> in the
// source's bucket (M17's return lane).
func (w *movesWorker) checkLane(ctx context.Context, rec provision.Record, fleet moveFleet) moves.Check {
	src, ok := fleet.targets[rec.Move.From]
	if !ok {
		src, ok = fleet.targets[rec.Region]
	}
	if !ok || src.DataBucket == "" {
		return moves.Check{Reason: t10NoBucket}
	}
	if w.cfg.LaneProbe == nil {
		return moves.Check{Reason: t10Failed}
	}
	if err := w.cfg.LaneProbe(ctx, src.DataBucket, src.DataBucketRegion, rec.Move.ID); err != nil {
		log.Printf("moves: %s: T10: %v", rec.Host, err)
		return moves.Check{Reason: t10Failed}
	}
	return moves.Check{OK: true}
}

// checkQuiet is T12: the claim's quiet_after for the target region passed.
func checkQuiet(obs hostclaim.Observed, region string, now time.Time) moves.Check {
	if until := park.QuietUntil(obs.Claim.QuietAfter, obs.Claim.Op.Intent["home_region"], region); !until.IsZero() && now.Before(until) {
		return moves.Check{Reason: t12NotQuiet(moveRegionName(obs.Claim, region), until)}
	}
	return moves.Check{OK: true}
}

// measureEvery is how long this box's own measurements are reused: half of
// moves.MeasurementFreshness, so a report never carries one about to expire,
// and a pass does not move 80 MiB through S3 and the disk every few seconds.
const measureEvery = moves.MeasurementFreshness / 2

// measure is this box's regional S3, disk and per-file sync rates, measured
// again when older than measureEvery. A failed measurement is logged and
// left out, which T11 reports as missing.
func (w *movesWorker) measure(ctx context.Context, rec provision.Record) moves.Measurements {
	cfg := w.cfg
	now := cfg.Now()
	w.mu.Lock()
	m := w.measured
	w.mu.Unlock()
	if cfg.MeasureS3 != nil && (m.S3.At.IsZero() || now.Sub(m.S3.At) > measureEvery) {
		if r, err := cfg.MeasureS3(ctx, moves.ProbeKey(rec.Host, rec.Move.ID)); err != nil {
			log.Printf("moves: measure regional S3: %v", err)
		} else {
			m.S3 = r
		}
	}
	if cfg.MeasureDisk != nil && (m.Disk.At.IsZero() || now.Sub(m.Disk.At) > measureEvery) {
		if d, f, err := cfg.MeasureDisk(ctx); err != nil {
			log.Printf("moves: measure the disk: %v", err)
		} else {
			m.Disk, m.Fsync = d, f
		}
	}
	w.mu.Lock()
	w.measured = m
	w.mu.Unlock()
	return m
}

func moveRegionName(c *hostclaim.Claim, region string) string {
	if park.SameRegion(c.Op.Intent["home_region"], region, park.HomeRegion) {
		if home := c.Op.Intent["home_region"]; home != "" {
			return home
		}
		return moves.HomeRegionText
	}
	return region
}