package host

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// The source's finalization after the commit (moves spec §7.9 step 3; Plan 3
// Task 8 with rulings H5, H7, H19 and H24), behind CARLOS_MOVES_PHASE1. Once
// the record is committed the target's region is authoritative, and the
// source box has four things left to do for the move, in this order:
//
//  1. the remote half: the source region's location park (the park-replica
//     the commit queued for instances/<host>, D6, H24) is run through the
//     park engine over this box's own bucket until its archive is complete
//     (park.Engine.Parked). Its acknowledgement after the horizon is the
//     engine's or the regional runner's (C4), never a reason to wait here.
//     The horizon starts at the flip (H7), which is after the source's last
//     write to its location: a park whose EvidenceAt is earlier is refused;
//  2. the local half (ruling H5(b), Paul): a source whose installation is
//     registered (the commit queued its park-local, because the inventory
//     lists paths of the journalled source stamp S) retires it through
//     park.Local.RetireInstallation under the host guard, which completes the
//     chore itself (H19). Everything else the source kept at the live flat
//     paths, registered or not, is parked by a moves-owned journalled rename
//     into the engine's layout, .parked/<incarnation>/<host>/<retirement>/
//     artifacts/ (parkLocalArtifacts): the retirement id and every artifact's
//     identity are journalled before the first rename, and nothing the plan
//     did not record is ever moved or removed. The plan and the renames run
//     inside the host guard with the claim and the record read fresh there
//     (parkLocal; Plan 5 Task 3, carried follow-up (d)), so a call that
//     resumes after authority moved on renames nothing;
//  3. the row and its timers rows go in one registry transaction, only while
//     the move holds the row (registry.DeleteIfHoldWithTimers, H5(d)), under
//     provision.WithHostHoldLock with the record re-read there; the journal
//     says row-deleted "pending" before it and "done" after it, and a replay
//     that finds "pending" and no row of this op's treats the delete as done
//     (round-2 finding 12). The box's config ack is copied into the move's
//     report prefix as config-ack-source.json and deleted (§6.3);
//  4. MoveSourceParked stamps source_parked_at, and the serial is nudged.
//     Last of all the op's own journal directory is renamed into the same
//     retirement, so a finished source keeps nothing under .moves/. Then the
//     source tries the close (closeMove, movesparks.go): it closes only when
//     the target has opened and both parks are acknowledged already, and
//     this is the last pass that routes the record here, so otherwise the
//     close is the target's.
//
// Routing (ruling H5(a)): after the flip the record's region is the target's
// and after step 3 this box has no row, so the region-and-row rule that
// admits every other Phase 1 round would never bring the replay of a crash
// between steps 3 and 4 back here. movesParticipantRecord is the rule for the
// states after the flip: this box's region is the move's From (or To), and it
// still has the row or the op's journal. Plan 4's rollback visits reuse it.
//
// Arming. Every hook that exists on main is wired (the guard, park.Local, the
// registry's delete, the versioned store). The claim's park leases, view and
// authorizer come from its MoveParks (armMovesParks, Plan 5 Task 9a), and
// park.Local's authorizer is its MoveParks local authorizer (M9);
// armMovesParks sets them only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b), so sourceFinalArmed answers
// false in production and the pass only logs until then. The tests arm it
// with fakes.

// The finalization's op journal steps (moves.Journal, the source's own
// journal of this op, which the cut began).
const (
	finalStepRemote       = "source-park-remote"       // the location park's archive is complete, or acknowledged
	finalStepInstallation = "source-park-installation" // retired, completed or unregistered
	finalStepLocal        = "source-park-local"        // parkLocalArtifacts' step: the retirement id
	finalStepRowDeleted   = "row-deleted"              // "pending" before the delete, "done" after it
	finalStepConfigAck    = "config-ack-source"        // the config ack is copied and deleted
	finalStepParked       = "source-parked"            // "pending" before the stamp's CAS, "done" after it
	finalStepJournal      = "source-park-journal"      // where the journal is going, written just before it goes
)

// finalConfigAckName is the config ack's copy under the move's report prefix.
const finalConfigAckName = "config-ack-source.json"

// movesParticipantStates are the record states after the flip (and a
// revert or an abort) in which a participant may have work for a record
// whose region no longer names it, or whose row it has already deleted.
var movesParticipantStates = map[string]bool{
	provision.MoveCommitted:   true,
	provision.MoveRollingBack: true,
	provision.MoveReverting:   true,
	provision.MoveAborting:    true,
}

