package host

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/instancekey"
	"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/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The target box's part of an offline move
// (docs/superpowers/plans/2026-10-20-offline-operator-move.md, Task 5): the
// landing, open on the target, and the target's revert. offlineRound calls
// offlineTargetStep on a box whose placement equals rec.Move.To, in every
// offline state, after the source and sealer steps.
//
// The landing is a list of ensures (Fable's B2). Every item is judged from
// what is on disk, in the registry and in the bucket, compared with the
// immutable seal (seal.json, moves.OfflineSealKey), never with the sealer's
// living report: an item that matches is OK and left alone, an item that is
// absent is landed, and an item that is present and different is renamed into
// the move's target-before park directory and landed again. So a pass may run
// at any point, after any item, after the CAS or after an agent restart, and
// it neither parks a correct item nor fails on its own earlier extraction.
//
// Nothing here starts the instance or takes its lease. The row stays held
// "landing" until open; open re-verifies every item against the seal and only
// then releases the hold, and the build arrives the way every new instance's
// does, through ordinary adoption and root's existing stage pass. The first
// request promotes the staged replica (ExecBackend.restorePlain).
//
// An item's reason is what `carlos move offline status` shows beside it, so
// each is a recorded sentence (copy rows MV-O110 on). An error behind a reason
// is logged on the box, never put in the reason.

// offlineTargetStep is the target step of one offline state. Every step is an
// ensure: an error is logged by the pass, throttled per host, and the next
// pass runs the step again.
func (w *movesWorker) offlineTargetStep(ctx context.Context, s release.Store, rec provision.Record) error {
	if w.cfg.Box == "" {
		return errors.New("offline move: this box has no label to report under")
	}
	switch rec.Move.State {
	case provision.OfflineLanding, provision.OfflineLanded:
		return w.offlineLand(ctx, s, rec)
	case provision.OfflineOpen:
		return w.offlineTargetOpen(ctx, s, rec)
	case provision.OfflineReverting:
		return w.offlineTargetRevert(ctx, s, rec)
	}
	return nil
}

// offlineLand is the landing pass in offline-landing and offline-landed:
// ensure every item, rewrite the target report, and, once every item is OK
// in offline-landing, move the record to offline-landed.
func (w *movesWorker) offlineLand(ctx context.Context, s release.Store, rec provision.Record) error {
	m := rec.Move
	if rec.Region != m.To {
		return nil // the flip has not reached this record yet
	}
	r, ok, err := w.landReport(ctx, s, rec)
	if err != nil {
		return err
	}
	if err := writeOfflineReport(ctx, s, r); err != nil {
		return err
	}
	if err := w.landCrash("report"); err != nil {
		return err
	}
	if !ok {
		_, bad := r.ItemsOK(moves.OfflineOwnerLanding)
		w.notice(rec.Host+"#offline-land", "moves: %s: offline move %s: landing, not yet OK: %s", rec.Host, m.ID, strings.Join(bad, ", "))
		return nil
	}
	if m.State != provision.OfflineLanding {
		return nil
	}
	if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.OfflineLandedBy(m.ID, m.To)); err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: offline move %s: not marked landed: %v", rec.Host, m.ID, err)
			return nil
		}
		return err
	}
	log.Printf("moves: %s: offline move %s: landed; every item is OK", rec.Host, m.ID)
	return w.landCrash("landed")
}

// landCrash is the landing's test seam (moveCfg.landFault, the landing's
// crash points): an error stands for a crash at point, after the item of
// that name was ensured ("report" after the report is written, "landed"
// after the CAS, "released" after open's release). Nil in production.
func (w *movesWorker) landCrash(point string) error {
	if w.cfg.landFault == nil {
		return nil
	}
	return w.cfg.landFault("offline-" + point)
}

// offlineTargetOpen is the target step at offline-open. While the row is
// still held by the move, every landing item is ensured and verified against
// the seal once more, and the verified report is written; only when every
// item is OK is the hold released (under reconcileMu and the host's hold
// lock, as the source's revert releases its own). The row then reads ready:
// ordinary adoption stages the channel's build, and the first request
// promotes moves/<host>/. Then the move's archive prefix is deleted from this
// box's bucket and the box reports opened. A pass after the release finds the
// row ready and does not verify again (the staging may already be promoted
// and gone); it goes on only from this box's own report with every item OK.
func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	rt, err := w.offlineRow(host)
	if err != nil {
		return err
	}
	prev := w.prevTargetReport(ctx, s, host, m.ID)
	var items map[string]moves.Check
	switch {
	case rt == nil:
		return fmt.Errorf("offline move %s: %s: no row on the target at open", m.ID, host)
	case rt.HoldOp == m.ID && registry.IsHold(rt.Status):
		r, ok, err := w.landReport(ctx, s, rec)
		if err != nil {
			return err
		}
		// The verified report goes first, so a crash after the release
		// leaves the items this pass verified, not an older pass's.
		if err := writeOfflineReport(ctx, s, r); err != nil {
			return err
		}
		if !ok {
			_, bad := r.ItemsOK(moves.OfflineOwnerLanding)
			return fmt.Errorf("offline move %s: %s: not opening: %s not OK on this box", m.ID, host, strings.Join(bad, ", "))
		}
		items = r.Items
		if cfg.ReleaseHold == nil {
			return errors.New("offline move: no registry to release the hold through")
		}
		var relErr error
		reconcileMu.Lock()
		provision.WithHostHoldLock(host, func() {
			_, relErr = cfg.ReleaseHold(host, m.ID)
		})
		reconcileMu.Unlock()
		if relErr != nil {
			return relErr
		}
		if rt, err = w.offlineRow(host); err != nil {
			return err
		}
		if rt == nil || registry.IsHold(rt.Status) {
			return fmt.Errorf("offline move %s: %s: the hold was not released", m.ID, host)
		}
		log.Printf("moves: %s: offline move %s: opened; the row is ready and adoption stages its build", host, m.ID)
		if err := w.landCrash("released"); err != nil {
			return err
		}
	case rt.HoldOp != "" || registry.IsHold(rt.Status):
		return fmt.Errorf("offline move %s: %s is held %s by %q, not by this move", m.ID, host, rt.Status, rt.HoldOp)
	default:
		// The row is ready: this box released it on an earlier pass, which
		// wrote the verified report first. A ready row this box never
		// verified (reconcile mints a missing row ready once the record is
		// open) is not vouched for.
		if prev == nil {
			return fmt.Errorf("offline move %s: %s is ready on this box, which has no landing report for the move", m.ID, host)
		}
		if ok, bad := prev.ItemsOK(moves.OfflineOwnerLanding); !ok {
			return fmt.Errorf("offline move %s: %s is ready on this box, but its last report has %s not OK", m.ID, host, strings.Join(bad, ", "))
		}
		items = prev.Items
	}
	if prev == nil || prev.Phase != moves.OfflinePhaseOpened {
		st, err := w.landStore(ctx)
		if err != nil {
			return err
		}
		if err := lease.DeletePrefix(ctx, st, cfg.OwnBucket, moves.OfflineArchivePrefix(host, m.ID)); err != nil {
			return fmt.Errorf("offline move %s: delete the archive: %w", m.ID, err)
		}
	}
	return writeOfflineReport(ctx, s, moves.OfflineReport{
		Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseOpened, At: cfg.now().UTC(),
		Items: landingItemsOnly(items), RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")),
	})
}

