package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// The moves owners for a recovery probe (moves Phase 1 Plan 4 Task 9;
// rulings Q7 and Q6 of the moves-after-Plan-12 preflight, J2, J14, H14). The
// moves owner set (movesProbeOwners) dispatches by the observed op's kind:
// a landing to the landing probe's owners, restoring and reverting here.
//
// One shape is served: the source's recovery probe, the claim restoring (a
// rollback's return) or reverting (rev_phase probing) at op.phase probing,
// this box executing as the move's source in the move's source region, not
// the target region and not lost; the record this move and this holder
// incarnation, rolling_back for the claim's rollback at source phase probing
// or reverting and not reopened; the row held by the rollback id (restoring)
// or the move id (reverting). Everything else is refused before any lease,
// stop or launch.
//
// Every recovery admission is cold and keeps no stamp (J14: S is retired by
// the first recovery admission). The first one is told by the holder still
// at the source's stamp S (op.journal.source_stamp, which ReturnToSource
// and Revert leave in holder.stamp); any later one is a restart.
//
//   - Select (read-only). A dirty return's first admission is into the
//     return generation (the rollback id) the source's install built, after
//     PreparedCheck, sealed when the return seal was built from the stopped
//     database, its baseline the return generation's pin at the return
//     seal's maximum (genselect.SelectPrepared). A clean return's first is
//     into a fresh generation restored from src_gen (the holder's generation
//     ReturnToSource named), coverage clean-return; a declared or fenced
//     return asks no coverage (Q6). A revert's first is into a fresh
//     generation restored from the holder's own generation, which a revert
//     never changed, with no coverage. A restart is a fresh generation with
//     no coverage, restored from the certified remote (genselect.Select), or,
//     for a dirty return while the location's marker is not yet the return's,
//     the return generation's pin (H14). The launch inputs are exactly the
//     recorded ones (Rollback.RecoveryInputs or Move.RecoveryInputs; the
//     start policy carries them), the live release and the launch
//     environment's report are theirs, and CARLOS_MOVE_PROBE=1 is in the
//     child's environment, as for the landing's probe.
//   - AdmissionEvidence: the same decisions re-derived and compared against
//     a baseline selected again. While a declared loss or a target fence is
//     pending, the writer replaces RemoteMaxTXID with the frozen
//     predecessor's maximum from its LossSources call (writer_loss.go), and
//     AdditionalWritesLost with the other frozen sources: the loss contract
//     (movesloss.go, Task 4) accepts exactly that substitution.
//   - Materialize, before every start: the held render and the held release
//     of the recorded inputs, each under its own claim selection at the
//     admitted capture (Works: held-config, held-release; MovesClaimPolicy's
//     recovery rows bind them to the recorded inputs), the launch
//     environment's report again, then, for a dirty return, the return
//     generation's verified marker under the admitted writer's fence, then
//     C8's Establish under the start's work, then the §5.1 start check.
//   - Publish, PublishFenced and PublicationEvidence work as the landing's:
//     the dirty return's prepared marker before replication, a fresh
//     generation's marker after its own first root, and this process's
//     receipt read back.
//
// Production: the moves wake source (movesWakeSources, Plan 5 Task 9b)
// registers this owner set, with both switches on. C8 is Task 6's
// provider, the claim's (bindClaimMovesFiles); without one every recovery
// kind is refused, typed.
type movesRecoveryOwners struct{ m *movesProbeOwners }

// recoveryKind reports whether o is a recovery the recovery owners serve.
func recoveryKind(o hostclaim.Observed) bool {
	return !o.Absent && o.Claim != nil && claimRecoveryKind(o.Claim.Op.Kind)
}

// recoveryProbe is what one recovery probe's checks read.
type recoveryProbe struct {
	homeRegion string
	kind       string
	// op is the move id; holdOp the op holding the row: the rollback id for
	// a return, the move id for a revert.
	op, holdOp string
	holder     *hostclaim.Holder
	rec        *provision.Record
	// inputs are the recorded recovery inputs.
	inputs *provision.MoveInputs
	// first: the holder is still at the source's stamp S, so no recovery
	// admission has happened yet.
	first bool
	// dirty: a dirty return. uncovered: a declared or fenced return (Q6).
	dirty, uncovered bool
	srcGen           string
	predecessor      string
	// cut is the cut's journalled session (op.journal.cut_session).
	cut string
	// A dirty return's: the rollback id (its generation), the return seal
	// and the install manifest its install journalled.
	rb       string
	seal     moves.Seal
	manifest moves.InstallManifest
}

// prepared is the dirty return's build as the landing's helpers read one:
// the return generation is the "op" whose verified marker is published.
func (v recoveryProbe) prepared() probeLanding {
	return probeLanding{homeRegion: v.homeRegion, op: v.rb, holder: v.holder, rec: v.rec, seal: v.seal, manifest: v.manifest}
}

// recoveryInputsChanged is a launch input that is not the recorded one: the
// next pass records the inputs again and probes again.
func recoveryInputsChanged(what string) error {
	return fmt.Errorf("%w: the recovery's %s is not the recorded one", hostbox.ErrChanged, what)
}

