package hostclaim

import (
	"context"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"time"

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

// Move phases on the claim's op (moves spec §7.5, §19.1). The record
// carries the detailed journal; the claim carries only what the flip,
// the commit and a rollback compare.
const (
	PhaseMoving   = ""         // moving, before the seal
	PhaseSealed   = "sealed"   // moving, after the cut's staged phase
	PhaseAborting = "aborting" // moving, the abort has begun (Abort): the lane park is queued
)

// Journal keys the cut and the flip write on the claim's op (moves spec
// §7.4, §7.5). The record carries the detail; the claim carries what the
// flip, the commit and a rollback compare.
const (
	JournalSeal           = "seal"            // the sealed cut's summary
	JournalSrcGen         = "src_gen"         // the source's generation, journalled at the flip
	JournalSourceStamp    = "source_stamp"    // the source installation's stamp S, journalled at the flip
	JournalCutSession     = "cut_session"     // the cut's session
	JournalRevPhase       = "rev_phase"       // a revert's own phase
	JournalEvidenceSource = "evidence_source" // the source's stop evidence (K3)
	JournalEvidenceTarget = "evidence_target" // the target's stop evidence (K3)
)

// journalReleasedMove is the key under which a serving op remembers the
// move that released it (K8).
const journalReleasedMove = "released_move"

// BeginMove is K2: the request door CASes serving → moving, with the move
// id, both participants and exec = the source box. The source must already
// be the executor; nothing moves exec here. Touched is not widened: K12
// has every write site add its own region or box first (Touch). Every
// other claim field (holder, admissions, generation, chores) is left as
// found.
func BeginMove(opID, by, sourceBox, targetBox, targetRegion string, now time.Time) func(*Claim) error {
	return func(c *Claim) error {
		if err := StartGuard(c, []string{targetRegion}, []string{targetBox}, ""); err != nil {
			return err
		}
		if opID == "" || sourceBox == "" || targetBox == "" || targetRegion == "" {
			return fmt.Errorf("%w: a move needs an op id, two boxes and a target region", ErrMalformed)
		}
		if c.Op.Exec != sourceBox {
			return fmt.Errorf("%w: exec is %q, the move's source is %q", ErrWrongState, c.Op.Exec, sourceBox)
		}
		c.Op = Op{Kind: KindMoving, ID: opID, Phase: PhaseMoving, By: by, StartedAt: now.UTC(), Exec: sourceBox,
			Participants: []string{sourceBox, targetBox},
			Intent:       map[string]string{"to": targetRegion, "source": sourceBox, "target": targetBox}}
		return nil
	}
}

// MoveAdmission is what a claim-mode request door proved before its start
// CAS (claim Plan 5 Task 7): the canonical app id, the id of the exact
// "move" registration the control writer minted for this claim observation
// (always the move's op id, so admission draining keeps it while the op
// holds the claim), and both boxes' enrollments from the signed roster.
// hostclaim only records them; hostops.BeginClaimMove does the proving.
type MoveAdmission struct {
	HomeRegion                                             string
	AppID, AdmissionID, SourceEnrollment, TargetEnrollment string
}

// BeginAdmittedMove is BeginMove plus the admission it ran under, in the same
// closure and so the same CAS. The move's own keys ("source", "target", "to")
// are unchanged; "app_id", "admission_id", "source_enrollment" and
// "target_enrollment" are additive. ConvertMovingToRelocating requires the
// first three.
func BeginAdmittedMove(opID, by, sourceBox, targetBox, targetRegion string, now time.Time, a MoveAdmission) func(*Claim) error {
	return func(c *Claim) error {
		if a.AppID == "" || a.AdmissionID == "" || a.AdmissionID != opID || !validEnrollment(a.SourceEnrollment) || !validEnrollment(a.TargetEnrollment) || a.SourceEnrollment == a.TargetEnrollment {
			return fmt.Errorf("%w: an admitted move needs its app, its registration and both enrollments", ErrMalformed)
		}
		if a.HomeRegion != "" && regionname.PlacementRegion(a.HomeRegion) != a.HomeRegion {
			return ErrMalformed
		}
		regions := []string{targetRegion}
		if regionname.SameRegion(a.HomeRegion, targetRegion, regionname.HomeRegion) {
			regions = append(regions, a.HomeRegion, regionname.HomeRegion)
		}
		if err := StartGuard(c, regions, []string{targetBox}, ""); err != nil {
			return err
		}
		if err := BeginMove(opID, by, sourceBox, targetBox, targetRegion, now)(c); err != nil {
			return err
		}
		if a.HomeRegion != "" {
			c.Op.Intent["home_region"] = a.HomeRegion
		}
		c.Op.Intent["app_id"] = a.AppID
		c.Op.Intent["admission_id"] = a.AdmissionID
		c.Op.Intent["source_enrollment"] = a.SourceEnrollment
		c.Op.Intent["target_enrollment"] = a.TargetEnrollment
		return nil
	}
}

// validEnrollment is the roster's enrollment token: 32 lowercase hex.
func validEnrollment(e string) bool {
	if len(e) != 32 {
		return false
	}
	for _, r := range e {
		if (r < '0' || r > '9') && (r < 'a' || r > 'f') {
			return false
		}
	}
	return true
}

// Touch is K12: a write site adds its region or box to touched before its
// first write there, under the op it serves. Idempotent.
func Touch(opID string, regions, boxes []string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		c.Touched.Regions = union(c.Touched.Regions, regions)
		c.Touched.Boxes = union(c.Touched.Boxes, boxes)
		return nil
	}
}

// RequireMoveOp is K5's revalidation in one closure: op id, kind and phase
// must be what the step expects.
func RequireMoveOp(opID, kind, phase string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != kind || c.Op.Phase != phase {
			return fmt.Errorf("%w: op is %s/%q, step expects %s/%q", ErrWrongState, c.Op.Kind, c.Op.Phase, kind, phase)
		}
		return nil
	}
}

