package host

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"slices"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/regionname"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootstate"
	"github.com/carlosframework/platform/internal/safefile"
)

// heldReleaseReceiptDir is where root keeps its publish receipts: its own
// claim state directory (carlos-instance-control's StateDirectory in claim
// mode), which the agent cannot write.
const heldReleaseReceiptDir = "/var/lib/carlos-claims/held-release"

// movesHeldReleaseRoot is root's side of the held release (Task 7b; see
// movesheldrelease.go). Every authority it checks comes from root's own
// providers: the authenticated peer, the claim and record it reads from the
// deployment store, the registry row, the render journal it reads through
// the anchored path, and its own held-release selection. A request only
// names what the agent expects them to say.
type movesHeldReleaseRoot struct {
	OpenRelease func() (rootstate.Store, error)
	// Authenticate is root's peer authentication (the edge unit's main
	// process), the claim root's own.
	Authenticate func(context.Context, net.Conn) (StopPeer, error)
	// Actor is this box's verified roster identity.
	Actor func(context.Context) (hostbox.Actor, error)
	// Store is the deployment store: releases and the source's report.
	Store   release.Store
	Claims  func(context.Context, string) (hostclaim.Observed, error)
	Records MovesRecordReader
	Route   func(string) (*registry.Route, error)
	// Works selects the claim's held-release work for host
	// (hostbox.WorkSource.Select, built for actor); Selected reads back the
	// selection it captured (hostbox.Selected).
	Works    func(ctx context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error)
	Selected func(hostwrite.Work) (hostbox.Selection, error)
	// Pub, Ratchet and Proof are §4.1's and D3's evidence, as for root's
	// adoption pass: the release key, root's pointer digest ratchet and a
	// fresh proof gate per request.
	Pub            ed25519.PublicKey
	Ratchet        func() *release.DigestRatchet
	Proof          func() *release.ProofGate
	OptDir         string
	generationPins release.PinSource
	// DataDir is the box's data directory: the render journals and the
	// held-stage lock.
	DataDir    string
	ReceiptDir string

	// Test seams. Nil is production.
	stage        func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate) (*release.Manifest, error)
	materialize  func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate, hostwrite.Work, func(func() error) error) (*release.Manifest, error)
	renderHash   func(dataDir, host, op string) (string, error)
	afterPublish func(host string) error

	// The bound (fix round 2, P3-4): one request per host at a time, and
	// heldReleaseConcurrency in all.
	mu       sync.Mutex
	inflight map[string]bool
}

// admit takes host's slot and one of the global ones, or refuses at once
// with errHeldReleaseBusy: root never queues an agent's requests. The
// returned func frees both.
func (h *movesHeldReleaseRoot) admit(host string) (func(), error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.inflight == nil {
		h.inflight = map[string]bool{}
	}
	if h.inflight[host] || len(h.inflight) >= heldReleaseConcurrency {
		return nil, errHeldReleaseBusy
	}
	h.inflight[host] = true
	return func() {
		h.mu.Lock()
		delete(h.inflight, host)
		h.mu.Unlock()
	}, nil
}

func (h *movesHeldReleaseRoot) complete() bool {
	return h != nil && h.OpenRelease != nil && h.Actor != nil && h.Store != nil && h.Claims != nil && h.Records != nil && h.Route != nil && h.Works != nil && h.Selected != nil &&
		len(h.Pub) == ed25519.PublicKeySize && h.Ratchet != nil && h.Proof != nil && filepath.IsAbs(h.OptDir) && filepath.IsAbs(h.DataDir) && filepath.IsAbs(h.ReceiptDir)
}

// ServeHTTP answers POST /claim/held-release for an authenticated peer.
// The peer is authenticated before the body is read.
func (h *movesHeldReleaseRoot) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if h == nil || h.Authenticate == nil || r.Method != http.MethodPost || r.URL.Path != heldReleasePath {
		http.Error(w, "refused", http.StatusServiceUnavailable)
		return
	}
	conn := instancectl.RequestConn(r.Context())
	if conn == nil {
		http.Error(w, "refused", http.StatusForbidden)
		return
	}
	peer, err := h.Authenticate(r.Context(), conn)
	if err != nil {
		http.Error(w, "refused", http.StatusForbidden)
		return
	}
	q, err := decodeHeldReleaseRequest(r.Body)
	if err != nil {
		http.Error(w, "refused", http.StatusBadRequest)
		return
	}
	out, err := h.Serve(r.Context(), peer, q)
	if err != nil {
		// Fix round 2, P3-2: a request's fields reach the log only once they
		// have passed validation, and then quoted, as is the error, so the
		// agent cannot forge a line in instance-control's journal.
		if q.validate() != nil {
			log.Printf("moves: held release refused: a malformed request: %q", err.Error())
		} else {
			log.Printf("moves: %q: held release %q of %q refused: %q", q.Host, q.Action, q.Version, err.Error())
		}
		http.Error(w, "refused", heldReleaseStatus(err))
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	_ = json.NewEncoder(w).Encode(out)
}