// recovery checks that o is a recovery probe this box may run and reads the
// record, the recorded inputs and, for a dirty return, the return seal and
// the journalled install manifest.
func (r movesRecoveryOwners) recovery(ctx context.Context, o hostclaim.Observed) (recoveryProbe, error) {
	m := r.m
	var v recoveryProbe
	if !m.valid() {
		return v, hostwrite.ErrUnavailable
	}
	if m.Files == nil {
		return v, claimFilesRefusal()
	}
	if !recoveryKind(o) || o.Claim.Holder == nil {
		return v, hostbox.ErrUnsupported
	}
	c := o.Claim
	home, err := park.VerifiedHome(m.FleetHomeRegion, m.HomeRegion, m.Region, c.Op.Intent["home_region"])
	if err != nil {
		return v, err
	}
	v.homeRegion, v.kind = home, c.Op.Kind
	if c.Op.Phase != hostclaim.PhaseProbing || (v.kind == hostclaim.KindReverting && c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing) {
		return v, hostbox.ErrChanged
	}
	if c.Host != o.Host || c.Op.ID == "" || c.Op.Exec != m.Box || len(c.Op.Participants) == 0 || c.Op.Participants[0] != m.Box ||
		c.Op.Intent["source"] != m.Box || c.Op.Intent["target"] == m.Box || park.SameRegion(m.HomeRegion, c.Op.Intent["to"], m.Region) {
		return v, hostbox.ErrChanged
	}
	for _, lost := range c.Lost {
		if lost == m.Box {
			return v, hostbox.ErrChanged
		}
	}
	v.op, v.holder = c.Op.ID, c.Holder
	stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
	srcGen, journalled := c.Op.Journal[hostclaim.JournalSrcGen]
	if err != nil || stamp == 0 || !journalled {
		return v, fmt.Errorf("%w: the source's stamp and generation are not journalled", hostbox.ErrChanged)
	}
	v.first, v.srcGen, v.cut = v.holder.Stamp == stamp, srcGen, c.Op.Journal[hostclaim.JournalCutSession]
	rec, err := m.Records(ctx, v.holder.Account, v.holder.App, o.Host)
	if err != nil {
		return v, err
	}
	if rec == nil || rec.Host != o.Host || rec.Move == nil {
		return v, hostbox.ErrChanged
	}
	mv := rec.Move
	if mv.ID != v.op || mv.Incarnation != v.holder.Incarnation || !park.PlacementMatches(m.HomeRegion, mv.From, m.Region) {
		return v, hostbox.ErrChanged
	}
	switch v.kind {
	case hostclaim.KindRestoring:
		rb := mv.Rollback
		if mv.State != provision.MoveRollingBack || rb == nil || rb.ID == "" || rb.ID != c.Op.Intent["rollback"] || rb.SourcePhase != provision.RbSourceProbing {
			return v, hostbox.ErrChanged
		}
		v.rb, v.holdOp, v.dirty, v.inputs = rb.ID, rb.ID, rb.Dirty, rb.RecoveryInputs
		// A declared or fenced return (Q6); the record and the claim
		// disagreeing on the target's loss wait (movesloss.go).
		if v.uncovered, err = recoveryDeclared(c, rb.LossRef); err != nil {
			return v, err
		}
		if v.dirty && v.uncovered {
			return v, fmt.Errorf("%w: a declared or fenced return is never dirty", hostbox.ErrChanged)
		}
	default: // hostclaim.KindReverting
		if mv.State != provision.MoveReverting || mv.ReopenedAt != nil {
			return v, hostbox.ErrChanged
		}
		v.holdOp, v.inputs = v.op, mv.RecoveryInputs
	}
	if v.first && !v.dirty && v.holder.Generation != v.srcGen {
		// A clean return names src_gen again (ReturnToSource); a revert
		// never left it.
		return v, fmt.Errorf("%w: the holder is not at the source's generation", hostbox.ErrChanged)
	}
	if v.inputs == nil || v.inputs.Release == "" {
		return v, fmt.Errorf("%w: no recorded recovery inputs", hostbox.ErrChanged)
	}
	predecessor, err := predecessorRegion(v.holder, mv.From)
	if err != nil {
		return v, err
	}
	v.predecessor, v.rec = predecessor, rec
	if !v.dirty {
		return v, nil
	}
	seal, err := returnSeal(c)
	if err != nil {
		return v, err
	}
	v.seal = seal
	j, err := moves.ReadJournal(m.DataDir, o.Host, v.rb)
	if err != nil {
		return v, err
	}
	raw, ok := j.Steps[landStepInstall]
	if !ok {
		return v, fmt.Errorf("%w: the return journalled no install manifest", hostbox.ErrChanged)
	}
	if err := json.Unmarshal([]byte(raw), &v.manifest); err != nil {
		return v, fmt.Errorf("the journalled install manifest: %w", err)
	}
	if v.manifest.Op != v.rb || v.manifest.Host != o.Host || v.manifest.N != v.seal.N || v.manifest.N == 0 || v.manifest.SnapshotKey == "" || v.manifest.SnapshotVersion == "" {
		return v, fmt.Errorf("%w: the install manifest is not this return's build at the seal", hostbox.ErrChanged)
	}
	return v, nil
}

// recoveryLossPending reports a declared loss or a target fence the writer
// is about to admit (writer.go's routing): an unconsumed operation loss, or
// the consumed fence's frozen target writer before its first admission.
func recoveryLossPending(c *hostclaim.Claim) (bool, error) {
	if _, pending := hostclaim.PendingMoveFenceLoss(c); pending {
		return true, nil
	}
	id, err := hostclaim.OperationLossID(c)
	if err != nil {
		return false, err
	}
	return id != "" && !hostclaim.OperationLossConsumed(c, id), nil
}