// ReleaseMove is K8: a move hands the host back to serving. It is legal
// from moving before the seal (phase "": the request door's compensation for
// a move whose record start failed), and from the end of each move outcome,
// read from op.phase (J1, J6): a committed landing, an opened rollback
// (restoring) and an opened revert (reverting). In those three the executor
// named must be the one executing. An aborting op is ReleaseAbort's (Plan 5
// ruling K7), never this. It is legal only once no obligation under this op
// remains. The serving op keeps exec and records which move released it:
// operation-bound completion evidence for the abort's crash boundary
// (RequireReleasedMove).
func ReleaseMove(opID, execBox string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		door := c.Op.Kind == KindMoving && c.Op.Phase == PhaseMoving
		ended := (c.Op.Kind == KindLanding && c.Op.Phase == PhaseCommitted) ||
			((c.Op.Kind == KindRestoring || c.Op.Kind == KindReverting) && c.Op.Phase == PhaseOpened)
		if !door && !ended {
			return fmt.Errorf("%w: op is %s/%q, a release needs moving before the seal, a committed landing or an opened recovery", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		if execBox == "" {
			return fmt.Errorf("%w: a release names the executor", ErrMalformed)
		}
		if ended && c.Op.Exec != execBox {
			return fmt.Errorf("%w: exec is %q, the release names %q", ErrWrongState, c.Op.Exec, execBox)
		}
		return releaseToServing(c, opID, execBox)
	}
}

// releaseToServing is every move release's last step: legal only once no
// obligation under the op remains; the serving op keeps the executor and
// records the move it released.
func releaseToServing(c *Claim, opID, execBox string) error {
	for _, o := range c.Obligations {
		if o.Op == opID {
			return fmt.Errorf("%w: %s %s%s is still owed under this move", ErrPendingChore, o.Kind, o.Region, o.Box)
		}
	}
	c.Op = Op{Kind: KindServing, Exec: execBox, Journal: map[string]string{journalReleasedMove: opID}}
	return nil
}

// Abort is the claim half of a move's abort (moves spec §7.11, joint N13;
// Plan 5 Task 1): this move's moving op before the seal, with the source
// still executing, goes to moving/aborting, and the park of its own staging
// lane is queued in the same write for the target region, owned by that
// region (QueueStagingPark's chore, index D6). The source cannot read or
// park the target's moves/*, so the lane is the target region's to retire;
// execution stays with the source, which serves throughout.
//
// The scope is moves/<host>/<op> with the holder's incarnation and the
// journalled abort bound the caller computed (moves.AbortScope, ruling K5).
// A replay finds the op already aborting: it is a no-op when the same lane
// park is owed, or when none is (the target already acknowledged it), and
// is refused when a lane park with another scope is owed. It allocates no
// ids. A cut past requested cannot abort, and neither can an op whose
// execution was given up (that is a revert).
func Abort(opID, targetRegion string, scope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindMoving || (c.Op.Phase != PhaseMoving && c.Op.Phase != PhaseAborting) || c.Op.Exec == "" {
			return fmt.Errorf("%w: op is %s/%q exec %q, an abort needs moving before the seal with the source executing", ErrWrongState, c.Op.Kind, c.Op.Phase, c.Op.Exec)
		}
		if targetRegion == "" || c.Op.Intent["to"] != targetRegion {
			return fmt.Errorf("%w: %q is not the move's target region", ErrWrongState, targetRegion)
		}
		var err error
		if scope, err = parkQueueGuard(c, opID, targetRegion, scope, newID); err != nil {
			return err
		}
		// Restates moves.ScopeLocation(host, op), as QueueStagingPark does.
		if scope.Location != "moves/"+c.Host+"/"+opID {
			return fmt.Errorf("%w: an abort parks this move's own lane, moves/%s/%s", ErrMalformed, c.Host, opID)
		}
		if c.Op.Phase == PhaseAborting {
			for _, o := range c.Obligations {
				if o.Op == opID && o.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, targetRegion) && o.Park != nil && isStagingLocation(o.Park.Location) && !sameParkScope(*o.Park, scope) {
					return fmt.Errorf("%w: %s already owes its lane park in %s with another scope", ErrWrongState, opID, targetRegion)
				}
			}
			return nil
		}
		if _, err := queueReplicaPark(c, opID, targetRegion, scope, newID); err != nil {
			return err
		}
		c.Op.Phase = PhaseAborting
		return nil
	}
}

// ReleaseAbort is the abort's release (ruling K7): from moving/aborting
// only, so always after Abort queued the lane park, with the named source
// still executing, and only once no obligation under the op remains (the
// target region acknowledged the lane park). The serving op keeps the
// source as its executor and records the move it released.
func ReleaseAbort(opID, sourceBox string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireMoveOp(opID, KindMoving, PhaseAborting)(c); err != nil {
			return err
		}
		if sourceBox == "" {
			return fmt.Errorf("%w: a release names the executor", ErrMalformed)
		}
		if c.Op.Exec != sourceBox {
			return fmt.Errorf("%w: exec is %q, the abort's release names %q", ErrWrongState, c.Op.Exec, sourceBox)
		}
		return releaseToServing(c, opID, sourceBox)
	}
}

// RequireReleasedMove is the replay rule for a crashed abort: nil when
// the claim is serving and its op records that move opID released it,
// else ErrStaleOp. A plain serving claim, or one released by another
// move, does not satisfy it.
func RequireReleasedMove(opID string) func(*Claim) error {
	return func(c *Claim) error {
		if opID == "" || c.Op.Kind != KindServing || c.Op.Journal[journalReleasedMove] != opID {
			return fmt.Errorf("%w: the claim does not record move %q as released", ErrStaleOp, opID)
		}
		return nil
	}
}

func setJournal(c *Claim, k, v string) {
	if c.Op.Journal == nil {
		c.Op.Journal = map[string]string{}
	}
	c.Op.Journal[k] = v
}

// SetJournal records move journal entries under the op (D1): what the
// claim half must read with the same fresh read as an admission. It names
// the op only, so any kind may use it.
func SetJournal(opID string, kv map[string]string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		for k, v := range kv {
			setJournal(c, k, v)
		}
		return nil
	}
}

// SetMovePhase moves the op's phase one step, under its kind. A replay
// finds the phase already moved and is refused (ErrWrongState).
func SetMovePhase(opID, kind, from, to string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireMoveOp(opID, kind, from)(c); err != nil {
			return err
		}
		c.Op.Phase = to
		return nil
	}
}