// heldReleaseStatus is the status each refusal answers with; the agent's
// client maps it back to the same typed error.
func heldReleaseStatus(err error) int {
	var refusal *MovesPolicyRefusal
	switch {
	case errors.Is(err, errHeldReleaseShape):
		return http.StatusBadRequest
	case errors.Is(err, ErrClaimStopIdentity):
		return http.StatusForbidden
	case errors.Is(err, errHeldReleaseNotYet):
		return http.StatusTooEarly
	case errors.Is(err, release.ErrHeldNotProven):
		return http.StatusUnprocessableEntity
	case errors.Is(err, errHeldReleaseUnavailable), errors.Is(err, errHeldReleaseBusy), errors.Is(err, hostwrite.ErrUnavailable):
		return http.StatusServiceUnavailable
	case errors.Is(err, release.ErrStaleHeldStage), errors.Is(err, hostbox.ErrChanged), errors.Is(err, errHeldReleaseUnsafe),
		errors.Is(err, release.ErrHeldNotAuthorized), errors.Is(err, release.ErrManifestDigest), errors.As(err, &refusal):
		return http.StatusConflict
	}
	return http.StatusInternalServerError
}

// Serve answers one request for an authenticated peer. Its order: the peer,
// the request's shape, this box's identity, then the action's own reads,
// every refusal before any effect.
func (h *movesHeldReleaseRoot) Serve(ctx context.Context, peer StopPeer, q heldReleaseRequest) (heldReleaseResponse, error) {
	if !h.complete() {
		return heldReleaseResponse{}, errHeldReleaseUnavailable
	}
	if peer.UID == 0 || peer.PID <= 0 || peer.ProcessStart == "" || peer.BootID == "" || peer.ControllerID == "" {
		return heldReleaseResponse{}, ErrClaimStopIdentity
	}
	if err := q.validate(); err != nil {
		return heldReleaseResponse{}, err
	}
	done, err := h.admit(q.Host)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	defer done()
	actor, err := h.Actor(ctx)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	if !actor.Eligible || actor.Quarantined || actor.Box == "" {
		return heldReleaseResponse{}, hostbox.ErrChanged
	}
	if q.Box != actor.Box {
		return heldReleaseResponse{}, staleHeldStage("it names box %q and this is %q", q.Box, actor.Box)
	}
	if !regionname.SameRegion(actor.HomeRegion, q.Region, regionname.PlacementRegion(actor.Region)) {
		return heldReleaseResponse{}, staleHeldStage("it names region %q and this box is in %q", q.Region, actor.Region)
	}
	var out heldReleaseResponse
	err = withRootReleaseStore(ctx, h.OpenRelease, func() error {
		var err error
		if q.Action == heldReleaseStage {
			out, err = h.stageRelease(ctx, actor, q)
		} else {
			out, err = h.publishRelease(ctx, actor, q)
		}
		return err
	})
	return out, err
}

// heldReleaseAuthority is root's own read of what a request names.
type heldReleaseAuthority struct {
	claim *hostclaim.Claim
	rec   *provision.Record
	row   *registry.Route
}