// movesParticipantRecord is the routing rule for the states after the flip
// (ruling H5(a), shared with Plan 4's J10 and J13): it answers this box's role
// in rec's move, roleSource when this box's region is the move's From,
// roleTarget when it is the move's To, and roleNone otherwise. A role is
// answered only while the box still has something of the move's: its row
// (OwnsRoute) or the op's journal directory (moveJournalPresent). The caller
// decides what a role does in a state; this only says the box is a
// participant with work it can find.
func movesParticipantRecord(cfg moveCfg, rec provision.Record) string {
	m := rec.Move
	if m == nil || m.Legacy() || !movesParticipantStates[m.State] || m.From == m.To {
		return roleNone
	}
	role := roleNone
	switch cfg.Region {
	case m.From:
		role = roleSource
	case m.To:
		role = roleTarget
	default:
		return roleNone
	}
	if cfg.OwnsRoute == nil || cfg.OwnsRoute(rec.Host) || moveJournalPresent(cfg.DataDir, rec.Host, m.ID) {
		return role
	}
	// A revert or an abort never flipped: its target has neither the row
	// nor a journal, and its one piece of work is the lane park the claim
	// owes its region (laneParkTarget; Plan 4 Task 6, Plan 5 Task 1).
	if role == roleTarget && (m.State == provision.MoveReverting || m.State == provision.MoveAborting) {
		return role
	}
	return roleNone
}

// moveJournalPresent reports whether this box holds the op's journal
// directory, <dataDir>/.moves/<host>/<op>/. A name that could leave its
// directory is never present.
func moveJournalPresent(dataDir, host, op string) bool {
	if dataDir == "" || !journalComponent(host) || !journalComponent(op) {
		return false
	}
	fi, err := os.Lstat(moves.JournalDir(dataDir, host, op))
	return err == nil && fi.IsDir()
}

// participantRound is a post-flip participant's step for a record the
// region rule did not admit (movesPass): the source's finalization of a
// committed move, the target's half of a rollback (movesrollback.go, J10),
// and the target region's lane park of a revert or an abort
// (laneParkTarget). The rest is Plan 4's later tasks'.
func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec provision.Record, role string) error {
	switch {
	case role == roleSource && rec.Move.State == provision.MoveCommitted:
		return w.finalizeSource(ctx, s, rec)
	case role == roleTarget && rec.Move.State == provision.MoveCommitted:
		return w.committedTarget(ctx, s, rec)
	case role == roleTarget && rec.Move.State == provision.MoveRollingBack:
		return w.rollbackTarget(ctx, s, rec)
	case role == roleTarget && rec.Move.State == provision.MoveReverting:
		return w.laneParkTarget(ctx, s, rec, hostclaim.KindReverting)
	case role == roleTarget && rec.Move.State == provision.MoveAborting:
		return w.laneParkTarget(ctx, s, rec, hostclaim.KindMoving, hostclaim.PhaseAborting)
	}
	return nil
}

// sourceFinalArmed reports whether this box can run the finalization: every
// hook it calls is wired, the claim's three park hooks and its MoveParks
// local authorizer (M9) included. Nothing stands in for a missing owner:
// the tests inject their own. In production armMovesParks sets them only
// once the wake-source gate opens (movesWakeSourceRegistered, Plan 5 Task
// 9b), so this answers false there until then.
func sourceFinalArmed(cfg moveCfg) bool {
	return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" &&
		parkHooksArmed(cfg) && cfg.ParkLocalAuthorize != nil &&
		cfg.Versioned != nil && cfg.Route != nil && cfg.DeleteHeld != nil
}

// sourceLocationParkScope is the source region's location park the commit
// queues (Task 7 builds the commit; Task 8 reads it): holder.location, the
// holder's incarnation, the cut's request limit, and a horizon that starts at
// the flip (ruling H7). The flip comes after ExecToNone, after which the
// source never writes its location again; the stop evidence's CheckedAt does
// not, because the final checkpoint writes after it.
func sourceLocationParkScope(rec provision.Record, holder hostclaim.Holder) (hostclaim.ParkScope, error) {
	if rec.Move == nil || rec.Move.FlippedAt == nil {
		return hostclaim.ParkScope{}, errors.New("the source's location park needs the flip's time")
	}
	return moves.SourceLocationScope(holder, *rec.Move.FlippedAt), nil
}