// ExecToNone is K3: the participant that holds execution and has written
// stop evidence for this op gives it up, citing the evidence under
// journalKey (evidence_source or evidence_target). Nobody moves exec from
// one box to another. A replay after a crash finds exec already empty and
// the same evidence journalled: that is idempotent, not a conflict.
func ExecToNone(opID, box, evidenceKey, journalKey string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if box == "" || evidenceKey == "" || (journalKey != JournalEvidenceSource && journalKey != JournalEvidenceTarget) {
			return fmt.Errorf("%w: execution is given up by a named executor, citing evidence", ErrMalformed)
		}
		if c.Op.Exec == "" && c.Op.Journal[journalKey] == evidenceKey {
			return nil
		}
		if c.Op.Exec != box {
			return fmt.Errorf("%w: exec is %q, not %q", ErrWrongState, c.Op.Exec, box)
		}
		c.Op.Exec = ""
		setJournal(c, journalKey, evidenceKey)
		return nil
	}
}

// phaseLandingChecking is the landing's first phase, the one the flip writes.
const phaseLandingChecking = "checking"

// Flip is the claim half of the record's flip (§7.5, K4): moving/sealed to
// landing/checking, exec none to the target, citing the source's evidence.
// holder.stamp becomes the gen this CAS produces (the target's installation
// stamp F; Update sets gen = observed + 1 after the transition runs, as
// CompleteCreate relies on); the source's stamp S and generation are
// journalled for a rollback's return. holder.generation is not touched.
//
// Either participant may write it (G12: the target after a source crash),
// so a second write of the same flip is a no-op: the claim is already
// landing/checking for this op and target with the same journalled S and
// generation. It does not restamp the holder. Any other flip over an
// already-flipped claim is refused.
func Flip(opID, targetBox string, sourceStamp uint64, srcGen string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind == KindLanding && c.Op.Phase == phaseLandingChecking && c.Op.Exec == targetBox &&
			len(c.Op.Participants) == 2 && c.Op.Participants[1] == targetBox &&
			c.Op.Journal[JournalSourceStamp] == strconv.FormatUint(sourceStamp, 10) && c.Op.Journal[JournalSrcGen] == srcGen {
			return nil
		}
		if err := RequireMoveOp(opID, KindMoving, PhaseSealed)(c); err != nil {
			return err
		}
		if c.Holder == nil {
			return fmt.Errorf("%w: the flip needs the holder", ErrWrongState)
		}
		if c.Op.Exec != "" || c.Op.Journal[JournalEvidenceSource] == "" {
			return fmt.Errorf("%w: the flip needs exec none and the source's evidence", ErrWrongState)
		}
		// The first flip names the source's own installation: its stamp S and
		// generation are what a rollback returns to (§6.4, §7.8), so they
		// must be what the claim holds now, before the stamp is overwritten.
		if sourceStamp != c.Holder.Stamp || srcGen != c.Holder.Generation {
			return fmt.Errorf("%w: the flip names stamp %d generation %q, the holder is at %d and %q", ErrWrongState, sourceStamp, srcGen, c.Holder.Stamp, c.Holder.Generation)
		}
		if len(c.Op.Participants) != 2 || targetBox == "" || c.Op.Participants[1] != targetBox {
			return fmt.Errorf("%w: %q is not the move's target", ErrWrongState, targetBox)
		}
		c.Op.Kind, c.Op.Phase, c.Op.Exec = KindLanding, phaseLandingChecking, targetBox
		setJournal(c, JournalSourceStamp, strconv.FormatUint(sourceStamp, 10))
		setJournal(c, JournalSrcGen, srcGen)
		c.Holder.Stamp = c.Gen + 1
		return nil
	}
}

// Revert is §7.4 reverting: the source takes execution back with its own
// evidence, before the flip. The target region's staging park is queued in
// the same write (§7.11: the revert's `opened` waits for it), owned by the
// target region and scoped to this move's own lane, moves/<host>/<op>, with
// the holder's incarnation and the writers' horizon the caller computed
// (G7). Nothing here may park the live location, the host-wide staging
// prefix or another move's lane. A park already owed under this op for the
// region is not queued twice.
//
// A replay after a crash finds the claim already reverting, with this op's
// source, region, rev_phase "evidence" and the same staging park: that is
// a no-op and allocates no ids.
func Revert(opID, sourceBox, targetRegion string, stagingScope *ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind == KindReverting {
			if c.Op.Exec == sourceBox && sourceBox != "" && len(c.Op.Participants) > 0 && c.Op.Participants[0] == sourceBox &&
				c.Op.Journal[JournalEvidenceSource] != "" && c.Op.Journal[JournalRevPhase] == RevPhaseEvidence &&
				targetRegion != "" && c.Op.Intent["to"] == targetRegion && stagingScope != nil {
				for _, o := range c.Obligations {
					if o.Op == opID && o.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, targetRegion) && o.Park != nil && sameParkScope(*o.Park, *stagingScope) {
						return nil
					}
				}
			}
			return fmt.Errorf("%w: the op is reverting, and not with this revert's source, region and staging park", ErrWrongState)
		}
		if c.Op.Kind != KindMoving || c.Op.Exec != "" || c.Op.Journal[JournalEvidenceSource] == "" {
			return fmt.Errorf("%w: revert needs a moving op with exec none and the source's evidence", ErrWrongState)
		}
		if len(c.Op.Participants) == 0 || sourceBox == "" || c.Op.Participants[0] != sourceBox {
			return fmt.Errorf("%w: %q is not the move's source", ErrWrongState, sourceBox)
		}
		if targetRegion == "" || c.Op.Intent["to"] != targetRegion {
			return fmt.Errorf("%w: %q is not the move's target region", ErrWrongState, targetRegion)
		}
		if stagingScope == nil || newID == nil || c.Holder == nil {
			return fmt.Errorf("%w: a revert names the staging lane it leaves behind", ErrMalformed)
		}
		if err := stagingScope.Validate(c.Host); err != nil {
			return err
		}
		if stagingScope.Location != "moves/"+c.Host+"/"+opID || stagingScope.Incarnation != c.Holder.Incarnation {
			return fmt.Errorf("%w: the revert's park must be this move's own lane and this holder's incarnation", ErrMalformed)
		}
		setJournal(c, JournalSourceStamp, strconv.FormatUint(c.Holder.Stamp, 10))
		setJournal(c, JournalSrcGen, c.Holder.Generation)
		c.Op.Kind, c.Op.Exec = KindReverting, sourceBox
		if c.Op.Journal[JournalRevPhase] == "" {
			setJournal(c, JournalRevPhase, RevPhaseEvidence)
		}
		for _, o := range c.Obligations {
			if o.Op == opID && o.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, targetRegion) {
				return nil
			}
		}
		s := *stagingScope
		c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-replica", Region: targetRegion, Retirement: newID(), Owner: "region:" + targetRegion, Park: &s})
		return nil
	}
}