// preparedAdmission: a dirty return's first admission, into the return
// generation.
func (v recoveryProbe) preparedAdmission() bool { return v.dirty && v.first }

// freshGeneration: g may name a recovery's fresh generation.
func (v recoveryProbe) freshGeneration(g string) bool {
	return g != v.op && (v.rb == "" || g != v.rb) && claimFreshGeneration(v.holder, g)
}

// admission is the recovery's admission and baseline for v, the new
// generation fresh unless it is the dirty return's first (see the file
// comment). dbPath is the held row's database, for PreparedCheck.
func (r movesRecoveryOwners) admission(ctx context.Context, o hostclaim.Observed, v recoveryProbe, dbPath, fresh string) (hostclaim.Admission, *genselect.Plan, error) {
	m := r.m
	h := v.holder
	a := hostclaim.Admission{ObservedOpID: v.op, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: m.Box, Mode: "cold",
		Region: m.Region, PredecessorRegion: v.predecessor}
	var (
		plan *genselect.Plan
		err  error
	)
	if v.preparedAdmission() {
		files, db, err := m.localMarkers(o.Host, dbPath)
		if err != nil {
			return a, nil, err
		}
		if err := moves.PreparedCheck(v.rb, *h, files, db); err != nil {
			return a, nil, err
		}
		if !claimFreshGeneration(h, v.rb) {
			return a, nil, hostbox.ErrChanged
		}
		a.NewGeneration = v.rb
		if v.seal.FromDB {
			a.Coverage = recoveryCoverageSealed
		}
		if plan, err = genselect.SelectPrepared(ctx, m.Data, m.request(o.Host, h), preparedPin(v.prepared())); err != nil {
			return a, nil, err
		}
		// The maximum is the return seal's.
		if plan.MaxTXID != v.seal.N || plan.Pin == nil || *plan.Pin != manifestPin(v.manifest) {
			return a, nil, fmt.Errorf("%w: the return generation is not at the return seal", hostbox.ErrChanged)
		}
	} else {
		a.NewGeneration = fresh
		if v.first && v.kind == hostclaim.KindRestoring && !v.uncovered {
			a.Coverage = recoveryCoverageClean
		}
		if plan, err = r.baseline(ctx, o.Host, v); err != nil {
			return a, nil, err
		}
	}
	a.RemoteMaxTXID = plan.MaxTXID
	return a, plan, nil
}

// baseline is a fresh generation's baseline: the certified remote; for a
// dirty return's restart while the location's marker is not yet the return
// generation's (nor a generation descending from it on this box), the
// return generation's pin (H14); for a clean return while the recovery has
// no root, the recorded source baseline (sourceBaseline, fix round 1 ruling
// 4).
func (r movesRecoveryOwners) baseline(ctx context.Context, host string, v recoveryProbe) (*genselect.Plan, error) {
	m := r.m
	if v.kind == hostclaim.KindRestoring && !v.dirty {
		rooted, err := r.recoveryRooted(ctx, host, v)
		if err != nil {
			return nil, err
		}
		if !rooted {
			return r.sourceBaseline(ctx, host, v)
		}
	}
	if v.dirty {
		digest, err := manifestDigest(v.manifest)
		if err != nil {
			return nil, err
		}
		marker, err := m.readMarker(ctx, v.holder)
		if err != nil {
			return nil, err
		}
		if !m.landingMarker(marker, host, v.prepared(), digest) {
			return genselect.SelectPrepared(ctx, m.Data, m.request(host, v.holder), preparedPin(v.prepared()))
		}
	}
	return genselect.Select(ctx, m.Data, m.request(host, v.holder))
}

// Select chooses the recovery probe's launch inputs and admission. It is
// read-only.
func (r movesRecoveryOwners) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	m := r.m
	v, err := r.recovery(ctx, o)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if !p.Probe {
		return ClaimLaunchInputs{}, hostbox.ErrUnsupported
	}
	if p.Inputs == nil || !sameInputs(v.inputs, *p.Inputs) || (p.Prepared != nil && (!v.dirty || *p.Prepared != v.manifest)) {
		return ClaimLaunchInputs{}, hostbox.ErrChanged
	}
	if err := heldRoute(rt, o.Host, v.holdOp); err != nil {
		return ClaimLaunchInputs{}, err
	}
	fresh := ""
	if !v.preparedAdmission() {
		if fresh = m.NewID(); !v.freshGeneration(fresh) {
			return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
		}
	}
	admission, plan, err := r.admission(ctx, o, v, rt.DBPath, fresh)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	h := v.holder
	ready, err := m.ReadyPath(ctx, h.Account, h.App)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	l, _, err := m.Launch(ctx, rt)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	if l.Version != v.inputs.Release {
		return ClaimLaunchInputs{}, recoveryInputsChanged("live release")
	}
	if v.inputs.EnvReport == "" || localEnvDigest(m.EnvKey, l.Env) != v.inputs.EnvReport {
		return ClaimLaunchInputs{}, recoveryInputsChanged("launch environment")
	}
	return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(v.inputs), Baseline: plan, Admission: admission,
		Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: probeEnv(l.Env),
		DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit, Prepared: v.preparedAdmission(), ReadyPath: ready}, nil
}