// errFinalChanged is the record or claim no longer being this move's commit
// when a step re-reads it.
var errFinalChanged = errors.New("the record is no longer this move's commit")

// sourceFinal is one pass's state for one committed record on its source.
type sourceFinal struct {
	w   *movesWorker
	s   release.Store
	rec provision.Record
	g   provision.MoveGuard
	j   *moves.Journal
}

// finalizeSource is the source's §7.9 step 3 for a committed record. It runs
// only when the record is committed and either this box's row is held by the
// move or the journal has row-deleted (Review Focus 5, round-2 finding 12):
// a landing record, a row held by another op, or no row and no journalled
// delete leaves everything as it is.
func (w *movesWorker) finalizeSource(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveCommitted || m.From != cfg.Region || m.To == cfg.Region {
		return nil
	}
	if !sourceFinalArmed(cfg) {
		w.notice(rec.Host+"#final", "moves: %s: the source's finalization of %s waits: this box's finalization is not armed (the claim's park hooks are set from MoveParks only once this box's moves wake source is registered, Plan 5 Task 9b)", rec.Host, m.ID)
		return nil
	}
	f := &sourceFinal{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}}
	return f.run(ctx)
}

func (f *sourceFinal) cfg() moveCfg { return f.w.cfg }

func (f *sourceFinal) now() time.Time {
	if n := f.cfg().Now; n != nil {
		return n()
	}
	return time.Now()
}

// fault is the test seam after a side effect.
func (f *sourceFinal) fault(point string) error {
	if fn := f.cfg().finalFault; fn != nil {
		return fn(point)
	}
	return nil
}

// mark journals a step durably, then passes its fault point.
func (f *sourceFinal) mark(step, value string) error {
	f.j.Mark(step, value)
	if err := f.j.Commit(f.cfg().DataDir); err != nil {
		return err
	}
	return f.fault("after-" + step)
}