func sameParkScope(a, b ParkScope) bool {
	return a.Location == b.Location && a.Incarnation == b.Incarnation && a.EvidenceAt.Equal(b.EvidenceAt) && a.RequestLimit == b.RequestLimit
}

// Park queue helpers for move retirements (moves spec §6.4, index D6). A
// move adds no obligation kind: a staging lane is a park-replica scoped to
// moves/<host>/<op>; a whole location retired where it is no longer
// authoritative is a park-replica scoped to instances/<host>[/<inc>]; a
// landing's earlier-stay residue is one park-generation per generation
// (RootGeneration for the origin); the box's own artifacts are a park-local.
// Claim.Validate accepts all of them unchanged, and the park engine archives
// them under parked/<incarnation>/<host>/<retirement>/ with the op in the
// manifest.
//
// Each helper changes obligations only, is idempotent per its key (a replay
// allocates no ids), and refuses a replay that names the same chore with a
// different scope rather than overwriting it. The scopes themselves come
// from internal/moves (SourceLocationScope, StagingScope, AbortScope,
// ReturnLaneScope), the one source for the box and the console; moves
// imports hostclaim, so they cannot be called from here.

// parkQueueGuard is the check every helper shares: the op is current, the
// claim has its holder, the scope is valid for this host and names this
// holder's incarnation.
func parkQueueGuard(c *Claim, opID, region string, scope ParkScope, newID func() string) (ParkScope, error) {
	if err := RequireOp(opID)(c); err != nil {
		return scope, err
	}
	if c.Holder == nil || newID == nil || region == "" {
		return scope, fmt.Errorf("%w: a move's park names its region and holder and has a way to mint ids", ErrMalformed)
	}
	scope.Location = strings.TrimSuffix(scope.Location, "/")
	if err := scope.Validate(c.Host); err != nil {
		return scope, err
	}
	if scope.Incarnation != c.Holder.Incarnation {
		return scope, fmt.Errorf("%w: the park's incarnation is not this holder's", ErrMalformed)
	}
	return scope, nil
}

// requireHolderLocation: location and residue parks name holder.location
// (§6.4), in whichever form the holder uses, not the other one.
func requireHolderLocation(c *Claim, scope ParkScope) error {
	if isStagingLocation(scope.Location) {
		return fmt.Errorf("%w: a location park is not a staging lane", ErrMalformed)
	}
	if scope.Location != strings.TrimSuffix(c.Holder.Location, "/") {
		return fmt.Errorf("%w: the park names %q, the holder's location is %q", ErrMalformed, scope.Location, c.Holder.Location)
	}
	return nil
}

func isStagingLocation(location string) bool { return strings.HasPrefix(location, "moves/") }

// queueReplicaPark appends a park-replica for the region unless this op
// already owes one of the same class (a lane or a location) there. It
// reports whether it appended.
func queueReplicaPark(c *Claim, opID, region string, scope ParkScope, newID func() string) (bool, error) {
	for _, o := range c.Obligations {
		if o.Op == opID && o.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, region) && o.Park != nil && isStagingLocation(o.Park.Location) == isStagingLocation(scope.Location) {
			if !sameParkScope(*o.Park, scope) {
				return false, fmt.Errorf("%w: %s already owes a park in %s with another scope", ErrWrongState, opID, region)
			}
			return false, nil
		}
	}
	s := scope
	c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-replica", Region: region, Retirement: newID(), Owner: "region:" + region, Park: &s})
	return true, nil
}

// queueLocalPark appends a park-local for the box unless this op has one.
func queueLocalPark(c *Claim, opID, box string, newID func() string) {
	for _, o := range c.Obligations {
		if o.Op == opID && o.Kind == "park-local" && o.Box == box {
			return
		}
	}
	c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-local", Box: box, Retirement: newID(), Owner: "box:" + box})
}

// PhaseCommitted is the landing after the commit: the source region's whole
// location is retired from here (§6.4, §7.9).
const PhaseCommitted = "committed"

// requireRetiredRegion is QueueLocationPark's lifecycle check. A whole
// location is parked only where the region has stopped being authoritative:
//   - the source's region (any region but the move's target) once the op is
//     landing/committed;
//   - the target's region once the op is restoring and the target has
//     released (released, or claimed by the source).
//
// The source before the commit and the target after it are live, and so is
// every other state, which is ErrWrongState. A box, when named, is the retiring
// side's box. The claim records the target region and both boxes at the start
// of the move (Intent), not the source's region, so a region that is neither
// is not told apart from the source here.
func requireRetiredRegion(c *Claim, region, box string) error {
	to := c.Op.Intent["to"]
	switch {
	case c.Op.Kind == KindLanding && c.Op.Phase == PhaseCommitted:
		if to != "" && region != to && (box == "" || box == c.Op.Intent["source"]) {
			return nil
		}
	case c.Op.Kind == KindRestoring && (c.Op.Phase == PhaseReleased || c.Op.Phase == PhaseClaimed):
		if to != "" && region == to && (box == "" || box == c.Op.Intent["target"]) {
			return nil
		}
	}
	return fmt.Errorf("%w: a location park in %s is not legal for %s/%q (the source's after the commit, a rolled-back target's after its release)", ErrWrongState, region, c.Op.Kind, c.Op.Phase)
}