// AdmissionEvidence re-derives Select's admission against the observation
// and a baseline selected again, immediately before the admission's CAS.
func (r movesRecoveryOwners) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	m := r.m
	v, err := r.recovery(ctx, o)
	if err != nil {
		return err
	}
	if baseline == nil || in.Mode != "cold" || in.PredecessorCertified || in.LossID != "" || in.KeepStamp {
		return hostbox.ErrChanged
	}
	// Under a pending loss or fence the writer substitutes the validated
	// sources' maxima; only the loss contract vouches for them
	// (movesloss.go). Otherwise the admission is Select's.
	pending, err := recoveryLossPending(o.Claim)
	if err != nil {
		return err
	}
	if pending {
		if err := r.lossEvidence(ctx, o, in, baseline); err != nil {
			return err
		}
	} else if len(in.AdditionalWritesLost) != 0 || in.RemoteMaxTXID != baseline.MaxTXID {
		return hostbox.ErrChanged
	}
	if !v.preparedAdmission() && !v.freshGeneration(in.NewGeneration) {
		return hostbox.ErrChanged
	}
	rt, err := m.route(o.Host, v.holdOp)
	if err != nil {
		return err
	}
	want, fresh, err := r.admission(ctx, o, v, rt.DBPath, in.NewGeneration)
	if err != nil {
		return err
	}
	if in.ObservedOpID != want.ObservedOpID || in.ObservedPhase != want.ObservedPhase || in.ObservedSession != want.ObservedSession || in.ObservedGeneration != want.ObservedGeneration ||
		in.Box != want.Box || !park.SameRegion(m.HomeRegion, in.Region, m.Region) || !park.SameRegion(v.homeRegion, in.PredecessorRegion, want.PredecessorRegion) ||
		in.NewGeneration != want.NewGeneration || in.Coverage != want.Coverage {
		return hostbox.ErrChanged
	}
	if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) || !reflect.DeepEqual(fresh.Pin, baseline.Pin) {
		return hostbox.ErrChanged
	}
	// A clean return's source baseline is pinned here, the owner's last
	// call before the admission's CAS and so before the activator restores
	// it (claimwake.go Restore precedes Materialize): a crash after the
	// restore leaves the pin a restart must match (fix round 2).
	// So is this admission's link in the recovery's lineage: the
	// generation it admits and the one it observed (fix round 3).
	if err := r.recordLineage(o.Host, v, in.NewGeneration); err != nil {
		return err
	}
	return r.recordSourceBaseline(o.Host, v, baseline)
}

// Materialize runs before every start of the admitted recovery probe, in the
// ruled order (see the file comment). The spawn follows only if all of it
// passed.
func (r movesRecoveryOwners) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	m := r.m
	v, err := r.recovery(ctx, o)
	if err != nil {
		return err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return err
	}
	if work == nil {
		return hostwrite.ErrUnavailable
	}
	if in.Identity != launchIdentity(v.inputs) {
		return hostbox.ErrChanged
	}
	fence := writerFence(work, in.Identity)
	if err := fence(ctx); err != nil {
		return err
	}
	rt, err := m.route(o.Host, v.holdOp)
	if err != nil {
		return err
	}
	if rt.DBPath != in.DBPath {
		return hostbox.ErrChanged
	}
	h := v.holder
	captured := hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}
	hr, err := m.Render(ctx, rt, v.holdOp, func(pre appconfig.HeldRender) (hostwrite.Work, error) {
		// The recorded inputs decide before any selection or write.
		if !sameBundle(pre.BundleETags, v.inputs.BundleETags) || pre.OverlayETag != v.inputs.OverlayETag {
			return nil, recoveryInputsChanged("configuration")
		}
		w, err := m.Works(ctx, o.Host, movesActionHeldConfig, captured)
		if err != nil {
			return nil, err
		}
		return heldIdentityWork{Work: w, ID: HeldConfigIdentity(pre.BundleETags, pre.OverlayETag)}, nil
	})
	if err != nil {
		return err
	}
	if !sameBundle(hr.BundleETags, v.inputs.BundleETags) || hr.OverlayETag != v.inputs.OverlayETag {
		return recoveryInputsChanged("configuration")
	}
	released, err := m.Works(ctx, o.Host, movesActionHeldRelease, captured)
	if err != nil {
		return err
	}
	if err := m.Release(ctx, rt, v.inputs.Release, heldIdentityWork{Work: released, ID: hostwrite.Identity{Release: v.inputs.Release}}); err != nil {
		return err
	}
	if localEnvDigest(m.EnvKey, in.Env) != v.inputs.EnvReport {
		return recoveryInputsChanged("launch environment")
	}
	m.recordFence(o.Host, h.Session, work)
	// The start gate: the tree is the one the recovery restored before its
	// first authorized start, and the recovery's own after it.
	if err := r.startGate(ctx, o, v, work, in.Identity); err != nil {
		return err
	}
	if v.dirty {
		receipt, err := m.preparedMarker(ctx, o.Host, v.prepared(), fence)
		if err != nil {
			return err
		}
		if in.Prepared {
			m.record(o.Host, h.Session, receipt)
		}
	}
	if err := m.Files.Establish(ctx, o, in, work); err != nil {
		return err
	}
	return m.startCheck(ctx, o, in)
}

// EmptyAuthority: a recovery never admits the empty bootstrap.
func (r movesRecoveryOwners) EmptyAuthority(context.Context, hostclaim.Observed, ClaimLaunchInputs) error {
	return hostbox.ErrUnsupported
}