// offlineTargetRevert is the target step at offline-reverting. When this box
// holds the row for the move, or has reported for the move at all, it first
// renames what the landing may have extracted and rendered (the files
// directory, <host>.env and <host>.config.env) into the move's target park
// directory, and only then deletes the held row (DeleteIfHold, which takes
// its timers with it) under reconcileMu and the hold lock, so a reconcile
// pass that listed the record at landing cannot mint it held again after
// the delete. Parking first means a crash anywhere leaves nothing of the
// move's in the data directory without a row to say why; a first pass that
// extracted and crashed before it could report is parked all the same,
// since its row is still held. Then it reports reverted. The source is
// untouched, so the revert loses nothing. A box that never had the row held
// for the move and never reported does nothing.
func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m, host := rec.Move, rec.Host
	rt, err := w.offlineRow(host)
	if err != nil {
		return err
	}
	prev := w.prevTargetReport(ctx, s, host, m.ID)
	held := rt != nil && rt.HoldOp == m.ID && registry.IsHold(rt.Status)
	if rt != nil && !held {
		// Not this move's row (the move holds its row until open, and an
		// open move is never aborted): leave it and what it serves.
		return fmt.Errorf("offline move %s: %s is %q held by %q on this box, not by the move; the revert leaves it", m.ID, host, rt.Status, rt.HoldOp)
	}
	if !held && prev == nil {
		return nil
	}
	parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleTarget)
	if parkDir == "" {
		return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir)
	}
	for _, p := range []string{filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env")} {
		if err := parkInto(p, parkDir); err != nil {
			return fmt.Errorf("offline move %s: park %s: %w", m.ID, p, err)
		}
	}
	if held {
		if cfg.DeleteIfHold == nil {
			return errors.New("offline move: no registry to delete the held row through")
		}
		var delErr error
		reconcileMu.Lock()
		provision.WithHostHoldLock(host, func() {
			_, delErr = cfg.DeleteIfHold(host, m.ID)
		})
		reconcileMu.Unlock()
		if delErr != nil {
			return delErr
		}
		if rt, err = w.offlineRow(host); err != nil {
			return err
		}
		if rt != nil && rt.HoldOp == m.ID {
			return fmt.Errorf("offline move %s: %s is still held by the move", m.ID, host)
		}
		log.Printf("moves: %s: offline move %s: the target's held row is deleted", host, m.ID)
	}
	return writeOfflineReport(ctx, s, moves.OfflineReport{
		Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(),
		RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")),
	})
}

// parkInto renames p into dir under its own name (parkName numbers it when
// the name is taken). An absent p is nothing to do.
func parkInto(p, dir string) error {
	if !lexists(p) {
		return nil
	}
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return err
	}
	dst := parkName(dir, filepath.Base(p))
	if err := os.Rename(p, dst); err != nil {
		return err
	}
	log.Printf("moves: parked %s at %s", p, dst)
	return nil
}

// landingItemsOnly keeps the items a target report may carry.
func landingItemsOnly(in map[string]moves.Check) map[string]moves.Check {
	if len(in) == 0 {
		return nil
	}
	out := map[string]moves.Check{}
	for id, c := range in {
		if owner, ok := moves.OfflineItemOwner(id); ok && owner == moves.OfflineOwnerLanding {
			out[id] = c
		}
	}
	return out
}

// prevTargetReport is this box's last target report for the move, nil when
// there is none or it cannot be trusted.
func (w *movesWorker) prevTargetReport(ctx context.Context, s release.Store, host, op string) *moves.OfflineReport {
	key := moves.OfflineReportKey(host, op, moves.OfflineRoleTarget, w.cfg.Box)
	if key == "" {
		return nil
	}
	body, _, err := s.Get(ctx, key)
	if err != nil {
		return nil
	}
	var r moves.OfflineReport
	if json.Unmarshal(body, &r) != nil || moves.ValidateOfflineReport(r) != nil || r.Host != host || r.Op != op || r.Role != moves.OfflineRoleTarget {
		return nil
	}
	return &r
}