func (f *sourceFinal) run(ctx context.Context) error {
	cfg := f.cfg()
	host, id := f.rec.Host, f.g.ID
	j, err := moves.ReadJournal(cfg.DataDir, host, id)
	if err != nil {
		return err
	}
	f.j = j
	rt, err := cfg.Route(host)
	if err != nil {
		return fmt.Errorf("read this box's row: %w", err)
	}
	held := rt != nil && rt.HoldOp == id
	if !held && !j.Done(finalStepRowDeleted) {
		f.w.notice(host+"#final-row", "moves: %s: the source's finalization of %s waits: this box's row is not held by the move", host, id)
		return nil
	}
	if held && rt.DBPath != "" && rt.DBPath != provision.DefaultDBPath(cfg.DataDir, host) {
		return fmt.Errorf("move %s: the source's row is not at the flat path (%s)", id, rt.DBPath)
	}
	// source_parked_at is stamped only after every step above it, so once it
	// is only the journal's own park is left, and the claim may already
	// have moved on to the close.
	if f.rec.Move.SourceParkedAt != nil {
		if j.Steps[finalStepParked] != "done" {
			if err := f.mark(finalStepParked, "done"); err != nil {
				return err
			}
		}
		return f.parkJournalAndClose(ctx)
	}
	obs, err := hostclaim.Read(ctx, f.s, host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if err := f.sourceOf(obs); err != nil {
		f.w.notice(host+"#final-claim", "moves: %s: the source's finalization of %s waits: %v", host, id, err)
		return nil
	}
	holder := *obs.Claim.Holder
	if !j.Done(finalStepRemote) {
		done, err := f.remote(ctx, obs)
		if err != nil || !done {
			return err
		}
		if err := f.mark(finalStepRemote, ""); err != nil {
			return err
		}
	}
	if !j.Done(finalStepInstallation) {
		how, err := f.installation(ctx)
		if err != nil {
			return err
		}
		if err := f.mark(finalStepInstallation, how); err != nil {
			return err
		}
	}
	if err := f.parkLocal(ctx, holder, obs.Claim); err != nil {
		return fmt.Errorf("park the source's local artifacts: %w", err)
	}
	if err := f.deleteRow(ctx); err != nil {
		return err
	}
	if !j.Done(finalStepConfigAck) {
		if err := f.configAck(ctx); err != nil {
			return err
		}
		if err := f.mark(finalStepConfigAck, ""); err != nil {
			return err
		}
	}
	if j.Steps[finalStepParked] != "done" {
		// "pending" is the stamp step reached: from here the journal is
		// the local sweep's to finish if the close overtakes this box
		// (sweepFinalJournals).
		if j.Steps[finalStepParked] != "pending" {
			j.Mark(finalStepParked, "pending")
			if err := j.Commit(cfg.DataDir); err != nil {
				return err
			}
			if err := f.fault("after-source-parked-pending"); err != nil {
				return err
			}
		}
		if err := f.stampParked(ctx); err != nil {
			return err
		}
		if err := f.mark(finalStepParked, "done"); err != nil {
			return err
		}
	}
	return f.parkJournalAndClose(ctx)
}

// parkJournalAndClose is step 4's end: the journal's park, then the close
// when the target has opened and nothing is owed (closeMove).
func (f *sourceFinal) parkJournalAndClose(ctx context.Context) error {
	if err := f.parkJournal(); err != nil {
		return err
	}
	return f.w.closeMove(ctx, f.s, f.rec, f.fault)
}

// sourceOf is the claim this finalization runs under: this move's landing,
// committed, this box its source with its evidence journalled, and the
// record's incarnation.
func (f *sourceFinal) sourceOf(obs hostclaim.Observed) error {
	cfg := f.cfg()
	if obs.Absent || obs.Claim == nil {
		return fmt.Errorf("%w: no host claim", park.ErrUnsafe)
	}
	c := obs.Claim
	if c.Op.ID != f.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted {
		return fmt.Errorf("%w: the claim is %s/%q for %q, not this move's commit", park.ErrUnsafe, c.Op.Kind, c.Op.Phase, c.Op.ID)
	}
	if len(c.Op.Participants) != 2 || c.Op.Participants[0] != cfg.Box || c.Op.Intent["source"] != cfg.Box {
		return fmt.Errorf("%w: this box is not the move's source in the claim", park.ErrUnsafe)
	}
	if c.Op.Journal[hostclaim.JournalEvidenceSource] == "" {
		return fmt.Errorf("%w: the claim does not journal the source's evidence", park.ErrUnsafe)
	}
	if c.Holder == nil || c.Holder.Incarnation != f.g.Incarnation {
		return fmt.Errorf("%w: the claim's holder is not this move's incarnation", park.ErrUnsafe)
	}
	return nil
}

// engine is the park engine over this box's own bucket, with the claim's
// leases and view, and the claim's authorizer behind the finalization's own
// check (sourceFinalArmed requires all three): the claim is still this
// move's commit with this box its source, and the chore is this op's in
// this region.
func (f *sourceFinal) engine(vs lease.VersionedStore) park.Engine {
	cfg := f.cfg()
	own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
		if err := f.sourceOf(obs); err != nil {
			return err
		}
		if o.Op != f.g.ID || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) {
			return fmt.Errorf("%w: not this finalization's chore", park.ErrUnsafe)
		}
		return nil
	}
	return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: f.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases,
		Now: f.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault}
}

// remote is step 1. done is true once the source region's location park has
// its archive complete, or is no longer in the claim (acknowledged: the
// commit queues it in its own write, so a committed claim without it has
// had it acknowledged). The engine takes the location's lease itself.
func (f *sourceFinal) remote(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := f.cfg()
	host, id := f.rec.Host, f.g.ID
	var chore *hostclaim.Obligation
	for i := range obs.Claim.Obligations {
		o := &obs.Claim.Obligations[i]
		if o.Op == id && o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) && o.Park != nil && !isStagingLocation(o.Park.Location) {
			chore = o
		}
	}
	if chore == nil {
		return true, nil
	}
	if flipped := f.rec.Move.FlippedAt; flipped == nil || chore.Park.EvidenceAt.Before(*flipped) {
		return false, fmt.Errorf("move %s: the source's location park starts its horizon at %s, before the flip (ruling H7)", id, chore.Park.EvidenceAt.Format(time.RFC3339))
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return false, err
	}
	e := f.engine(vs)
	work := park.Work{Host: host, ID: chore.ID, Op: id}
	if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) {
		return false, fmt.Errorf("park the source's location: %w", err)
	}
	parked, err := e.Parked(ctx, work)
	if errors.Is(err, park.ErrChanged) {
		return true, nil
	}
	if err != nil {
		return false, fmt.Errorf("the source's location park: %w", err)
	}
	if !parked {
		f.w.notice(host+"#final-remote", "moves: %s: the source's finalization of %s waits for its location park's archive", host, id)
		return false, nil
	}
	log.Printf("moves: %s: the source's location %s is parked", host, chore.Park.Location)
	return true, nil
}

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