// Publish is Group D's call for the dirty return's prepared admission,
// before replication: Materialize published the return generation's
// verified marker; this finds it under the admitted writer's fence and
// records the receipt.
func (r movesRecoveryOwners) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	m := r.m
	v, err := r.recovery(ctx, o)
	if err != nil {
		return err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return err
	}
	h := v.holder
	if !v.dirty || !in.Prepared || in.Admission.NewGeneration != v.rb || h.Generation != v.rb {
		return hostbox.ErrChanged
	}
	work := m.fence(o.Host, h.Session)
	if work == nil {
		return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
	}
	receipt, err := m.preparedMarker(ctx, o.Host, v.prepared(), writerFence(work, in.Identity))
	if err != nil {
		return err
	}
	if receipt.MarkerVersion == "" {
		return hostbox.ErrChanged
	}
	m.record(o.Host, h.Session, receipt)
	return nil
}

// PublishFenced writes a fresh recovery generation's marker after the
// session's own first root, under the publisher's fence, as the landing's
// restarted attempt and serving do.
func (r movesRecoveryOwners) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	m := r.m
	v, err := r.recovery(ctx, o)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if err := admittedProbe(o, in, m.Box); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if pub.Sync == nil || pub.Check == nil {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	h := v.holder
	if in.Prepared || h.Generation == v.op || (v.rb != "" && h.Generation == v.rb) {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	work := m.fence(o.Host, h.Session)
	if work == nil {
		return ClaimPublicationReceipt{}, fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
	}
	// The files on disk must be the tree the recovery placed, held by this
	// recovery's admitted lineage, before anything is published (fix round
	// 3): its own writes since its first start are authoritative.
	placed, ok, err := r.placement(o.Host, v)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if !ok {
		return ClaimPublicationReceipt{}, fmt.Errorf("%w: the recovery recorded no placed tree", hostbox.ErrChanged)
	}
	if err := work.With(ctx, in.Identity, func() error {
		_, err := r.vouchFiles(o.Host, v, placed)
		return err
	}); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	return m.publishFresh(ctx, o.Host, h, pub, func() error { return r.refreshFilesGen(ctx, o.Host, v, placed, work, in.Identity) })
}

// vouchFiles requires the files on disk to be the placed tree (identity: a
// replaced tree is refused) and the holder's session to be this recovery's
// own through its admitted lineage, and reports whether .files-gen will
// move to holder.generation: it names that generation's cold predecessor
// for this incarnation. It never checks content: since the first start the
// tree is the recovery's, and its writes are authoritative (fix round 3,
// spec §7.8). The caller runs it under the session's fence.
func (r movesRecoveryOwners) vouchFiles(host string, v recoveryProbe, placed recoveryPlacement) (bool, error) {
	m := r.m
	h := v.holder
	if err := r.verifyPlaced(host, placed); err != nil {
		return false, err
	}
	if err := r.lineage(host, v); err != nil {
		return false, err
	}
	dir, _ := m.FilesDir(host)
	files, _, err := m.localMarkers(host, filepath.Join(dir, ".none"))
	if err != nil || files == nil || files.Incarnation != h.Incarnation || files.Op == h.Generation {
		return false, err
	}
	if pred, ok := coldPredecessor(h, h.Generation); !ok || pred == "" || files.Op != pred {
		return false, nil
	}
	return true, nil
}

// refreshFilesGen points .files-gen at the published generation, inside the
// admitted writer's fence, after the files on disk are proved the tree the
// recovery placed and the session this recovery's own (vouchFiles); a
// replaced tree or an unproven session fails the publication and relabels
// nothing. It relabels only a .files-gen naming the generation's cold
// predecessor for this incarnation (the files the start check accepted for
// this session); absent, or anything else, is left for the start check to
// decide.
func (r movesRecoveryOwners) refreshFilesGen(ctx context.Context, host string, v recoveryProbe, placed recoveryPlacement, work hostwrite.Work, id hostwrite.Identity) error {
	m := r.m
	h := v.holder
	dir, _ := m.FilesDir(host)
	return work.With(ctx, id, func() error {
		relabel, err := r.vouchFiles(host, v, placed)
		if err != nil || !relabel {
			return err
		}
		body, err := json.Marshal(moves.FilesGen{Op: h.Generation, Incarnation: h.Incarnation})
		if err != nil {
			return err
		}
		if err := syncPath(dir); err != nil {
			return err
		}
		return writeFileAtomic(filepath.Join(dir, archive.MarkerName), body, 0o644)
	})
}

// PublicationEvidence requires this process's receipt for the admitted
// session, as the landing's does; the return generation's marker also
// carries its build's pin and install.json's digest.
func (r movesRecoveryOwners) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
	v, err := r.recovery(ctx, o)
	if err != nil {
		return err
	}
	prepared := ""
	if v.dirty {
		prepared = v.rb
	}
	return r.m.publicationEvidence(ctx, o.Host, v.holder, prepared, v.manifest)
}

// rcStepFilesPlaced is the recovery journal's record of the tree it placed
// (JSON recoveryPlacement): the files directory's and <host>.env's
// identities and the tree's plain digest before the recovery's first start.
// The recovery journal is the rollback's (a return) or the move's (a
// revert).
const rcStepFilesPlaced = "rc_files_placed"