// landStore is this box's own data bucket (moveCfg.Mirror for its own
// region): the staged replica, the archive, the timers snapshot and the
// source's env report.
func (w *movesWorker) landStore(ctx context.Context) (lease.MirrorStore, error) {
	if w.cfg.Mirror == nil || w.cfg.OwnBucket == "" {
		return nil, errors.New("offline move: no data bucket on this box")
	}
	return w.mirror(ctx, w.cfg.OwnBucketRegion)
}

// landRun is one landing pass's state.
type landRun struct {
	w    *movesWorker
	s    release.Store
	rec  provision.Record
	m    *provision.MoveState
	host string
	seal moves.OfflineReport
	rt   *registry.Route
	prev *moves.OfflineReport

	items map[string]moves.Check
	hard  bool // an item failed in a way the next pass will not clear by itself
	// staged is the staged replica's listing as verified this pass (the
	// report's Staged): the db item restores once per listing.
	staged map[string]moves.StagedObject

	st       lease.MirrorStore
	arch     *offlineArchive
	archErr  error
	archHard bool
}

// offlineArchive is the move's archive in this box's bucket, read once per
// pass and only when an item needs it.
type offlineArchive struct {
	manifest archive.Manifest
	sealed   []byte
}

func (r *landRun) cfg() moveCfg { return r.w.cfg }

func (r *landRun) ok(id string) { r.items[id] = moves.Check{OK: true} }

// landWait records an item not yet OK that a later pass may make OK. reason is
// a recorded sentence.
func (r *landRun) landWait(id, reason string) { r.items[id] = moves.Check{Reason: reason} }

// landWaitErr is landWait for an item an error stopped this pass: the error is the
// box's log line, never the reason.
func (r *landRun) landWaitErr(id, reason string, err error) {
	r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err)
	r.items[id] = moves.Check{Reason: reason}
}

// landFail records a mismatch with the seal: the report's phase becomes failed.
func (r *landRun) landFail(id, reason string) {
	r.items[id] = moves.Check{Reason: reason}
	r.hard = true
}

// landFailErr is landFail with the error behind it logged on the box.
func (r *landRun) landFailErr(id, reason string, err error) {
	r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err)
	r.landFail(id, reason)
}

func (r *landRun) isOK(id string) bool { return r.items[id].OK }

func (r *landRun) filesDir() string   { return filepath.Join(r.cfg().DataDir, r.host) }
func (r *landRun) envPath() string    { return filepath.Join(r.cfg().DataDir, r.host+".env") }
func (r *landRun) configPath() string { return filepath.Join(r.cfg().DataDir, r.host+".config.env") }
func (r *landRun) beforeDir() string {
	return moves.OfflineParkDir(r.cfg().DataDir, r.host, r.m.ID, moves.OfflineParkTargetBefore)
}

// landReport ensures every landing item and returns the target report with
// its phase (landed, or failed on a mismatch), and whether every item is OK.
// Items wait, never fail, for what a later pass can mend (an S3 error, a row
// not minted yet, a render that could not run); they fail on a mismatch with
// the seal.
func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provision.Record) (moves.OfflineReport, bool, error) {
	cfg := w.cfg
	r := &landRun{w: w, s: s, rec: rec, m: rec.Move, host: rec.Host, items: map[string]moves.Check{}}
	r.prev = w.prevTargetReport(ctx, s, r.host, r.m.ID)
	rt, err := w.offlineRow(r.host)
	if err != nil {
		return moves.OfflineReport{}, false, err
	}
	r.rt = rt
	r.row()
	if r.isOK("row") && r.readSeal(ctx) {
		// In this order: files before config and the env check (the key is
		// read from both files), config before the env pin.
		for _, step := range []struct {
			id  string
			run func()
		}{
			{"warm-cache", r.warmCache},
			{"key-adoption", r.keyAdoption},
			{"lease", func() { r.lease(ctx) }},
			{"db", func() { r.db(ctx) }},
			{"files", func() { r.files(ctx) }},
			{"config", func() { r.config(ctx) }},
			{"env", r.env},
			{"env-pin", func() { r.envPin(ctx) }},
			{"timers", func() { r.timers(ctx) }},
			{"next-due", r.nextDue},
			{"replication", r.replication},
			{"release", func() { r.release(ctx) }},
		} {
			step.run()
			if err := w.landCrash(step.id); err != nil {
				return moves.OfflineReport{}, false, err
			}
		}
	}
	for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
		if _, done := r.items[it.ID]; done {
			continue
		}
		if r.isOK("row") {
			r.landWait(it.ID, "not checked: the seal could not be used")
		} else {
			r.landWait(it.ID, "not checked: the row is not held for this move")
		}
	}
	phase := moves.OfflinePhaseLanded
	if r.hard {
		phase = moves.OfflinePhaseFailed
	}
	out := moves.OfflineReport{
		Op: r.m.ID, Host: r.host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: phase, At: cfg.now().UTC(), Items: r.items,
		Held: r.rt != nil && r.rt.HoldOp == r.m.ID && registry.IsHold(r.rt.Status), RowPresent: r.rt != nil,
		LocalDB: r.anyWarm(), Files: isDir(r.filesDir()), Env: lexists(r.envPath()), Staged: r.staged,
	}
	if r.rt != nil {
		out.NextDue, out.Replication = r.rt.NextDue, r.rt.Replication
	}
	ok, _ := out.ItemsOK(moves.OfflineOwnerLanding)
	return out, ok, nil
}