// authority re-reads the claim, the record and the registry row, and
// refuses a request that is not exactly theirs: the holder's incarnation,
// stamp, session, generation, account and app; the op's id, kind and
// phase; this box as the move's target (moving, landing) or as the
// executing source (restoring, reverting), and not declared lost; the hold
// op the row must be held by (the move id, or the claim's intent.rollback
// for a restoring op, ruling J3); the record's move and incarnation; and a
// registry row, when there is one, held by that op for that account and
// app.
func (h *movesHeldReleaseRoot) authority(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseAuthority, error) {
	obs, err := h.Claims(ctx, q.Host)
	if err != nil {
		return heldReleaseAuthority{}, fmt.Errorf("host claim for %s: %w", q.Host, err)
	}
	if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Host != q.Host {
		return heldReleaseAuthority{}, staleHeldStage("the host has no claim")
	}
	c, hd := obs.Claim, obs.Claim.Holder
	if hd.Incarnation != q.Incarnation || hd.Stamp != q.Stamp || hd.Session != q.Session || hd.Generation != q.Generation {
		return heldReleaseAuthority{}, staleHeldStage("the claim's holder is not the one the request observed")
	}
	if hd.Account != q.Account || hd.App != q.App {
		return heldReleaseAuthority{}, staleHeldStage("the claim is not held by %s/%s", q.Account, q.App)
	}
	if c.Op.ID != q.MoveID || c.Op.Kind != q.Kind || c.Op.Phase != q.Phase {
		return heldReleaseAuthority{}, staleHeldStage("the claim is %s %s at %q", c.Op.Kind, c.Op.ID, c.Op.Phase)
	}
	if slices.Contains(c.Lost, actor.Box) {
		return heldReleaseAuthority{}, staleHeldStage("this box is declared lost")
	}
	hold := q.MoveID
	switch c.Op.Kind {
	case hostclaim.KindMoving, hostclaim.KindLanding:
		if c.Op.Intent["target"] != actor.Box {
			return heldReleaseAuthority{}, staleHeldStage("the claim's target is %q", c.Op.Intent["target"])
		}
	case hostclaim.KindRestoring, hostclaim.KindReverting:
		if c.Op.Exec != actor.Box || c.Op.Intent["source"] != actor.Box {
			return heldReleaseAuthority{}, staleHeldStage("this box is not the move's executing source")
		}
		if c.Op.Kind == hostclaim.KindRestoring {
			if hold = c.Op.Intent["rollback"]; !provision.ValidRollbackID(hold) {
				return heldReleaseAuthority{}, staleHeldStage("the claim names no rollback")
			}
		}
	default:
		return heldReleaseAuthority{}, staleHeldStage("the claim's op is %s", c.Op.Kind)
	}
	if q.HoldOp != hold {
		return heldReleaseAuthority{}, staleHeldStage("the row is held by %q, not %q", hold, q.HoldOp)
	}
	rec, err := h.Records(ctx, q.Account, q.App, q.Host)
	if err != nil {
		return heldReleaseAuthority{}, fmt.Errorf("record for %s: %w", q.Host, err)
	}
	if rec == nil || rec.Host != q.Host || rec.Account != q.Account || rec.App != q.App || rec.Move == nil || rec.Move.ID != q.MoveID || rec.Move.Incarnation != hd.Incarnation {
		return heldReleaseAuthority{}, staleHeldStage("the record is not this move")
	}
	inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	if err != nil || inc != hd.Incarnation || hd.Key != provision.RecordKey(rec.Account, rec.App, rec.Host) {
		return heldReleaseAuthority{}, hostbox.ErrChanged
	}
	if c.Op.Kind == hostclaim.KindRestoring {
		if rb := rec.Move.Rollback; rec.Move.State != provision.MoveRollingBack || rb == nil || rb.ID != hold {
			return heldReleaseAuthority{}, staleHeldStage("the record is not this rollback")
		}
	}
	row, err := h.Route(q.Host)
	if err != nil {
		return heldReleaseAuthority{}, fmt.Errorf("registry row for %s: %w", q.Host, err)
	}
	if row != nil && (row.HoldOp != hold || release.RouteAccount(*row) != q.Account || row.App != q.App) {
		return heldReleaseAuthority{}, staleHeldStage("the registry row is %s/%s held by %q", release.RouteAccount(*row), row.App, row.HoldOp)
	}
	return heldReleaseAuthority{claim: c, rec: rec, row: row}, nil
}

// heldReleaseRoute is the held route a request's effect runs for.
func heldReleaseRoute(q heldReleaseRequest) registry.Route {
	return registry.Route{Host: q.Host, Kind: "instance", App: q.App, Account: q.Account, HoldOp: q.HoldOp}
}