// rcStepLineage prefixes the recovery journal's record of each generation
// the recovery admitted (rcStepLineage+<generation>, JSON recoveryLineage).
const rcStepLineage = "rc_lineage."

// recoveryPlacement is what rcStepFilesPlaced records.
type recoveryPlacement struct {
	Files string `json:"files"`
	Env   string `json:"env,omitempty"`
	Plain string `json:"plain"`
}

// journalOp is the recovery's journal: the rollback's for a return, the
// move's for a revert.
func (v recoveryProbe) journalOp() string {
	if v.kind == hostclaim.KindRestoring {
		return v.rb
	}
	return v.op
}

// placement reads the recorded placed tree; false when none is recorded.
func (r movesRecoveryOwners) placement(host string, v recoveryProbe) (recoveryPlacement, bool, error) {
	j, err := moves.ReadJournal(r.m.DataDir, host, v.journalOp())
	if err != nil {
		return recoveryPlacement{}, false, err
	}
	raw, ok := j.Steps[rcStepFilesPlaced]
	if !ok {
		return recoveryPlacement{}, false, nil
	}
	var p recoveryPlacement
	if err := json.Unmarshal([]byte(raw), &p); err != nil || p.Files == "" || p.Plain == "" {
		return recoveryPlacement{}, false, fmt.Errorf("%w: the recorded placed tree does not read (%v)", hostbox.ErrChanged, err)
	}
	return p, true, nil
}

// startGate runs under the start's work before every start. Before the
// recovery's first authorized start (no placement journalled) the tree must
// be the one the recovery restored (recordPlacement); after it, the tree is
// the recovery's own: its identity is the placed one, and its content is
// never compared again (fix round 3). Every start's session must be this
// recovery's through its admitted lineage.
func (r movesRecoveryOwners) startGate(ctx context.Context, o hostclaim.Observed, v recoveryProbe, work hostwrite.Work, id hostwrite.Identity) error {
	placed, ok, err := r.placement(o.Host, v)
	if err != nil {
		return err
	}
	if !ok {
		expected, err := restoredDigest(o.Claim, v)
		if err != nil {
			return err
		}
		if err := r.recordPlacement(ctx, o.Host, v, expected, work, id); err != nil {
			return err
		}
	} else if err := work.With(ctx, id, func() error { return r.verifyPlaced(o.Host, placed) }); err != nil {
		return err
	}
	return r.lineage(o.Host, v)
}

// restoredDigest is the plain digest of the tree the recovery restores, as
// its baselines record it: a dirty return's is the return baseline the
// target journalled in the claim (the tree its install extracted); a clean
// return's or a revert's is the source's stopped baseline (Move.Source,
// which the clean verification and the revert's verify compared). "" when
// the record keeps none (a revert from stopping, J8): the first start then
// takes the tree as it finds it.
func restoredDigest(c *hostclaim.Claim, v recoveryProbe) (string, error) {
	if v.dirty {
		raw, ok := c.Op.Journal[moves.JournalReturnBaseline]
		if !ok {
			return "", fmt.Errorf("%w: the claim carries no return baseline", hostbox.ErrChanged)
		}
		b, err := moves.DecodeReturnBaseline(raw)
		if err != nil {
			return "", err
		}
		return b.Files.Plain, nil
	}
	if src := v.rec.Move.Source; src != nil && src.Files != nil {
		return src.Files.Plain, nil
	}
	return "", nil
}

// recordPlacement journals the tree the recovery restored, as placed, under
// the start's work, before the recovery's first start: the files directory's
// and <host>.env's identities and the plain digest, which must be expected
// (the restored tree's, restoredDigest) when that is known. A dirty
// return's tree is the one its install placed (landStepFilesPlaced), and
// must still be. A tree that is not the restored one is refused and nothing
// is journalled, so the next attempt checks again. Once recorded it is never
// rewritten.
func (r movesRecoveryOwners) recordPlacement(ctx context.Context, host string, v recoveryProbe, expected string, work hostwrite.Work, id hostwrite.Identity) error {
	m := r.m
	dir, env := m.FilesDir(host)
	var p recoveryPlacement
	if err := work.With(ctx, id, func() error {
		var err error
		if p.Files, err = pathIdentity(dir); err != nil {
			return err
		}
		switch envID, err := pathIdentity(env); {
		case err == nil:
			p.Env = envID
		case !errors.Is(err, fs.ErrNotExist):
			return err
		}
		if v.dirty {
			j, err := moves.ReadJournal(m.DataDir, host, v.rb)
			if err != nil {
				return err
			}
			installed, err := journalledStaged(j, landStepFilesPlaced)
			if err != nil {
				return fmt.Errorf("%w: %v", hostbox.ErrChanged, err)
			}
			if installed.Files != p.Files || installed.Env != p.Env {
				return fmt.Errorf("%w: the files are not the tree the return installed", hostbox.ErrChanged)
			}
		}
		if p.Plain, err = m.filesDigest(host); err != nil {
			return err
		}
		if expected != "" && p.Plain != expected {
			return fmt.Errorf("%w: the files are not the tree the recovery restored", hostbox.ErrChanged)
		}
		return nil
	}); err != nil {
		return err
	}
	body, err := json.Marshal(p)
	if err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	err = moves.UpdateJournal(m.DataDir, host, v.journalOp(), func(j *moves.Journal) error {
		if _, ok := j.Steps[rcStepFilesPlaced]; ok {
			return errJournalUnchanged
		}
		j.Mark(rcStepFilesPlaced, string(body))
		return nil
	})
	if errors.Is(err, errJournalUnchanged) {
		return nil
	}
	return err
}