// QueueLocationPark queues the park of a whole location in a region where it
// is no longer authoritative: the source's after the commit, a rolled-back
// target's at cleaning (requireRetiredRegion; any other state is
// ErrWrongState with nothing queued). The scope is holder.location, exactly.
// With a box, the box's park-local is queued in the same write. Idempotent
// per (op, region) and (op, box).
func QueueLocationPark(opID, region, box string, scope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		var err error
		if scope, err = parkQueueGuard(c, opID, region, scope, newID); err != nil {
			return err
		}
		if err := requireHolderLocation(c, scope); err != nil {
			return err
		}
		if err := requireRetiredRegion(c, region, box); err != nil {
			return err
		}
		if _, err := queueReplicaPark(c, opID, region, scope, newID); err != nil {
			return err
		}
		if box != "" {
			queueLocalPark(c, opID, box, newID)
		}
		return nil
	}
}

// QueueStagingPark queues the park of this move's own staging lane in a
// region. The scope's location must be exactly moves/<host>/<op>: not the
// live location, not the host-wide staging prefix, not another move's lane.
// A lane park a revert already queued under the op is the same chore.
// Idempotent per (op, region). EvidenceAt is the recorded writers' bound
// (G7); the caller computes it.
func QueueStagingPark(opID, region string, scope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		var err error
		if scope, err = parkQueueGuard(c, opID, region, scope, newID); err != nil {
			return err
		}
		// Restates moves.ScopeLocation(host, op); moves imports hostclaim, so
		// it cannot be called from here. Keep the two together.
		if scope.Location != "moves/"+c.Host+"/"+opID {
			return fmt.Errorf("%w: a staging park is this move's own lane, moves/%s/%s", ErrMalformed, c.Host, opID)
		}
		_, err = queueReplicaPark(c, opID, region, scope, newID)
		return err
	}
}

// QueueLocalPark queues a park-local for the box alone, for the landing's
// local phase: park.Local runs only against a persisted park-local chore of
// the claim and treats an unknown id as already completed, so the chore must
// be in the claim before the call. Idempotent per (op, box). Queue it only
// when InventoryHasStamp says the box has paths of the earlier stay (G5).
func QueueLocalPark(opID, box string, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if box == "" || newID == nil {
			return fmt.Errorf("%w: a local park names its box and has a way to mint ids", ErrMalformed)
		}
		queueLocalPark(c, opID, box, newID)
		return nil
	}
}

// InventoryHasStamp reports whether the box's inventory lists any path
// tagged with the stamp: the landing's earlier stay S' (G5). A first landing
// has none, and then no park-local is queued and none is retired.
func InventoryHasStamp(c *Claim, box string, stamp uint64) bool {
	if c == nil || stamp == 0 {
		return false
	}
	for _, s := range c.Inventory[box].Paths {
		if s == stamp {
			return true
		}
	}
	return false
}

// CompleteLocalPark is the replay of a local phase that finds the box's
// park-local chore and no paths of the earlier stamp left: there is nothing
// to retire, so the chore is completed under the op (CompleteObligation).
// With paths of that stamp still listed it refuses (ErrWrongState): the
// retirement has to run. No chore is a no-op.
func CompleteLocalPark(opID, box string, earlierStamp uint64) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if box == "" {
			return fmt.Errorf("%w: a local park names its box", ErrMalformed)
		}
		for _, o := range c.Obligations {
			if o.Op == opID && o.Kind == "park-local" && o.Box == box {
				if InventoryHasStamp(c, box, earlierStamp) {
					return fmt.Errorf("%w: %s still lists paths of stamp %d", ErrWrongState, box, earlierStamp)
				}
				return CompleteObligation(opID, o.ID)(c)
			}
		}
		return nil
	}
}

// Commit is the claim half of the record's commit (§7.5, §7.9 step 1): the
// landing goes to committed and, in the same write, the parks the commit owes
// are queued through the Plan 2 helpers: the source region's whole location
// (QueueLocationPark, which holds the lifecycle and exact-location checks),
// the source box's park-local, and the target region's staging lane
// (QueueStagingPark). The source's park-local is queued only when the source
// box lists paths of the journalled source stamp S (InventoryHasStamp, H5b);
// otherwise none is. CoverRegionPredecessors then covers the source region's
// admission records. holder.stamp, session and generation are not touched
// (R10-3, H3).
//
// It follows probed only (platform-1e m-2). Idempotent under the op: a
// landing already committed returns nil and queues nothing. A refused commit
// leaves the claim as it found it.
func Commit(opID, sourceRegion, sourceBox, targetRegion string, sourceScope, stagingScope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindLanding {
			return fmt.Errorf("%w: kind %q", ErrWrongState, c.Op.Kind)
		}
		if c.Op.Phase == PhaseCommitted {
			return nil
		}
		// The commit follows probed only (platform-1e m-2): with Group D's
		// adjacent-step rule (probing, probed, committed) the admitted
		// session survives it.
		if c.Op.Phase != PhaseLandingProbed {
			return fmt.Errorf("%w: phase %q", ErrWrongState, c.Op.Phase)
		}
		if len(c.Op.Participants) != 2 || c.Op.Exec != c.Op.Participants[1] {
			return fmt.Errorf("%w: exec %q is not the target", ErrWrongState, c.Op.Exec)
		}
		if sourceBox != c.Op.Participants[0] {
			return fmt.Errorf("%w: %q is not the move's source", ErrWrongState, sourceBox)
		}
		parkBox := ""
		if stamp, err := strconv.ParseUint(c.Op.Journal[JournalSourceStamp], 10, 64); err == nil && InventoryHasStamp(c, sourceBox, stamp) {
			parkBox = sourceBox
		}
		phase, obligations := c.Op.Phase, len(c.Obligations)
		c.Op.Phase = PhaseCommitted
		for _, step := range []func(*Claim) error{
			QueueLocationPark(opID, sourceRegion, parkBox, sourceScope, newID),
			QueueStagingPark(opID, targetRegion, stagingScope, newID),
			CoverRegionPredecessors(opID, PhaseCommitted, sourceRegion),
		} {
			if err := step(c); err != nil {
				c.Op.Phase, c.Obligations = phase, c.Obligations[:obligations]
				return err
			}
		}
		return nil
	}
}