// installation is step 2's registered half (ruling H5(b)). With the box's
// park-local and inventory paths of the journalled source stamp S, the
// installation is retired under the host guard, which completes the chore; a
// park-local with no such paths left is completed; no park-local is an
// unregistered source, whose artifacts are all parkLocalArtifacts'. It
// answers which, for the journal.
func (f *sourceFinal) installation(ctx context.Context) (string, error) {
	cfg := f.cfg()
	host, id := f.rec.Host, f.g.ID
	obs, err := hostclaim.Read(ctx, f.s, host)
	if err != nil {
		return "", fmt.Errorf("read the host claim: %w", err)
	}
	if err := f.sourceOf(obs); err != nil {
		return "", fmt.Errorf("%w: %v", errFinalChanged, err)
	}
	c := obs.Claim
	stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
	if err != nil || stamp == 0 {
		return "", fmt.Errorf("move %s: the claim journals no source stamp", id)
	}
	var chore *hostclaim.Obligation
	for i := range c.Obligations {
		if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
			chore = o
		}
	}
	registered := hostclaim.InventoryHasStamp(c, cfg.Box, stamp)
	switch {
	case chore != nil && registered:
		captured, err := hostguard.Capture(cfg.DataDir, host)
		if err != nil {
			return "", fmt.Errorf("the source's installation record: %w", err)
		}
		// A retry after the guard retired the record and before the claim
		// completed the chore captures stamp 0; the stamp is the claim's S.
		if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != stamp) {
			return "", fmt.Errorf("move %s: the source's installation record is at %s/%d, the claim's source is %s/%d", id, captured.Incarnation, captured.Stamp, c.Holder.Incarnation, stamp)
		}
		guard := cfg.Guard
		if guard.Claims == nil {
			guard.Claims = hostguard.StoreReader{Store: f.s}
		}
		local := cfg.Local
		local.Guard, local.Claims = guard, f.s
		if local.Box == "" {
			local.Box = cfg.Box
		}
		if local.Authorize == nil {
			// park.Local asks the claim's MoveParks local authorizer (M9),
			// behind the finalization's own check over the claim it read
			// under the guard. The claim's engine authorizer refuses a
			// park-local, and nothing stands in: sourceFinalArmed requires it.
			local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, func(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
				if err := f.sourceOf(hostclaim.Observed{Host: host, Claim: c}); err != nil {
					return err
				}
				if o.Op != id || o.Kind != "park-local" || o.Box != cfg.Box {
					return fmt.Errorf("%w: not this finalization's local chore", park.ErrUnsafe)
				}
				return nil
			})
		}
		if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: stamp}, chore.ID, id); err != nil {
			return "", fmt.Errorf("retire the source's installation: %w", err)
		}
		log.Printf("moves: %s: the source's registered installation (stamp %d) is retired", host, stamp)
		return "retired", nil
	case chore != nil:
		if _, err := hostclaim.Update(ctx, f.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, stamp)); err != nil {
			return "", fmt.Errorf("complete the source's local park: %w", err)
		}
		return "completed", nil
	case registered:
		return "", fmt.Errorf("move %s: this box's inventory lists the source's paths, and the commit queued no park-local for them", id)
	}
	return "unregistered", nil
}