// verifyPlaced requires the files directory and <host>.env on disk to be
// the ones the recovery placed.
func (r movesRecoveryOwners) verifyPlaced(host string, p recoveryPlacement) error {
	dir, env := r.m.FilesDir(host)
	if ok, err := sameIdentity(dir, p.Files); err != nil || !ok {
		return fmt.Errorf("%w: %s is not the files directory the recovery placed (%v)", hostbox.ErrChanged, dir, err)
	}
	if p.Env != "" {
		if ok, err := sameIdentity(env, p.Env); err != nil || !ok {
			return fmt.Errorf("%w: %s is not the environment file the recovery placed (%v)", hostbox.ErrChanged, env, err)
		}
	}
	return nil
}

// recoveryLineage is one rcStepLineage record: the generation an admission
// of this recovery observed (its predecessor), and whether it was the
// recovery's first admission (Root: the holder still at the source's stamp
// S).
type recoveryLineage struct {
	Predecessor string `json:"predecessor"`
	Root        bool   `json:"root,omitempty"`
}

// recordLineage journals the admission's link before its CAS, in its
// evidence: the generation it admits, the generation it observed. The
// claim's CAS admits only over the observed generation, so a holder at a
// journalled generation proves that this recovery admitted it over its
// predecessor. A link for a generation never admitted (a refused CAS) is
// inert: generation names are fresh.
func (r movesRecoveryOwners) recordLineage(host string, v recoveryProbe, g string) error {
	m := r.m
	body, err := json.Marshal(recoveryLineage{Predecessor: v.holder.Generation, Root: v.first})
	if err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	err = moves.UpdateJournal(m.DataDir, host, v.journalOp(), func(j *moves.Journal) error {
		if have, ok := j.Steps[rcStepLineage+g]; ok {
			if have != string(body) {
				return fmt.Errorf("%w: generation %s is already another admission's", hostbox.ErrChanged, g)
			}
			return errJournalUnchanged
		}
		j.Mark(rcStepLineage+g, string(body))
		return nil
	})
	if errors.Is(err, errJournalUnchanged) {
		return nil
	}
	return err
}

// lineage proves the holder's session is this recovery's own: this box
// writes the holder's generation, admitted in this region, and from it the
// journalled links reach the recovery's first admission, each generation
// admitted by this recovery over the one before. Any other writer admitted
// in between breaks the chain (its generation was never journalled here):
// the ownership interval is unproven and refused.
//
// The links bind generations, not sessions, and that is complete because
// the claim admits no writer during a restoring or reverting op except
// through these owners, and never a warm one (fix round 4; checked against
// the claim code at e0f4d1f7):
//
//   - the holder's session, generation and writer box change only in
//     AdmitWriter's CAS (hostclaim/generation.go:166-170); ReturnToSource
//     sets the generation and stamp only from released to claimed, before
//     any recovery admission (hostclaim/move.go:1025-1046);
//   - AdmitWriter's only production callers are hostbox.Writer's three
//     paths (hostbox/writer.go:63; writer_loss.go:58 through
//     hostclaim/lost.go:218; writer_loss.go:79 through
//     hostclaim/move_fence_admission.go:44);
//   - Writer.Admit refuses every warm or certified admission before any
//     path (hostbox/writer.go:34-36), so no admission keeps a generation
//     while advancing the session: the counterexample's intervening warm
//     admission cannot be written;
//   - only the op's executor admits (hostbox/writer.go:37,
//     hostclaim/generation.go:40-42), under the observed op and phase
//     (hostclaim/generation.go:43-45);
//   - every cold admission's evidence, on every path, is
//     ClaimOwnerDispatch.AdmissionEvidence (host/claimassembly.go:181,
//     host/claimconstruction.go:267), which hands restoring and reverting
//     only to the moves owners (host/claimowners.go:220-224), which hand them
//     to these recovery owners (movesprobeowners.go, recoveryKind); their
//     AdmissionEvidence journals the link before it returns nil. Without
//     the moves owners bound, the serving owners' evidence refuses any move
//     kind (host/claimowners.go:328).
//
// So every session after the root is a cold admission evidenced here, and
// a chain of generations is a chain of this recovery's sessions.
// TestRecoveryOpAdmitsNoOutsideWriter pins the refusals.
func (r movesRecoveryOwners) lineage(host string, v recoveryProbe) error {
	m := r.m
	h := v.holder
	n := len(h.Admissions)
	if n == 0 || h.WriterBox != m.Box {
		return fmt.Errorf("%w: this box does not write the holder's session", hostbox.ErrChanged)
	}
	if a := h.Admissions[n-1]; a.Box != m.Box || a.Generation != h.Generation || !park.SameRegion(m.HomeRegion, a.Region, m.Region) {
		return fmt.Errorf("%w: the holder's admission is not this box's", hostbox.ErrChanged)
	}
	j, err := moves.ReadJournal(m.DataDir, host, v.journalOp())
	if err != nil {
		return err
	}
	seen := map[string]bool{}
	for g := h.Generation; !seen[g]; {
		seen[g] = true
		raw, ok := j.Steps[rcStepLineage+g]
		if !ok {
			return fmt.Errorf("%w: generation %s is not this recovery's admission", hostbox.ErrChanged, g)
		}
		var link recoveryLineage
		if err := json.Unmarshal([]byte(raw), &link); err != nil {
			return fmt.Errorf("%w: the lineage of %s does not read: %v", hostbox.ErrChanged, g, err)
		}
		if link.Root {
			return nil
		}
		g = link.Predecessor
	}
	return fmt.Errorf("%w: the recovery's lineage loops", hostbox.ErrChanged)
}