// stageRelease is T7 under the claim, for a move's target (moving or
// landing) or a dirty return's source (restoring): the version must be the
// source's reported release and, once pinned, the record's pin (for a
// return, its recorded recovery release, else the pin). It stages and
// verifies (release.StageHeldVerified), never touches live/<host>, selects
// nothing and holds no hostguard: the version store is shared and a stage
// is no host's write.
func (h *movesHeldReleaseRoot) stageRelease(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseResponse, error) {
	a, err := h.authority(ctx, actor, q)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	m := a.rec.Move
	switch q.Kind {
	case hostclaim.KindMoving, hostclaim.KindLanding:
		raw, _, err := h.Store.Get(ctx, moves.SourceReportKey(moves.ReportPrefix(q.Account, q.App, q.Host, q.MoveID)))
		if errors.Is(err, os.ErrNotExist) {
			return heldReleaseResponse{}, staleHeldStage("the source has not reported its release")
		}
		if err != nil {
			return heldReleaseResponse{}, fmt.Errorf("source report for %s: %w", q.Host, err)
		}
		var sr moves.SourceReport
		if err := json.Unmarshal(raw, &sr); err != nil {
			return heldReleaseResponse{}, fmt.Errorf("source report for %s: %w", q.Host, err)
		}
		if sr.Release == "" || sr.Release != q.Version {
			return heldReleaseResponse{}, staleHeldStage("the source runs %q", sr.Release)
		}
		if m.Inputs != nil && m.Inputs.Release != q.Version {
			return heldReleaseResponse{}, staleHeldStage("the pinned release is %q", m.Inputs.Release)
		}
	case hostclaim.KindRestoring:
		if want := heldReturnRelease(m); want == "" || want != q.Version {
			return heldReleaseResponse{}, staleHeldStage("the return's release is %q", want)
		}
	default:
		return heldReleaseResponse{}, staleHeldStage("a %s op stages nothing", q.Kind)
	}
	stage := h.stage
	if stage == nil {
		pins := h.heldGenerationPins(ctx, q.Version)
		stage = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, ratchet *release.DigestRatchet, opt string, rt registry.Route, version string, proof *release.ProofGate) (*release.Manifest, error) {
			return release.StageHeldVerifiedWithPins(ctx, s, pub, ratchet, opt, rt, version, proof, pins)
		}
	}
	if err := withHeldStageLock(h.DataDir, q.Host, func() error {
		_, err := stage(ctx, h.Store, h.Pub, h.Ratchet(), h.OptDir, heldReleaseRoute(q), q.Version, h.Proof())
		return err
	}); err != nil {
		return heldReleaseResponse{}, err
	}
	log.Printf("moves: %q: held release %q staged for %q (no publication)", q.Host, q.Version, q.MoveID)
	return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomeStaged}, nil
}

// heldReturnRelease is a return's release: its recorded recovery release,
// else the source's pin.
func heldReturnRelease(m *provision.MoveState) string {
	if rb := m.Rollback; rb != nil && rb.RecoveryInputs != nil && rb.RecoveryInputs.Release != "" {
		return rb.RecoveryInputs.Release
	}
	if m.Inputs != nil {
		return m.Inputs.Release
	}
	return ""
}

// publishVersion is the release a publication may name, by the claim's
// kind and phase, and "" with the reason when the claim grants none: a
// landing's configured or probing step, on this box as the executor, with
// the record at that landing phase (the pinned release); a dirty return's
// configured step (claimed) or its probe (its recovery release, the pin at
// configured when none is recorded yet). Anything else, a moving op
// included, publishes nothing: the policy is never widened.
//
// A revert's probe and a clean return's probe are refused outright (fix
// round 2, P2-1, platform-1e's authority review). Neither runs configured,
// so no render is journalled under a rule root checks, and the journal is
// agent-written: its absence cannot be what refuses them, since a
// compromised agent could plant one. They stay refused until those probes
// journal their own render (RenderHeldUnderBefore at an equivalent point)
// under a rule root checks.
func publishVersion(c *hostclaim.Claim, m *provision.MoveState, box string) (string, error) {
	if c.Op.Exec != box {
		return "", staleHeldStage("this box is not the executor")
	}
	switch c.Op.Kind {
	case hostclaim.KindLanding:
		if c.Op.Phase != provision.LandConfigured && c.Op.Phase != provision.LandProbing {
			return "", staleHeldStage("a landing at %q publishes nothing", c.Op.Phase)
		}
		if m.State != provision.MoveLanding || m.LandingPhase != c.Op.Phase {
			return "", staleHeldStage("the record is %s at %q", m.State, m.LandingPhase)
		}
		if m.Inputs == nil || m.Inputs.Release == "" {
			return "", staleHeldStage("no pinned release")
		}
		return m.Inputs.Release, nil
	case hostclaim.KindRestoring:
		rb := m.Rollback
		switch {
		case c.Op.Phase == hostclaim.PhaseClaimed && rb.SourcePhase == provision.RbSourceConfigured && rb.Dirty:
		case c.Op.Phase == hostclaim.PhaseProbing && rb.SourcePhase == provision.RbSourceProbing:
			if !rb.Dirty {
				return "", staleHeldStage("a clean return's probe journals no render root checks, so root publishes nothing for it")
			}
			if rb.RecoveryInputs == nil || rb.RecoveryInputs.Release == "" {
				return "", staleHeldStage("no recorded recovery release")
			}
		default:
			return "", staleHeldStage("a rollback at %q (source %q) publishes nothing", c.Op.Phase, rb.SourcePhase)
		}
		if v := heldReturnRelease(m); v != "" {
			return v, nil
		}
		return "", staleHeldStage("no recorded recovery release")
	case hostclaim.KindReverting:
		return "", staleHeldStage("a revert's probe journals no render root checks, so root publishes nothing for it")
	}
	return "", staleHeldStage("a %s op publishes nothing", c.Op.Kind)
}