// QueueResidueParks queues one park-generation per listed generation for a
// landing's earlier-stay residue in the target region: RootGeneration for
// the origin, the generation name for the rest, all scoped to the same
// location. Deduplicated by (op, region, generation); a generation already
// queued with another scope is refused, and nothing is queued then.
func QueueResidueParks(opID, region string, location ParkScope, generations []string, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		var err error
		if location, err = parkQueueGuard(c, opID, region, location, newID); err != nil {
			return err
		}
		if err := requireHolderLocation(c, location); err != nil {
			return err
		}
		for _, g := range generations {
			if g == "" {
				return fmt.Errorf("%w: a residue generation has a name", ErrMalformed)
			}
		}
		queued := map[string]bool{}
		for _, o := range c.Obligations {
			if o.Op == opID && o.Kind == "park-generation" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, region) {
				if o.Park == nil || !sameParkScope(*o.Park, location) {
					for _, g := range generations {
						if g == o.Generation {
							return fmt.Errorf("%w: %s already owes generation %q in %s with another scope", ErrWrongState, opID, g, region)
						}
					}
				}
				queued[o.Generation] = true
			}
		}
		for _, g := range generations {
			if queued[g] {
				continue
			}
			queued[g] = true
			s := location
			c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-generation", Region: region, Generation: g, Retirement: newID(), Owner: "region:" + region, Park: &s})
		}
		return nil
	}
}

// Recovery phases on the claim's op (spec §7.4, §7.8, §7.14; J1). The record
// carries the detailed recovery journal; the claim mirrors the phases the
// wake policy and the release key on, as the landing mirrors its own.
//
// restoring (a rollback, after the flip): the target's releasing and
// released, the source's claimed, then probing (written before the source's
// first recovery attempt), probed and opened.
//
// reverting (before the flip): op.phase stays the moving phase the revert
// began from ("" or "sealed") through rev_phase evidence and verify, then
// mirrors probing, probed and opened in the CAS that writes rev_phase.
const (
	PhaseReleasing = "releasing" // restoring: the target owes its stop evidence
	PhaseReleased  = "released"  // restoring: the target gave execution up
	PhaseClaimed   = "claimed"   // restoring: the source took execution back
	PhaseProbing   = "probing"   // restoring, reverting: the recovery probe
	PhaseProbed    = "probed"    // restoring, reverting: the probe passed
	PhaseOpened    = "opened"    // restoring, reverting: the source serves again

	RevPhaseEvidence = "evidence" // rev_phase: stop evidence before anything starts
	RevPhaseVerify   = "verify"   // rev_phase: files and timers against the stopped digests
)

// JournalTargetStamp is the target installation's stamp F, journalled at the
// rollback for its release (J15): the target's park-local is queued only
// when its inventory lists paths of F.
const JournalTargetStamp = "target_stamp"

// journalReturnLane marks the return lane's park as queued, so a replay after
// the park engine has completed it does not queue it again.
const journalReturnLane = "return_lane"

// RecoveryPhaseOrder is the op.phase order a recovery kind writes, or nil for
// any other kind. A revert's phases before probing are the moving phase it
// began from and are not listed.
func RecoveryPhaseOrder(kind string) []string {
	switch kind {
	case KindRestoring:
		return []string{PhaseReleasing, PhaseReleased, PhaseClaimed, PhaseProbing, PhaseProbed, PhaseOpened}
	case KindReverting:
		return []string{PhaseProbing, PhaseProbed, PhaseOpened}
	}
	return nil
}

// rollbackIDShape restates moves' rb_ op id shape (moves imports hostclaim, so
// it cannot be called from here). TestRollBackTakesMintedRollbackIDs holds
// the two together.
var rollbackIDShape = regexp.MustCompile(`^rb_[a-z2-7]{26}$`)

// inOneWrite runs steps in order as one transition: if any refuses, the claim
// is put back exactly as it was found.
func inOneWrite(c *Claim, steps ...func(*Claim) error) error {
	saved, err := cloneClaim(c)
	if err != nil {
		return err
	}
	for _, step := range steps {
		if err := step(c); err != nil {
			*c = *saved
			return err
		}
	}
	return nil
}

// RollBack is the claim half of the record's rollback CAS (§7.5 rolling_back,
// §7.8): landing, at any phase before the commit, to restoring/releasing
// under the same op id, with the rollback id in the intent. Execution stays
// with the target, which still owes its stop evidence; the holder is not
// touched. The target installation's stamp F (holder.stamp now) is
// journalled for TargetReleased (J15).
//
// The commit and the rollback race on the claim as on the record: a committed
// landing refuses the rollback, and Commit refuses a restoring op. A replay
// that finds this rollback already written (any restoring phase) is a no-op;
// another rollback id is refused.
func RollBack(opID, rbID string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if !rollbackIDShape.MatchString(rbID) {
			return fmt.Errorf("%w: %q is not a rollback id", ErrMalformed, rbID)
		}
		if c.Op.Kind == KindRestoring {
			if c.Op.Intent["rollback"] == rbID {
				return nil
			}
			return fmt.Errorf("%w: the op is restoring under rollback %q", ErrWrongState, c.Op.Intent["rollback"])
		}
		if c.Op.Kind != KindLanding || c.Op.Phase == PhaseCommitted || c.Holder == nil {
			return fmt.Errorf("%w: op is %s/%q, a rollback needs a landing before the commit", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		intent := make(map[string]string, len(c.Op.Intent)+1)
		for k, v := range c.Op.Intent {
			intent[k] = v
		}
		intent["rollback"] = rbID
		c.Op.Kind, c.Op.Phase, c.Op.Intent = KindRestoring, PhaseReleasing, intent
		setJournal(c, JournalTargetStamp, strconv.FormatUint(c.Holder.Stamp, 10))
		return nil
	}
}

// RollBackFlipped is the rollback's claim writer for a record already at
// landing (J16): when the claim is still moving/sealed (the flip's claim half
// not yet written, G12) it writes Flip first, then RollBack, in one write;
// over any other kind it is RollBack. A refused flip leaves the claim as found.
func RollBackFlipped(opID, rbID, targetBox string, sourceStamp uint64, srcGen string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if !rollbackIDShape.MatchString(rbID) {
			return fmt.Errorf("%w: %q is not a rollback id", ErrMalformed, rbID)
		}
		if c.Op.Kind != KindMoving {
			return RollBack(opID, rbID)(c)
		}
		return inOneWrite(c, Flip(opID, targetBox, sourceStamp, srcGen), RollBack(opID, rbID))
	}
}