// parkLocal is step 2's moves-owned rename (parkLocalArtifacts) of what the
// source kept at the flat paths, its plan and its renames inside the host
// guard (Plan 5 Task 3, carried follow-up (d); the pattern of Plan 4's
// installer, installRun.guardedWith, and its rollback's parkLeftovers). In
// the guard, with the claim and the record read fresh:
//   - the claim is still this move's commit with this box its source
//     (sourceOf), and this box does not execute the host;
//   - no inventory entry of this box claims a path the rename would move;
//   - the record is still this move's commit.
//
// The guard is the installation record's state: none (an unregistered
// source with no record, WithNoInstallation), or the source's stamp, active
// or retired by step 2's registered half (RetireInstallationWith, which
// retires an active one first, so the guard refuses this box any later
// write at it). A retirement already journalled is done and takes no guard.
func (f *sourceFinal) parkLocal(ctx context.Context, holder hostclaim.Holder, c *hostclaim.Claim) error {
	j := f.j
	if _, done := j.Steps[finalStepLocal]; done {
		return nil
	}
	cfg := f.cfg()
	host, id := f.rec.Host, f.g.ID
	paths := sourceArtifactPaths(host)
	check := func(fresh *hostclaim.Claim) error {
		if err := f.sourceOf(hostclaim.Observed{Host: host, Claim: fresh}); err != nil {
			return fmt.Errorf("%w: %v", errFinalChanged, err)
		}
		if fresh.Op.Exec == cfg.Box {
			return fmt.Errorf("%w: this box executes the host again", errFinalChanged)
		}
		for _, p := range paths {
			if _, ok := fresh.Inventory[cfg.Box].Paths[p]; ok {
				return fmt.Errorf("%w: the inventory claims %s", errFinalChanged, p)
			}
		}
		rec, err := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
		if err != nil {
			return fmt.Errorf("re-read the record: %w", err)
		}
		if rec == nil || rec.Move == nil || rec.Move.ID != id || rec.Move.Incarnation != f.g.Incarnation || rec.Move.State != provision.MoveCommitted {
			return errFinalChanged
		}
		return nil
	}
	work := func() error {
		_, err := parkLocalArtifacts(ctx, j, f.mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: holder.Incarnation,
			Host: host, Op: id, Step: finalStepLocal, Paths: paths, NewID: obligationID, Fault: f.fault})
		return err
	}
	guard := cfg.Guard
	if guard.Claims == nil {
		guard.Claims = hostguard.StoreReader{Store: f.s}
	}
	// The guard's record directory: a Phase 1 source without host claims
	// may not have one yet (the landing makes the same two, ensureLocalDirs).
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(guard.DataDir, d), 0o770); err != nil {
			return err
		}
	}
	captured, err := guard.Capture(host)
	switch {
	case errors.Is(err, hostguard.ErrNoRecord):
		return guard.WithNoInstallation(ctx, host, f.g.Incarnation, check, work)
	case err != nil:
		return fmt.Errorf("the source's installation record: %w", err)
	}
	stamp := captured.Stamp
	if stamp == 0 {
		if stamp, err = strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64); err != nil || stamp == 0 {
			return fmt.Errorf("move %s: the claim journals no source stamp", id)
		}
	}
	return guard.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: f.g.Incarnation, Stamp: stamp}, check, work)
}

// sourceArtifactPaths are the source's artifacts at the flat paths (G11),
// relative to the data directory: the installation's (installArtifactPaths:
// the database with its marker, journal, quarantine and litestream shadow,
// the files directory and <host>.env), then the rendered config and the
// sidecar's env. The order is the journalled plan's and does not change.
func sourceArtifactPaths(host string) []string {
	return append(installArtifactPaths(host), host+".config.env", host+"-sidecar.env")
}

// deleteRow is step 3's row and timers delete, under the host's hold lock
// with the record re-read there (the lock reconcile's hold decision holds).
func (f *sourceFinal) deleteRow(ctx context.Context) error {
	if f.j.Steps[finalStepRowDeleted] == "done" {
		return nil
	}
	cfg := f.cfg()
	host, id := f.rec.Host, f.g.ID
	var err error
	provision.WithHostHoldLock(host, func() {
		fresh, e := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
		if e != nil {
			err = fmt.Errorf("re-read the record: %w", e)
			return
		}
		if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != f.g.Incarnation || m.State != provision.MoveCommitted {
			err = errFinalChanged
			return
		}
		if f.j.Steps[finalStepRowDeleted] != "pending" {
			f.j.Mark(finalStepRowDeleted, "pending")
			if err = f.j.Commit(cfg.DataDir); err != nil {
				return
			}
			if err = f.fault("after-row-pending"); err != nil {
				return
			}
		}
		// false: no row is held by this op. With "pending" journalled that is
		// this delete, already done; a row another op holds was never this
		// move's, and is left as it is.
		deleted, e := cfg.DeleteHeld(host, id)
		if e != nil {
			err = fmt.Errorf("delete the source's row: %w", e)
			return
		}
		if deleted {
			log.Printf("moves: %s: the source's row and its timers are deleted", host)
		}
		if err = f.fault("after-row-delete"); err != nil {
			return
		}
		err = f.mark(finalStepRowDeleted, "done")
	})
	return err
}