// row: the reconciled row exists, held "landing" by this move, at flat paths.
func (r *landRun) row() {
	rt := r.rt
	switch {
	case rt == nil:
		r.landWait("row", "no row yet: reconcile mints it held")
	case rt.HoldOp != "" && rt.HoldOp != r.m.ID:
		r.landFail("row", "the row is held by another operation")
	case rt.HoldOp != r.m.ID || rt.Status != "landing":
		r.landWait("row", "the row is not held landing by this move yet")
	case provision.NestedPaths(r.cfg().DataDir, *rt):
		r.landFail("row", "the row's paths are nested, and an offline move lands flat rows only")
	default:
		r.ok("row")
	}
}

// readSeal reads seal.json and checks it against the record's evidence:
// the key and the sha256 of the exact bytes the sealer wrote, once.
func (r *landRun) readSeal(ctx context.Context) bool {
	src := r.m.Source
	if src == nil || src.Evidence == "" {
		r.landFailAll("the record carries no seal evidence")
		return false
	}
	key := moves.OfflineSealKey(r.host, r.m.ID)
	raw, _, err := r.s.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		r.landFailAll("the seal is missing")
		return false
	}
	if err != nil {
		r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: read %s: %v", r.host, r.m.ID, key, err)
		r.landWaitAll("the seal could not be read")
		return false
	}
	if moves.OfflineSealEvidence(r.host, r.m.ID, raw) != src.Evidence {
		r.landFailAll("the seal is not the one the record names")
		return false
	}
	var seal moves.OfflineReport
	if err := json.Unmarshal(raw, &seal); err != nil {
		r.landFailAll("the seal does not decode")
		return false
	}
	if err := moves.ValidateOfflineSeal(seal, r.host, r.m.ID); err != nil {
		r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: the seal: %v", r.host, r.m.ID, err)
		r.landFailAll("the seal is not a valid seal of this move")
		return false
	}
	r.seal = seal
	return true
}

// landFailAll fails every item but row for a seal that cannot be used.
func (r *landRun) landFailAll(reason string) {
	for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
		if it.ID != "row" {
			r.landFail(it.ID, reason)
		}
	}
}

func (r *landRun) landWaitAll(reason string) {
	for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) {
		if it.ID != "row" {
			r.landWait(it.ID, reason)
		}
	}
}

// warmPaths are the local database artifacts a box could trust as a warm
// cache for the host: the row's database and the default flat one, each
// with its WAL, SHM, both litestream meta directories and its seed marker.
func (r *landRun) warmPaths() []string {
	dbs := []string{provision.DefaultDBPath(r.cfg().DataDir, r.host)}
	if r.rt != nil && r.rt.DBPath != "" && r.rt.DBPath != dbs[0] {
		dbs = append(dbs, r.rt.DBPath)
	}
	var out []string
	for _, db := range dbs {
		out = append(out, db, db+"-wal", db+"-shm", db+"-litestream", activatorMetaDir(db), db+activator.SeedMarkerSuffix)
	}
	return out
}

func (r *landRun) anyWarm() bool {
	for _, p := range r.warmPaths() {
		if lexists(p) {
			return true
		}
	}
	return false
}

// ensureAbsent renames every present path into target-before and reports
// the item from what is left.
func (r *landRun) ensureAbsent(id string, paths []string) {
	dir := r.beforeDir()
	for _, p := range paths {
		if !lexists(p) {
			continue
		}
		if err := parkInto(p, dir); err != nil {
			r.landWaitErr(id, "a path the target held for the host could not be parked", err)
			return
		}
	}
	for _, p := range paths {
		if lexists(p) {
			r.landWait(id, "a path the target held for the host is still there")
			return
		}
	}
	r.ok(id)
}

// warmCache: no local database or meta directory for the host. The target
// never woke it, so anything here predates the move and would win over the
// staged replica on the first wake.
func (r *landRun) warmCache() { r.ensureAbsent("warm-cache", r.warmPaths()) }

// keyAdoption: no instance-key adoption record (written only for claim
// launches; never carried).
func (r *landRun) keyAdoption() {
	r.ensureAbsent("key-adoption", []string{filepath.Join(r.cfg().DataDir, r.host+instancekey.AdoptionSuffix)})
}

// lease: the target bucket's lock for the host is absent, a lifted
// tombstone, or an ordinary lease that has expired. A tombstone of another
// move is lifted. Nothing here ever takes the lease. Expiry is read on the
// wall clock, as the lease package reads it.
func (r *landRun) lease(ctx context.Context) {
	cfg := r.cfg()
	loc := activator.HibernatePrefix(cfg.OwnPrefix, r.host)
	_, owner, expires, err := lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion)
	if err != nil {
		r.landWaitErr("lease", "the target's lock could not be read", err)
		return
	}
	switch {
	case owner == "" && expires.IsZero():
		r.ok("lease")
		return
	case lease.IsLiftedTombstone(owner, expires):
		r.ok("lease")
		return
	case owner == moves.TombstoneOwner(r.m.ID):
		r.landFail("lease", "the target's lock is this move's own tombstone")
		return
	case !strings.HasPrefix(owner, lease.TombstonePrefix):
		if time.Now().After(expires) {
			r.ok("lease")
			return
		}
		r.w.notice(r.host+"#offline-land-lease", "moves: %s: offline move %s: the target's lock is held by %s until %s", r.host, r.m.ID, owner, expires.UTC().Format(time.RFC3339))
		r.landWait("lease", "a writer holds the target's lock")
		return
	}
	stale := owner
	if _, err := lease.LiftTombstone(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion, func(o string) bool { return o == stale }); err != nil {
		r.landWaitErr("lease", "a stale tombstone in the target's bucket could not be lifted", err)
		return
	}
	_, owner, expires, err = lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion)
	if err != nil {
		r.landWaitErr("lease", "the target's lock could not be read", err)
		return
	}
	if (owner == "" && expires.IsZero()) || lease.IsLiftedTombstone(owner, expires) {
		log.Printf("moves: %s: offline move %s: lifted the stale tombstone %s in the target's bucket", r.host, r.m.ID, stale)
		r.ok("lease")
		return
	}
	r.landWait("lease", "the target's lock is still held after the lift")
}