// TargetReleased is the target's released CAS (§7.8): restoring/releasing,
// with the target executing, gives execution up citing its stop evidence
// (ExecToNone, evidence_target), moves to released and, in the same write,
// queues the park of the target region's whole location (QueueLocationPark,
// D6) and covers every admission record whose predecessor is in the target
// region (CoverRegionPredecessors, claim rev 6.14 item 6, R12-3). The
// target's park-local is queued only when its inventory lists paths of the
// stamp journalled at the rollback (J15); otherwise none is, and the
// target's cleaning parks its unregistered leftovers itself.
//
// A replay that finds the release written with the same evidence (released
// or any later phase) is a no-op and queues nothing, even after a chore has
// completed. A refused release leaves the claim as found.
const JournalTargetSession = "target_session"
const JournalTargetGeneration = "target_generation"

func TargetReleased(opID, targetBox, targetRegion, evidenceKey string, scope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindRestoring {
			return fmt.Errorf("%w: op is %s/%q, a release needs a rollback", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		if targetBox == "" || targetRegion == "" || evidenceKey == "" || newID == nil {
			return fmt.Errorf("%w: a release names the target, its region and its evidence", ErrMalformed)
		}
		if c.Op.Phase != PhaseReleasing {
			switch c.Op.Phase {
			case PhaseReleased, PhaseClaimed, PhaseProbing, PhaseProbed, PhaseOpened:
				if c.Op.Journal[JournalEvidenceTarget] == evidenceKey {
					return nil
				}
			}
			return fmt.Errorf("%w: op is %s/%q, not releasing with this evidence", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		if len(c.Op.Participants) != 2 || c.Op.Participants[1] != targetBox || c.Op.Intent["target"] != targetBox || c.Op.Exec != targetBox {
			return fmt.Errorf("%w: %q is not the executing target", ErrWrongState, targetBox)
		}
		if c.Op.Intent["to"] != targetRegion {
			return fmt.Errorf("%w: %q is not the move's target region", ErrWrongState, targetRegion)
		}
		parkBox := ""
		if stamp, err := strconv.ParseUint(c.Op.Journal[JournalTargetStamp], 10, 64); err == nil && InventoryHasStamp(c, targetBox, stamp) {
			parkBox = targetBox
		}
		return inOneWrite(c,
			func(c *Claim) error {
				if c.Holder != nil && c.Holder.WriterBox == targetBox {
					setJournal(c, JournalTargetSession, strconv.FormatUint(c.Holder.Session, 10))
					setJournal(c, JournalTargetGeneration, c.Holder.Generation)
				}
				return nil
			},
			ExecToNone(opID, targetBox, evidenceKey, JournalEvidenceTarget),
			func(c *Claim) error { c.Op.Phase = PhaseReleased; return nil },
			QueueLocationPark(opID, targetRegion, parkBox, scope, newID),
			CoverRegionPredecessors(opID, PhaseReleased, targetRegion),
		)
	}
}

// ReturnFence is a termination fence for a move's target that the caller
// verified before Update (VerifyReturnFence, J5): the op, the box and the
// exact fence.<box> journal entry it verified. The zero value vouches for
// nothing.
type ReturnFence struct{ op, box, ref, observedAt string }

// VerifyReturnFence re-reads and verifies the fence the op journals for box
// (VerifyOperationFence) and returns what ReturnToSource checks inside its
// CAS. Run it on the observation that Update will write over.
func VerifyReturnFence(ctx context.Context, s release.Store, authority FenceAuthority, op Op, box string) (ReturnFence, error) {
	if err := VerifyOperationFence(ctx, s, authority, op, box); err != nil {
		return ReturnFence{}, err
	}
	return ReturnFence{op: op.ID, box: box, ref: op.Journal["fence."+box]}, nil
}

// targetGaveUp is ReturnToSource's authority: the target's own stop evidence
// (evidence_target); or an operator's declaration under this op (a loss.<id>
// journal entry the claim records in loss_refs) that took the target out of
// the participants; or a verified termination fence whose journal entry the
// claim still carries.
func targetGaveUp(c *Claim, opID string, fence ReturnFence) bool {
	target := c.Op.Intent["target"]
	if target == "" {
		return false
	}
	if c.Op.Journal[JournalEvidenceTarget] != "" {
		return true
	}
	if !stringPresent(c.Op.Participants, target) {
		if _, consumed := ConsumedMoveTargetFence(c); consumed {
			return true
		}
		for _, id := range c.LossRefs {
			if _, ok := c.Op.Journal["loss."+id]; ok {
				return true
			}
		}
	}
	return fence.op == opID && fence.box == target && fence.ref != "" && c.Op.Journal["fence."+target] == fence.ref
}

// ReturnToSource is the source's claimed CAS (§7.8, K4): restoring/released
// with nobody executing ("" after the target's evidence, "none" after a
// declaration) and the target's authority given up (targetGaveUp) makes the
// source, the move's first participant, the executor again at claimed.
// holder.stamp is the source installation's stamp S the flip journalled; a
// clean return also names the source's generation (src_gen) again. The
// admissions are not touched (R11-4): the source's re-admission names the
// target's writer, their last record, as its predecessor.
//
// A declared-lost target is not released by TargetReleased: the resolve door
// steps the claim to released and queues the target region's park itself
// (J16), so this transition is the same for every form. A replay at claimed
// with the source executing is a no-op.
func ReturnToSource(opID, sourceBox string, clean bool, fence ReturnFence) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if sourceBox == "" {
			return fmt.Errorf("%w: a return names the source", ErrMalformed)
		}
		if c.Op.Kind != KindRestoring || c.Holder == nil {
			return fmt.Errorf("%w: op is %s/%q, a return needs a rollback", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		if c.Op.Phase == PhaseClaimed && c.Op.Exec == sourceBox {
			return nil
		}
		if c.Op.Phase != PhaseReleased || (c.Op.Exec != "" && c.Op.Exec != "none") {
			return fmt.Errorf("%w: op is %s/%q with exec %q, a return needs the target released", ErrWrongState, c.Op.Kind, c.Op.Phase, c.Op.Exec)
		}
		if len(c.Op.Participants) == 0 || c.Op.Participants[0] != sourceBox || c.Op.Intent["source"] != sourceBox {
			return fmt.Errorf("%w: %q is not the move's source", ErrWrongState, sourceBox)
		}
		if !targetGaveUp(c, opID, fence) {
			return fmt.Errorf("%w: no evidence, declaration or verified fence for the target", ErrWrongState)
		}
		stamp, err := strconv.ParseUint(c.Op.Journal[JournalSourceStamp], 10, 64)
		srcGen, flipped := c.Op.Journal[JournalSrcGen]
		if err != nil || stamp == 0 || !flipped {
			return fmt.Errorf("%w: the flip's source stamp and generation are not journalled", ErrWrongState)
		}
		c.Op.Exec, c.Op.Phase = sourceBox, PhaseClaimed
		c.Holder.Stamp = stamp
		if clean {
			c.Holder.Generation = srcGen
		}
		return nil
	}
}

// SetRestoringPhase steps a rollback's source recovery on the claim (J1):
// claimed, probing, probed, opened, one step at a time, with the source
// executing. releasing to released is TargetReleased's and released to
// claimed is ReturnToSource's. A replay finds the phase moved and is refused.
func SetRestoringPhase(opID, from, to string) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireMoveOp(opID, KindRestoring, from)(c); err != nil {
			return err
		}
		step := from + ">" + to
		if step != PhaseClaimed+">"+PhaseProbing && step != PhaseProbing+">"+PhaseProbed && step != PhaseProbed+">"+PhaseOpened {
			return fmt.Errorf("%w: restoring does not step %q to %q", ErrWrongState, from, to)
		}
		if len(c.Op.Participants) == 0 || c.Op.Exec != c.Op.Participants[0] {
			return fmt.Errorf("%w: exec %q is not the source", ErrWrongState, c.Op.Exec)
		}
		c.Op.Phase = to
		return nil
	}
}

// SetRevPhase steps a revert's journalled rev_phase (§7.4): evidence, verify,
// probing, probed, opened. verify needs the files baseline the record keeps
// from stopped (sourceBaseline); a revert from stopping has none and goes from
// evidence straight to probing (J8). From probing on, op.phase is written in
// the same CAS (J1); before it, op.phase is the moving phase the revert began
// from. The source must be executing. A replay finds the phase moved and is
// refused.
func SetRevPhase(opID, from, to string, sourceBaseline bool) func(*Claim) error {
	return func(c *Claim) error {
		if err := RequireOp(opID)(c); err != nil {
			return err
		}
		if c.Op.Kind != KindReverting || c.Op.Journal[JournalRevPhase] != from {
			return fmt.Errorf("%w: op is %s with rev_phase %q, the step expects reverting at %q", ErrWrongState, c.Op.Kind, c.Op.Journal[JournalRevPhase], from)
		}
		if len(c.Op.Participants) == 0 || c.Op.Exec != c.Op.Participants[0] {
			return fmt.Errorf("%w: exec %q is not the source", ErrWrongState, c.Op.Exec)
		}
		legal := false
		switch from + ">" + to {
		case RevPhaseEvidence + ">" + RevPhaseVerify:
			legal = sourceBaseline
		case RevPhaseEvidence + ">" + PhaseProbing:
			legal = !sourceBaseline
		case RevPhaseVerify + ">" + PhaseProbing, PhaseProbing + ">" + PhaseProbed, PhaseProbed + ">" + PhaseOpened:
			legal = true
		}
		if !legal {
			return fmt.Errorf("%w: rev_phase does not step %q to %q (baseline %v)", ErrWrongState, from, to, sourceBaseline)
		}
		mirrored := c.Op.Phase == from
		if from == RevPhaseEvidence || from == RevPhaseVerify {
			mirrored = c.Op.Phase == PhaseMoving || c.Op.Phase == PhaseSealed
		}
		if !mirrored {
			return fmt.Errorf("%w: op.phase %q disagrees with rev_phase %q", ErrWrongState, c.Op.Phase, from)
		}
		setJournal(c, JournalRevPhase, to)
		if to != RevPhaseVerify {
			c.Op.Phase = to
		}
		return nil
	}
}

// QueueReturnLanePark queues the park of a dirty rollback's return lane,
// moves/<host>/<rollback id>, in the source's region (§7.8 probed): the
// rollback's own lane, never the move's, the live location or the host-wide
// staging prefix. It is legal at restoring/probed, composed into the probed
// CAS. Idempotent: a chore already owed is not queued twice, and once the
// park engine has completed it a replay does not queue it again.
func QueueReturnLanePark(opID, sourceRegion string, scope ParkScope, newID func() string) func(*Claim) error {
	return func(c *Claim) error {
		var err error
		if scope, err = parkQueueGuard(c, opID, sourceRegion, scope, newID); err != nil {
			return err
		}
		rb := c.Op.Intent["rollback"]
		if c.Op.Kind != KindRestoring || c.Op.Phase != PhaseProbed || !rollbackIDShape.MatchString(rb) {
			return fmt.Errorf("%w: op is %s/%q, the return lane is parked at a rollback's probed", ErrWrongState, c.Op.Kind, c.Op.Phase)
		}
		if sourceRegion == c.Op.Intent["to"] {
			return fmt.Errorf("%w: the return lane is in the source's region, not %s", ErrWrongState, sourceRegion)
		}
		if scope.Location != "moves/"+c.Host+"/"+rb {
			return fmt.Errorf("%w: the return lane is this rollback's own lane, moves/%s/%s", ErrMalformed, c.Host, rb)
		}
		owed := false
		for _, o := range c.Obligations {
			if o.Op == opID && o.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], o.Region, sourceRegion) && o.Park != nil && isStagingLocation(o.Park.Location) {
				owed = true
			}
		}
		if !owed && c.Op.Journal[journalReturnLane] == scope.Location {
			return nil
		}
		if _, err := queueReplicaPark(c, opID, sourceRegion, scope, newID); err != nil {
			return err
		}
		setJournal(c, journalReturnLane, scope.Location)
		return nil
	}
}