// configAck copies this box's config ack for the host into the move's report
// prefix and deletes it (§6.3). Absent is nothing to copy, or a replay after
// the delete.
func (f *sourceFinal) configAck(ctx context.Context) error {
	cfg := f.cfg()
	key := appconfig.ConfigAckKey(cfg.Box, f.rec.Host)
	body, _, err := f.s.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("read the config ack: %w", err)
	}
	// The copy is create-only. A replay after the copy finds it: the same
	// body is this copy; another (the ack was rewritten in between) is
	// replaced under its etag, so the copy is the last ack the box wrote.
	dst := moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.g.ID) + finalConfigAckName
	if _, err := f.s.Put(ctx, dst, body, ""); errors.Is(err, release.ErrCASConflict) {
		had, etag, gerr := f.s.Get(ctx, dst)
		if gerr != nil {
			return fmt.Errorf("read the config ack's copy: %w", gerr)
		}
		if string(had) != string(body) {
			if _, err := f.s.Put(ctx, dst, body, etag); err != nil {
				return fmt.Errorf("copy the config ack: %w", err)
			}
		}
	} else if err != nil {
		return fmt.Errorf("copy the config ack: %w", err)
	}
	if err := f.fault("after-config-ack-copy"); err != nil {
		return err
	}
	if err := f.s.Delete(ctx, key); err != nil {
		return fmt.Errorf("delete the config ack: %w", err)
	}
	return nil
}

// stampParked is step 4: MoveSourceParked, then the nudge. A record that has
// moved past this commit (closed, or another move) had it stamped, since the
// close needs it; this commit without the stamp is an error.
func (f *sourceFinal) stampParked(ctx context.Context) error {
	host := f.rec.Host
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, f.s, f.rec.Account, f.rec.App, host, observedMove(&pre, provision.MoveSourceParked(f.g, f.now())))
	if err == nil {
		f.w.logTransition(ctx, f.s, pre, *after)
	}
	if errors.Is(err, provision.ErrMoveConflict) {
		fresh, gerr := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
		if gerr != nil {
			return fmt.Errorf("re-read the record: %w", gerr)
		}
		if m := fresh.Move; m != nil && m.ID == f.g.ID && m.SourceParkedAt == nil {
			return fmt.Errorf("stamp source_parked_at: %w", err)
		}
		err = nil
	}
	if err != nil {
		return fmt.Errorf("stamp source_parked_at: %w", err)
	}
	if err := f.fault("after-source-parked-record"); err != nil {
		return err
	}
	nudgeSerial(ctx, f.s)
	log.Printf("moves: %s: the source has parked what it kept for %s", host, f.g.ID)
	return nil
}

// parkJournal renames the op's journal directory into the local retirement,
// after source_parked_at is stamped: until then it is what brings a replay
// back here. A crash before the rename leaves the journal, and the next pass
// (or the local sweep) renames it; after it nothing of the move is left
// under .moves/.
func (f *sourceFinal) parkJournal() error {
	return parkFinalJournal(f.cfg().DataDir, f.rec.Host, f.g.ID, f.j, f.mark)
}

// parkFinalJournal parks a finalized op's journal into the retirement its
// own local park recorded, journalling the destination just before it goes.
func parkFinalJournal(dataDir, host, op string, j *moves.Journal, mark func(step, value string) error) error {
	retirement := j.Steps[finalStepLocal]
	if retirement == "" {
		return fmt.Errorf("move %s: the journal records no local retirement for its own park", op)
	}
	plan, err := journalledLocalPlan(j, finalStepLocal)
	if err != nil {
		return err
	}
	dst := filepath.Join(parkedArtifactsDir(dataDir, plan.Incarnation, host, retirement), ".moves", host, op)
	if j.Steps[finalStepJournal] != dst {
		if err := mark(finalStepJournal, dst); err != nil {
			return err
		}
	}
	if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil {
		return fmt.Errorf("park the op journal: %w", err)
	}
	log.Printf("moves: %s: the source's finalization of %s is complete", host, op)
	return nil
}