func (r *landRun) store(ctx context.Context) (lease.MirrorStore, error) {
	if r.st != nil {
		return r.st, nil
	}
	st, err := r.w.landStore(ctx)
	if err != nil {
		return nil, err
	}
	r.st = st
	return st, nil
}

// stagedDiff describes how got differs from want, "" when equal: a log line.
func stagedDiff(want, got map[string]moves.StagedObject) string {
	var missing, extra, changed []string
	for k, w := range want {
		g, ok := got[k]
		switch {
		case !ok:
			missing = append(missing, k)
		case g.Size != w.Size || g.ETag != w.ETag:
			changed = append(changed, k)
		}
	}
	for k := range got {
		if _, ok := want[k]; !ok {
			extra = append(extra, k)
		}
	}
	sort.Strings(missing)
	sort.Strings(extra)
	sort.Strings(changed)
	if len(missing)+len(extra)+len(changed) == 0 {
		return ""
	}
	return fmt.Sprintf("missing %v, extra %v, changed %v", missing, extra, changed)
}

// db: the staged replica under moves/<host>/ is exactly what the seal
// staged (nothing consumes it before open), its newest TXID
// (replicaPosition, the sealer's own rule) is the seal's FinalTXID, and it
// restores clean (lease.VerifyReplica into a scratch directory under the
// move's verify park directory, removed after). The restore runs once per
// listing: the verified listing is the report's Staged, and a pass whose
// listing equals the last report's, with db OK there, does not restore
// again. The report is this move's, so the seal it was verified against is
// the same immutable one.
func (r *landRun) db(ctx context.Context) {
	cfg := r.cfg()
	st, err := r.store(ctx)
	if err != nil {
		r.landWaitErr("db", "the staged replica could not be listed", err)
		return
	}
	prefix := activator.MovesPrefix(r.host)
	objs, err := st.List(ctx, cfg.OwnBucket, prefix)
	if err != nil {
		r.landWaitErr("db", "the staged replica could not be listed", err)
		return
	}
	got := make(map[string]moves.StagedObject, len(objs))
	for k, o := range objs {
		got[strings.TrimPrefix(k, prefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag}
	}
	if len(r.seal.Staged) == 0 {
		r.landFail("db", "the seal staged nothing")
		return
	}
	if diff := stagedDiff(r.seal.Staged, got); diff != "" {
		r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica differs from the seal's: %s", r.host, r.m.ID, diff)
		r.landFail("db", "the staged replica is not the one the seal staged")
		return
	}
	if final := replicaPosition(objs, prefix); final != r.seal.FinalTXID {
		r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica ends at TXID %d, the seal at %d", r.host, r.m.ID, final, r.seal.FinalTXID)
		r.landFail("db", "the staged replica does not end at the sealed position")
		return
	}
	if p := r.prev; p != nil && p.Items["db"].OK && reflect.DeepEqual(p.Staged, got) {
		r.staged = got
		r.ok("db")
		return
	}
	dir := moves.OfflineParkDir(cfg.DataDir, r.host, r.m.ID, moves.OfflineParkVerify)
	if dir == "" {
		r.landFail("db", "the staged replica does not restore clean")
		return
	}
	if err := os.MkdirAll(dir, 0o700); err != nil {
		r.landWaitErr("db", "the staged replica could not be verified this pass", err)
		return
	}
	defer os.RemoveAll(dir)
	if _, err := lease.VerifyReplica(ctx, cfg.OwnBucket, strings.TrimSuffix(prefix, "/"), cfg.OwnBucketRegion, dir); err != nil {
		r.landFailErr("db", "the staged replica does not restore clean", err)
		return
	}
	r.staged = got
	r.ok("db")
}

// readArchive fetches the move's archive and its manifest from this box's
// bucket (the sealer's offlineArchiveSealed and offlineArchiveManifest
// under the move's archive prefix), once per pass and only when an item
// needs it. hard is a manifest that is not the seal's.
func (r *landRun) readArchive(ctx context.Context) (*offlineArchive, bool, error) {
	if r.arch != nil || r.archErr != nil {
		return r.arch, r.archHard, r.archErr
	}
	r.arch, r.archHard, r.archErr = r.fetchArchive(ctx)
	return r.arch, r.archHard, r.archErr
}

func (r *landRun) fetchArchive(ctx context.Context) (*offlineArchive, bool, error) {
	body, err := r.object(ctx, offlineArchiveManifest)
	if err != nil {
		return nil, false, err
	}
	out := &offlineArchive{}
	if err := json.Unmarshal(body, &out.manifest); err != nil {
		return nil, true, fmt.Errorf("the archive manifest: %w", err)
	}
	am := out.manifest
	if am.Op != r.m.ID || am.Plain != r.seal.FilesPlain || (r.seal.FilesSealed != "" && am.Sealed != r.seal.FilesSealed) {
		return nil, true, fmt.Errorf("the archive manifest names op %s, plain %s, sealed %s", am.Op, am.Plain, am.Sealed)
	}
	if out.sealed, err = r.object(ctx, offlineArchiveSealed); err != nil {
		return nil, false, err
	}
	return out, false, nil
}

// object reads one of the sealer's objects under the move's archive prefix
// in this box's bucket.
func (r *landRun) object(ctx context.Context, name string) ([]byte, error) {
	st, err := r.store(ctx)
	if err != nil {
		return nil, err
	}
	return st.Get(ctx, r.cfg().OwnBucket, moves.OfflineArchivePrefix(r.host, r.m.ID)+name)
}

// filesMatch reports whether the flat files directory and <host>.env are
// the seal's, by the archive's own plaintext digest over the pair. A seal
// of a host with no files directory carries no files/ entry at all and
// requires no file; extraction then leaves an empty directory, whose digest
// with the env file is the seal's.
func (r *landRun) filesMatch() bool {
	if !isDir(r.filesDir()) {
		return false
	}
	plain, _, _, err := archive.PlainDigest(r.filesDir(), r.envPath())
	return err == nil && plain == r.seal.FilesPlain
}

// files: <DataDir>/<host>/ and <host>.env carry the seal's plaintext digest.
// OK: left alone. Absent: fetched, verified and extracted. Present and
// different: the files directory (when not empty) and an env file that is
// not the seal's are renamed into target-before first, then extracted.
func (r *landRun) files(ctx context.Context) {
	if r.filesMatch() {
		r.ok("files")
		return
	}
	cfg := r.cfg()
	if cfg.ConfigPriv == nil {
		r.landWait("files", "this box holds no CARLOS_SECRETS_KEY to open the archive with")
		return
	}
	a, hard, err := r.readArchive(ctx)
	switch {
	case err != nil && hard:
		r.landFailErr("files", "the archive is not the one the seal names", err)
		return
	case err != nil:
		r.landWaitErr("files", "the archive could not be read", err)
		return
	}
	bind := appconfig.Binding{Account: r.rec.Account, App: r.rec.App}
	if err := archive.Verify(bytes.NewReader(a.sealed), a.manifest, cfg.ConfigPriv, bind); err != nil {
		r.landFailErr("files", "the archive does not verify", err)
		return
	}
	if nonEmptyOrNotDir(r.filesDir()) {
		if err := parkInto(r.filesDir(), r.beforeDir()); err != nil {
			r.landWaitErr("files", "the files directory the target held could not be parked", err)
			return
		}
	}
	if lexists(r.envPath()) && !sameSHA(r.seal.EnvSHA256, hexFileSHA256(r.envPath())) {
		if err := parkInto(r.envPath(), r.beforeDir()); err != nil {
			r.landWaitErr("files", "the env file the target held could not be parked", err)
			return
		}
	}
	if err := extractSealed(a, cfg, bind, r.filesDir(), r.envPath()); err != nil {
		r.landWaitErr("files", "the archive could not be extracted", err)
		return
	}
	log.Printf("moves: %s: offline move %s: extracted the files and env", r.host, r.m.ID)
	if !r.filesMatch() {
		r.landFail("files", "after extraction the files are not the seal's")
		return
	}
	r.ok("files")
}

// extractSealed opens the sealed archive and extracts it, as the Phase 1
// install does (openAndExtract): Extract admits nothing it did not read
// whole.
func extractSealed(a *offlineArchive, cfg moveCfg, bind appconfig.Binding, filesDir, envPath string) error {
	pr, pw := io.Pipe()
	done := make(chan error, 1)
	go func() {
		err := archive.Open(bytes.NewReader(a.sealed), a.manifest.SealedDataKey, cfg.ConfigPriv, bind, pw)
		pw.CloseWithError(err)
		done <- err
	}()
	xerr := archive.Extract(pr, filesDir, envPath)
	pr.CloseWithError(errors.New("extract stopped reading"))
	if oerr := <-done; oerr != nil && xerr == nil {
		xerr = oerr
	}
	return xerr
}

// nonEmptyOrNotDir reports a path Extract would refuse as its files
// directory: present and not an empty directory.
func nonEmptyOrNotDir(p string) bool {
	st, err := os.Lstat(p)
	if err != nil {
		return false
	}
	if !st.IsDir() {
		return true
	}
	ents, err := os.ReadDir(p)
	return err != nil || len(ents) > 0
}

// hexFileSHA256 is the hex sha256 of a file's bytes, "" when it cannot be
// read.
func hexFileSHA256(p string) string {
	b, err := os.ReadFile(p)
	if err != nil {
		return ""
	}
	sum := sha256.Sum256(b)
	return hex.EncodeToString(sum[:])
}

// sameSHA reports whether a sealed sha256 names gotHex, the bare hex form a
// box computes. The seal may carry it bare or as "sha256:<hex>"; "" names
// "" only (nothing there).
func sameSHA(want, gotHex string) bool {
	if gotHex == "" {
		return want == ""
	}
	return want == gotHex || want == "sha256:"+gotHex
}

// config: the held row's <host>.config.env rendered from the bucket bundle
// (RenderHeld, which admits the row held by this move and writes a
// move-bound ack). OK when the file holds the rendered bytes and this box's
// ack names this move and those bytes. Rendered every pass, so a bundle
// changed since the seal reaches the file and fails the env pin.
func (r *landRun) config(ctx context.Context) {
	cfg := r.cfg()
	if cfg.RenderHeld == nil {
		r.landWait("config", "this box has no held render")
		return
	}
	hr, err := cfg.RenderHeld(ctx, r.s, *r.rt, r.m.ID, nil, func(string) error { return nil })
	if err != nil {
		r.landWaitErr("config", "the config could not be rendered", err)
		return
	}
	if hr.Path != r.configPath() {
		r.landFail("config", "the config was rendered at another path")
		return
	}
	if hexFileSHA256(hr.Path) != hr.SHA256 {
		r.landWait("config", "the config file does not hold the rendered bytes")
		return
	}
	body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, r.host))
	if err != nil {
		r.landWaitErr("config", "the config ack could not be read", err)
		return
	}
	var ack appconfig.ConfigAck
	if err := json.Unmarshal(body, &ack); err != nil || ack.MoveOp != r.m.ID || ack.FileSHA256 != hr.SHA256 {
		r.landWait("config", "the config ack does not name this move and the rendered file")
		return
	}
	r.ok("config")
}