func (h *movesHeldReleaseRoot) publishRelease(ctx context.Context, actor hostbox.Actor, q heldReleaseRequest) (heldReleaseResponse, error) {
	a, err := h.authority(ctx, actor, q)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	want, err := publishVersion(a.claim, a.rec.Move, actor.Box)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	if q.Version != want {
		return heldReleaseResponse{}, staleHeldStage("the release to publish is %q", want)
	}
	if a.row == nil {
		return heldReleaseResponse{}, staleHeldStage("the host has no held row")
	}
	replayHint := h.replayed(q)
	if !replayHint {
		read := h.renderHash
		if read == nil {
			read = readHeldRenderHash
		}
		rendered, err := read(h.DataDir, q.Host, q.HoldOp)
		switch {
		case errors.Is(err, os.ErrNotExist) || (err == nil && rendered == ""):
			return heldReleaseResponse{}, errHeldReleaseNotYet
		case err != nil:
			return heldReleaseResponse{}, err
		case rendered != q.RenderHash:
			return heldReleaseResponse{}, staleHeldStage("the journalled render is %s", rendered)
		}

	}
	work, err := h.Works(ctx, actor, q.Host, movesActionHeldRelease)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	sel, err := h.Selected(work)
	if err != nil {
		return heldReleaseResponse{}, err
	}
	if sel.Host != q.Host || sel.Action != movesActionHeldRelease || sel.Box != actor.Box || sel.Captured.Incarnation != q.Incarnation || sel.Captured.Stamp != q.Stamp ||
		sel.Session != q.Session || sel.Generation != q.Generation || sel.Kind != q.Kind || sel.Op != q.MoveID || sel.Phase != q.Phase {
		return heldReleaseResponse{}, hostbox.ErrChanged
	}
	checked := heldReleaseCheckedWork{Work: work, check: func() error {
		fresh, err := h.authority(ctx, actor, q)
		if err != nil {
			return err
		}
		version, err := publishVersion(fresh.claim, fresh.rec.Move, actor.Box)
		if err != nil {
			return err
		}
		if version != q.Version || fresh.row == nil {
			return hostbox.ErrChanged
		}
		return nil
	}}
	if replayHint {
		if err := checked.With(ctx, hostwrite.Identity{Release: q.Version}, func() error {
			if !h.replayed(q) {
				return hostbox.ErrChanged
			}
			return nil
		}); err != nil {
			return heldReleaseResponse{}, err
		}
		return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomePublished, Replayed: true}, nil
	}
	checked.after = func() error {
		if h.afterPublish != nil {
			if err := h.afterPublish(q.Host); err != nil {
				return err
			}
		}
		if err := h.writeReceipt(q); err != nil {
			return fmt.Errorf("%w: %v", errHeldReleaseReceipt, err)
		}
		return nil
	}
	materialize := h.materialize
	if materialize == nil {
		pins := h.heldGenerationPins(ctx, q.Version)
		materialize = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, ratchet *release.DigestRatchet, opt string, rt registry.Route, version string, proof *release.ProofGate, work hostwrite.Work, guard func(func() error) error) (*release.Manifest, error) {
			return release.MaterializeHeldUnderWithPins(ctx, s, pub, ratchet, opt, rt, version, proof, work, guard, pins)
		}
	}
	lock := func(publish func() error) error { return withHeldStageLock(h.DataDir, q.Host, publish) }
	if _, err := materialize(ctx, h.Store, h.Pub, h.Ratchet(), h.OptDir, heldReleaseRoute(q), q.Version, h.Proof(), checked, lock); err != nil {
		return heldReleaseResponse{}, err
	}
	log.Printf("moves: %q: held release %q published for %q at %q %q", q.Host, q.Version, q.MoveID, q.Kind, q.Phase)
	return heldReleaseResponse{Schema: heldReleaseSchema, RequestID: q.RequestID, Outcome: heldReleaseOutcomePublished}, nil
}