// rcStepSourceBaseline is the rollback journal's record of a clean return's
// source baseline (JSON genselect.Plan): src_gen's files by exact version,
// as the first admission that restores them was evidenced.
const rcStepSourceBaseline = "rc_source_baseline"

// recoveryRooted reports whether the clean return's recovery has a root: an
// object under a generation one of its admissions made (this box, this
// region, a session after the cut's). Before it has, ordinary selection
// would follow the claim's predecessor chain through the target's admission
// (AdmitWriter records it as the first recovery admission's predecessor,
// though the restore was from src_gen) and can end in a covered baseline.
func (r movesRecoveryOwners) recoveryRooted(ctx context.Context, host string, v recoveryProbe) (bool, error) {
	m := r.m
	h := v.holder
	cut, err := strconv.ParseUint(v.cut, 10, 64)
	if err != nil {
		return false, fmt.Errorf("%w: the cut's session is not journalled", hostbox.ErrChanged)
	}
	location := strings.TrimSuffix(h.Location, "/")
	for _, a := range h.Admissions {
		if a.Box != m.Box || a.Session <= cut || a.Generation == "" || !park.SameRegion(m.HomeRegion, a.Region, m.Region) {
			continue
		}
		prefix := location + "/g-" + a.Generation + "/"
		versions, err := m.Data.ListVersions(ctx, m.Bucket, prefix)
		if err != nil {
			return false, err
		}
		for _, ver := range versions {
			if strings.HasPrefix(ver.Key, prefix) && ver.IsLatest && !ver.DeleteMarker {
				return true, nil
			}
		}
	}
	return false, nil
}

// sourceBaseline is a clean return's restore point before the recovery has
// a root: src_gen in this region, selected as the first admission selected
// it (the holder viewed at src_gen, whose marker names it), and, once
// recorded, exactly the recorded files by version.
func (r movesRecoveryOwners) sourceBaseline(ctx context.Context, host string, v recoveryProbe) (*genselect.Plan, error) {
	m := r.m
	view := *v.holder
	view.Generation = v.srcGen
	plan, err := genselect.Select(ctx, m.Data, m.request(host, &view))
	if err != nil {
		return nil, err
	}
	if plan.Generation != selectedGeneration(v.srcGen) {
		return nil, fmt.Errorf("%w: the source baseline is not src_gen", hostbox.ErrChanged)
	}
	recorded, ok, err := r.recordedSourceBaseline(host, v)
	if err != nil {
		return nil, err
	}
	if ok && (recorded.Generation != plan.Generation || recorded.MaxTXID != plan.MaxTXID || !reflect.DeepEqual(recorded.Files, plan.Files) || !reflect.DeepEqual(recorded.Pin, plan.Pin)) {
		return nil, fmt.Errorf("%w: src_gen is not the recorded source baseline", hostbox.ErrChanged)
	}
	return plan, nil
}

func (r movesRecoveryOwners) recordedSourceBaseline(host string, v recoveryProbe) (genselect.Plan, bool, error) {
	j, err := moves.ReadJournal(r.m.DataDir, host, v.journalOp())
	if err != nil {
		return genselect.Plan{}, false, err
	}
	raw, ok := j.Steps[rcStepSourceBaseline]
	if !ok {
		return genselect.Plan{}, false, nil
	}
	var p genselect.Plan
	if err := json.Unmarshal([]byte(raw), &p); err != nil || p.Generation == "" || len(p.Files) == 0 {
		return genselect.Plan{}, false, fmt.Errorf("%w: the recorded source baseline does not read (%v)", hostbox.ErrChanged, err)
	}
	return p, true, nil
}

// recordSourceBaseline journals a clean return's source baseline from the
// admission's evidence, before the admission's CAS and the restore; once
// recorded it is never rewritten.
func (r movesRecoveryOwners) recordSourceBaseline(host string, v recoveryProbe, plan *genselect.Plan) error {
	if v.kind != hostclaim.KindRestoring || v.dirty || plan == nil || plan.Generation != selectedGeneration(v.srcGen) {
		return nil
	}
	m := r.m
	body, err := json.Marshal(genselect.Plan{Generation: plan.Generation, Files: plan.Files, Root: plan.Root, Pin: plan.Pin, MaxTXID: plan.MaxTXID, MaxLevel0TXID: plan.MaxLevel0TXID})
	if err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	err = moves.UpdateJournal(m.DataDir, host, v.journalOp(), func(j *moves.Journal) error {
		if _, ok := j.Steps[rcStepSourceBaseline]; ok {
			return errJournalUnchanged
		}
		j.Mark(rcStepSourceBaseline, string(body))
		return nil
	})
	if errors.Is(err, errJournalUnchanged) {
		return nil
	}
	return err
}

// selectedGeneration is generation g as a selection names it: the claim
// writes the origin as "" (a legacy source's src_gen) and genselect as
// hostclaim.RootGeneration.
func selectedGeneration(g string) string {
	if g == "" {
		return hostclaim.RootGeneration
	}
	return g
}