// env: <host>.env is the source's byte for byte (the seal's sha256), and
// the instance key the app would be handed, from the env file and the
// rendered config as the activator merges them (instancekey.Effective), is
// the source's (the seal's key sha256). Both travel in the archive, so the
// files item lands them; this is the check.
func (r *landRun) env() {
	if !r.isOK("files") {
		r.landWait("env", "waits for the files item")
		return
	}
	own, err := os.ReadFile(r.envPath())
	switch {
	case errors.Is(err, os.ErrNotExist) && r.seal.EnvSHA256 == "":
	case errors.Is(err, os.ErrNotExist):
		r.landFail("env", "the env file is missing")
		return
	case err != nil:
		r.landWaitErr("env", "the env file could not be read", err)
		return
	default:
		if sum := sha256.Sum256(own); !sameSHA(r.seal.EnvSHA256, hex.EncodeToString(sum[:])) {
			r.landFail("env", "the env file is not the source's")
			return
		}
	}
	if !r.isOK("config") {
		r.landWait("env", "waits for the config item: the instance key is read from both files")
		return
	}
	cfgEnv, err := os.ReadFile(r.configPath())
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		r.landWaitErr("env", "the config file could not be read", err)
		return
	}
	key, _ := instancekey.Effective(instancekey.ParseEnv(cfgEnv), instancekey.ParseEnv(own))
	got := ""
	if key != "" {
		got = instancekey.Digest(key)
	}
	if !sameSHA(r.seal.KeySHA256, got) {
		r.landFail("env", "the instance key is not the source's")
		return
	}
	r.ok("env")
}