// Serve holds release exclusion until publication returns. Capture before the
// leaf held-stage lock: retirement may acquire root-host and legacy-host locks.
// Keep a failed read restrictive only for writes, so an unchanged artifact can
// still be used and a wholly absent version tree can still be staged.
func (h *movesHeldReleaseRoot) heldGenerationPins(ctx context.Context, version string) release.PinSource {
	if h.generationPins == nil {
		return nil
	}
	if _, err := os.Lstat(filepath.Join(h.OptDir, "versions", version)); errors.Is(err, os.ErrNotExist) {
		return h.generationPins
	}
	pins, err := h.generationPins.StagedPins(ctx, h.OptDir)
	return release.PinSourceFunc(func(context.Context, string) (release.StagedPaths, error) { return pins, err })
}

type heldReleaseCheckedWork struct {
	hostwrite.Work
	check func() error
	after func() error
}

func (w heldReleaseCheckedWork) With(ctx context.Context, id hostwrite.Identity, effect func() error) error {
	return w.Work.With(ctx, id, func() error {
		if err := w.check(); err != nil {
			return err
		}
		if err := effect(); err != nil {
			return err
		}
		if w.after != nil {
			return w.after()
		}
		return nil
	})
}

// heldReleaseReceipt is root's record of a publication it made, one per
// host (the latest), keyed by the session and the render hash with the
// identity the publication was for, and the action, kind and phase it was
// made at (fix round 2, P3-1), so a landing's configured receipt never
// answers its probing request.
type heldReleaseReceipt struct {
	Schema      string    `json:"schema"`
	Action      string    `json:"action"`
	Kind        string    `json:"kind"`
	Phase       string    `json:"phase"`
	Host        string    `json:"host"`
	Account     string    `json:"account"`
	App         string    `json:"app"`
	Version     string    `json:"version"`
	MoveID      string    `json:"move_id"`
	HoldOp      string    `json:"hold_op"`
	Incarnation string    `json:"incarnation"`
	Stamp       uint64    `json:"stamp"`
	Session     uint64    `json:"session"`
	Generation  string    `json:"generation"`
	RenderHash  string    `json:"render_hash"`
	Outcome     string    `json:"outcome"`
	At          time.Time `json:"at"`
}

func receiptFor(q heldReleaseRequest) heldReleaseReceipt {
	return heldReleaseReceipt{Schema: heldReleaseSchema, Action: q.Action, Kind: q.Kind, Phase: q.Phase, Host: q.Host, Account: q.Account, App: q.App, Version: q.Version, MoveID: q.MoveID, HoldOp: q.HoldOp,
		Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, Generation: q.Generation, RenderHash: q.RenderHash, Outcome: heldReleaseOutcomePublished}
}

func (h *movesHeldReleaseRoot) receiptPath(host string) string {
	return filepath.Join(h.ReceiptDir, host+".json")
}

// replayed reports whether root's receipt for q's host is for exactly q:
// the same session and render hash, the same identity, action, kind and
// phase. Anything else, an unreadable receipt included, is not a replay and
// runs the whole path.
func (h *movesHeldReleaseRoot) replayed(q heldReleaseRequest) bool {
	f, err := safefile.OpenFile(h.receiptPath(q.Host), os.O_RDONLY, 0)
	if err != nil {
		return false
	}
	defer f.Close()
	raw, err := io.ReadAll(io.LimitReader(f, heldReleaseRequestMax+1))
	if err != nil || len(raw) > heldReleaseRequestMax {
		return false
	}
	var got heldReleaseReceipt
	if err := json.Unmarshal(raw, &got); err != nil {
		return false
	}
	got.At = time.Time{}
	return got == receiptFor(q)
}

func (h *movesHeldReleaseRoot) writeReceipt(q heldReleaseRequest) error {
	r := receiptFor(q)
	r.At = time.Now().UTC()
	raw, err := json.Marshal(r)
	if err != nil {
		return err
	}
	if err := safefile.MkdirAll(h.ReceiptDir, 0o700); err != nil {
		return err
	}
	return safefile.WriteFile(h.receiptPath(q.Host), raw, 0o600)
}