// sweepFinalJournals is the local sweep of this box's op journals (fix round
// 1, ruling 1; a rollback target's journals too, sweepRollbackJournal),
// independent of any active move: a crash after the stamp and
// before the journal's park, then the close, leaves a journal no record
// routes back here (movesParticipantRecord answers nothing for a record with
// no move). Each journal under <DataDir>/.moves/<host>/<op>/ that has reached
// the stamp step is finished here:
//   - the stamp journalled "done": the journal is parked;
//   - "pending", and the listed record still carries this op: with
//     source_parked_at set it is journalled done and parked; without, it is
//     left to the finalization's own pass;
//   - "pending", and the listed record no longer carries this op (closed, or
//     another move since): the stamp landed, because the only transition
//     that clears a committed move is MoveClose, which refuses unless
//     source_parked_at is set (the aborts end at cutting/requested); so the
//     stamp is journalled done and the journal parked;
//   - a host the listing does not name (its record deleted, or unreadable)
//     is left as it is.
//
// A journal that has not reached the stamp step (no source-parked step: a
// landing's, a cut's, a finalization still under way) is never touched, and
// nothing is ever deleted: a journal is only renamed into its retirement.
func (w *movesWorker) sweepFinalJournals(ctx context.Context, s release.Store, recs []provision.Record) {
	dd := w.cfg.DataDir
	if dd == "" {
		return
	}
	hosts, err := os.ReadDir(filepath.Join(dd, ".moves"))
	if err != nil {
		if !errors.Is(err, os.ErrNotExist) {
			w.notice("#final-sweep", "moves: the op journal sweep cannot list %s: %v", filepath.Join(dd, ".moves"), err)
		}
		return
	}
	byHost := map[string][]provision.Record{}
	for _, rec := range recs {
		byHost[rec.Host] = append(byHost[rec.Host], rec)
	}
	for _, hd := range hosts {
		host := hd.Name()
		if !hd.IsDir() || !journalComponent(host) {
			continue
		}
		if w.hostBlocked(ctx, s, host) {
			continue // the lost-box gate (movespass.go hostBlocked)
		}
		ops, err := os.ReadDir(filepath.Join(dd, ".moves", host))
		if err != nil {
			w.notice(host+"#final-sweep", "moves: %s: the op journal sweep cannot list its journals: %v", host, err)
			continue
		}
		for _, od := range ops {
			op := od.Name()
			if !od.IsDir() || !journalComponent(op) {
				continue
			}
			if err := w.sweepFinalJournal(dd, host, op, byHost[host]); err != nil {
				w.logTrouble(host, fmt.Errorf("the op journal sweep, %s: %w", op, err))
			}
		}
	}
}

// sweepFinalJournal is one journal of sweepFinalJournals.
func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provision.Record) error {
	j, err := moves.ReadJournal(dataDir, host, op)
	if err != nil {
		return err
	}
	stamp, reached := j.Steps[finalStepParked]
	if !reached {
		// Not a finalization's: a rollback source's recovery journal
		// (movesrecover.go, Task 5 fix round 1), or a rollback's target
		// journal (movesrollback.go, Task 4 fix round 1).
		if _, ok := j.Steps[rcStepSource]; ok {
			return w.sweepRecoveryJournal(dataDir, host, op, j, recs)
		}
		if _, ok := j.Steps[rvStepSource]; ok {
			return w.sweepRevertJournal(dataDir, host, op, j, recs) // movesrevert.go, Task 6
		}
		return w.sweepRollbackJournal(dataDir, host, op, j, recs)
	}
	if stamp != "done" {
		if len(recs) == 0 {
			w.notice(host+"#final-sweep-"+op, "moves: %s: the journal of %s reached its stamp, and no listed record names the host; it is left", host, op)
			return nil
		}
		for _, rec := range recs {
			if m := rec.Move; m != nil && m.ID == op && m.SourceParkedAt == nil {
				return nil // the finalization's own pass stamps it
			}
		}
	}
	mark := func(step, value string) error {
		j.Mark(step, value)
		if err := j.Commit(dataDir); err != nil {
			return err
		}
		if fn := w.cfg.finalFault; fn != nil {
			return fn("after-" + step)
		}
		return nil
	}
	if stamp != "done" {
		if err := mark(finalStepParked, "done"); err != nil {
			return err
		}
	}
	return parkFinalJournal(dataDir, host, op, j, mark)
}