// pinNames is the variable names a failed env pin reports. It is not a
// string, so a reason that prints it carries names, never values.
type pinNames []string

func (n pinNames) String() string { return strings.Join(n, ", ") }

// envPin: the composed launch environment of the held row
// (ExecBackend.LaunchLayers) has the source's portable digest. App settings
// must be equal; box settings are left out (moves.Pinned). On a mismatch the
// reason names the variables that differ (moves.CompareEnv against the
// source's report the sealer wrote beside the archive), never their values.
func (r *landRun) envPin(ctx context.Context) {
	cfg := r.cfg()
	if !r.isOK("files") || !r.isOK("config") {
		r.landWait("env-pin", "waits for the files and config items")
		return
	}
	if r.m.Source.EnvDigest != "" && r.m.Source.EnvDigest != r.seal.EnvDigest {
		r.landFail("env-pin", "the record and the seal disagree on this item")
		return
	}
	if cfg.LaunchLayers == nil {
		r.landWait("env-pin", "this box has no exec backend to compose the launch environment")
		return
	}
	key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
	if err != nil {
		r.landWaitErr("env-pin", "this box holds no CARLOS_SECRETS_KEY to key the env report", err)
		return
	}
	env, inst, _, err := cfg.LaunchLayers(*r.rt)
	if err != nil {
		r.landWaitErr("env-pin", "the launch environment could not be composed", err)
		return
	}
	target := moves.ReportLaunch(key, env, inst)
	if target.Digest == r.seal.EnvDigest {
		r.ok("env-pin")
		return
	}
	names := r.envDiff(ctx, target)
	if len(names) == 0 {
		r.landFail("env-pin", "the launch environment differs from the source's")
		return
	}
	r.landFail("env-pin", fmt.Sprintf("the launch environment differs from the source's in %v", names))
}

// envDiff is the names CompareEnv finds between the source's env report
// (the sealer's offlineEnvReportObject, used only when its digest is the
// seal's EnvDigest) and the target's, notes left out. None when the report
// cannot be read or trusted.
func (r *landRun) envDiff(ctx context.Context, target moves.EnvReport) pinNames {
	body, err := r.object(ctx, offlineEnvReportObject)
	if err != nil {
		r.w.notice(r.host+"#offline-land-envreport", "moves: %s: offline move %s: the source's env report: %v", r.host, r.m.ID, err)
		return nil
	}
	var src moves.EnvReport
	if json.Unmarshal(body, &src) != nil || !moves.CurrentEnvReport(src) || src.Digest != r.seal.EnvDigest {
		return nil
	}
	var out pinNames
	for _, d := range moves.CompareEnv(src, target, true, r.m.To == "") {
		if !d.Note() {
			out = append(out, d.Name+" ("+d.Kind+")")
		}
	}
	return out
}

// timers: the row's timer rows digest to the seal's. Otherwise they are
// replaced from the snapshot beside the archive (its digest checked first)
// and read back.
func (r *landRun) timers(ctx context.Context) {
	reg := r.cfg().Registry
	if reg == nil {
		r.landWait("timers", "no registry on this box")
		return
	}
	if r.m.Source.TimersSHA256 != "" && r.m.Source.TimersSHA256 != r.seal.TimersDigest {
		r.landFail("timers", "the record and the seal disagree on this item")
		return
	}
	rows, err := reg.TimersFor(r.host)
	if err != nil {
		r.landWaitErr("timers", "the timers could not be read", err)
		return
	}
	if moves.TimersDigest(rows) == r.seal.TimersDigest {
		r.ok("timers")
		return
	}
	body, err := r.object(ctx, offlineTimersObject)
	if err != nil {
		r.landWaitErr("timers", "the timers snapshot could not be read", err)
		return
	}
	snap, err := moves.DecodeTimers(body)
	if err != nil || snap.Host != r.host || moves.TimersDigest(snap.Rows) != r.seal.TimersDigest {
		r.landFail("timers", "the timers snapshot is not the one the seal names")
		return
	}
	if err := reg.ReplaceTimers(r.host, snap.Rows); err != nil {
		r.landWaitErr("timers", "the timers could not be replaced", err)
		return
	}
	if rows, err = reg.TimersFor(r.host); err != nil {
		r.landWaitErr("timers", "the timers could not be read", err)
		return
	}
	if moves.TimersDigest(rows) != r.seal.TimersDigest {
		r.landFail("timers", "the timers read back differently from the seal's")
		return
	}
	r.ok("timers")
}

// reread refreshes the row after a column write.
func (r *landRun) reread(id string) bool {
	rt, err := r.w.offlineRow(r.host)
	if err == nil && rt == nil {
		err = errors.New("the row is gone")
	}
	if err != nil {
		r.landWaitErr(id, "the row could not be read back", err)
		return false
	}
	r.rt = rt
	return true
}

// nextDue: the row's NextDue is the seal's; else set and read back. A write
// error waits; only a read-back that differs fails.
func (r *landRun) nextDue() {
	if r.rt.NextDue == r.seal.NextDue {
		r.ok("next-due")
		return
	}
	reg := r.cfg().Registry
	if reg == nil {
		r.landWait("next-due", "no registry on this box")
		return
	}
	if err := reg.SetNextDue(r.host, r.seal.NextDue); err != nil {
		r.landWaitErr("next-due", "next due could not be set", err)
		return
	}
	if !r.reread("next-due") {
		return
	}
	if r.rt.NextDue != r.seal.NextDue {
		r.landFail("next-due", "next due reads back differently from the seal's")
		return
	}
	r.ok("next-due")
}

// replication: the row's replication rung is the seal's; else set and read
// back. A write error waits; only a read-back that differs fails.
func (r *landRun) replication() {
	if r.rt.Replication == r.seal.Replication {
		r.ok("replication")
		return
	}
	reg := r.cfg().Registry
	if reg == nil {
		r.landWait("replication", "no registry on this box")
		return
	}
	if err := reg.SetReplication(r.host, r.seal.Replication); err != nil {
		r.landWaitErr("replication", "the replication rung could not be set", err)
		return
	}
	if !r.reread("replication") {
		return
	}
	if r.rt.Replication != r.seal.Replication {
		r.landFail("replication", "the replication rung reads back differently from the seal's")
		return
	}
	r.ok("replication")
}

// release: the app's channel resolves to the build the seal recorded, and
// the target holds no live link to another build of the host
// (liveNotSealed). A promote during the move fails it; the build itself
// arrives after open through ordinary adoption.
func (r *landRun) release(ctx context.Context) {
	cfg := r.cfg()
	if r.m.Source.Release != "" && r.m.Source.Release != r.seal.Release {
		r.landFail("release", "the record and the seal disagree on this item")
		return
	}
	if cfg.ChannelVersion == nil {
		r.landWait("release", "this box cannot resolve channels")
		return
	}
	v, err := cfg.ChannelVersion(ctx, r.s, *r.rt)
	if err != nil {
		r.landWaitErr("release", "the channel could not be resolved", err)
		return
	}
	if v != r.seal.Release {
		r.w.notice(r.host+"#offline-land-release", "moves: %s: offline move %s: the channel points at %q, the seal recorded %q", r.host, r.m.ID, v, r.seal.Release)
		r.landFail("release", "the channel points at another build than the sealed one")
		return
	}
	if err := r.liveNotSealed(); err != nil {
		r.landFailErr("release", "the target box still holds an older build of this instance. see the offline move runbook", err)
		return
	}
	r.ok("release")
}

// liveNotSealed is the release item's second half: nil when the target has
// no live link for the host, or when the link names the sealed build. The
// link (<opt>/live/<host>) is the one file ExecBackend.ensureLiveLink and
// the activator exec. A box the instance once left keeps it, root-owned and
// still naming the build it ran then (retirePaths leaves it); a wake before
// ordinary adoption execs whatever it names, since ensureLiveLink accepts
// any executable and default update settings gate no version. So on a
// return move after an upgrade the old build could open the sealed
// database. Any other link fails the item, and open, which verifies every
// item again, never releases the hold over it. No link is fine: nothing can
// run until adoption stages the sealed build. The link is only read here,
// never changed. The error is the box's log line.
//
// The check reads the link itself and nothing beyond it (the census ruling
// on this unit): Lstat, then Readlink, then a string comparison of the
// target with the sealed build's path, exactly as PointLive writes it. It
// never follows the link to stat or open what it names, so a target that
// reaches the sealed build only through another link or a relative
// spelling fails, and so does a live path that is not a symlink.
func (r *landRun) liveNotSealed() error {
	optDir := r.cfg().OptDir
	if optDir == "" {
		optDir = release.OptDir()
	}
	link := release.LiveLink(optDir, r.host)
	fi, err := os.Lstat(link)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	} else if err != nil {
		return err
	}
	if fi.Mode()&os.ModeSymlink == 0 {
		return fmt.Errorf("%s is not a symlink (%s)", link, fi.Mode().Type())
	}
	target, err := os.Readlink(link)
	if err != nil {
		return err
	}
	want := release.VersionPath(optDir, release.RouteAccount(*r.rt), r.seal.Release, r.rt.App)
	if target != want {
		return fmt.Errorf("%s names %s, the seal recorded %q at %s", link, target, r.seal.Release, want)
	}
	return nil
}