package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
)

// The moves write policy (moves Phase 1 Plan 2 Task 10b; ruling T10b-scope).
//
// hostbox.Policy.Check separates ordinary host writes from the moves-owned
// ones: under a move kind it checks the registry hold and the selection's
// op and box, and then asks Policy.Moves. MovesClaimPolicy builds that
// answer, the value for ClaimOwnerBindings.MovesPolicy. It grants a write
// only when every one of these holds, and refuses everything else with a
// *MovesPolicyRefusal (fail closed):
//
//   - the claim's op is the landing of a move (hostclaim.KindLanding), or a
//     recovery (below). A moving op grants nothing: the source writes
//     nothing held while moving;
//   - the selection is the claim's: the same host, op, kind, phase and
//     capture (holder incarnation and stamp);
//   - this box is the target after the flip: the executor, the second
//     participant, intent.target, and not intent.source; in the move's
//     target region (intent.to); and not in lost;
//   - the landing's phase permits the action (movesPhaseRights): held-timers
//     at local; held-config, held-release, restore and install at
//     configured, the phase whose work installs the database before its
//     closing CAS writes restored (G8; the final review's finding 3), so
//     restored, the install complete, grants none of them. probing (Plan 3
//     Task 5) grants the held render, the held release and the spawn the
//     probe's whole admitted sequence runs under; probed grants nothing to
//     a fresh selection, and the probe's own spawn selection continues at
//     probed and committed on the same session (H3). The ordinary wake
//     rights at committed (§19.1) are Task 7's row;
//   - the identity fits the action: a held-config write names the config
//     and nothing else, a held-release write names the release and nothing
//     else, the probe's spawn names the launch identity (the pinned config
//     and release, H4), and every other action carries no identity;
//   - a held write names the move's pinned inputs (§9.1, §9.2). Policy.Inputs
//     is consulted for serving claims only (wakepolicy.go), so this policy
//     reads the record itself, through the reader it is built with: the
//     record must be this landing (the op, the holder's incarnation, state
//     landing), Move.Inputs must be pinned, a held-release identity must
//     equal Inputs.Release, and a held-config identity must equal
//     HeldConfigIdentity(Inputs.BundleETags, Inputs.OverlayETag), the
//     canonical form of the pinned etags (the empty configuration included:
//     Move.Inputs present is the pin). Missing Move.Inputs, a record that
//     is not this landing, a read error or no reader refuse. The record is
//     read for these two actions only.
//
// The recoveries (moves Plan 4 Task 3b, ruling J12) are the source's rows,
// granted only when every one of these holds:
//
//   - this box is the claim op's executor, as Policy.Check requires; the
//     move's source (the first participant and intent.source, not
//     intent.target); not in the target region (intent.to) but in the
//     record's From; and not in lost;
//   - the record, read for every recovery action, is this move and this
//     holder incarnation: rolling_back for this rollback (Rollback.ID is
//     intent.rollback) under restoring, reverting under reverting;
//   - the row permits the action. restoring is keyed by the record's
//     Rollback.SourcePhase: local grants held-timers, configured held-config,
//     held-release, restore and install (the installer restores inside
//     configured, before the CAS that writes restored, G8; restored grants
//     nothing), each under the claim at claimed, for a dirty rollback, with the holder at the source's
//     pre-move stamp S (op.journal.source_stamp); probing grants the held
//     config render and the release materialization before every probe
//     (§7.7) and spawn, under the claim at probing. reverting is keyed by
//     the claim's rev_phase: probing grants the same three, with op.phase at
//     probing (J1);
//   - the identity fits the action, a spawn carrying the launch identity
//     (Plan 3 H4): the config and the release it launches with;
//   - the held config and release and the launch name the inputs the
//     recovery rendered, Rollback.RecoveryInputs as the CAS that ends
//     configured records them, never Move.Inputs (§7.7, J2-R2-5). Before
//     they are recorded the render is granted on the rows above: configured
//     is the render the record takes them from, and a clean return renders
//     before its probe with no configured. A dirty return's probe, and every
//     spawn, need them recorded. A revert's are Move.RecoveryInputs, which
//     its probe records before it renders (MoveRecordRevertInputs; M11,
//     Q11): its render, its release and its launch need them recorded and
//     are bound to them, never to Move.Inputs.
//
// The target's cleaning (ReplaceTimers(host, nil) at cleaning) and the
// source's revert close are not rows here: their callers use plain registry
// calls under WithHostHoldLock, as Plan 3 H5(d) ruled for the source's
// finalization, and this policy refuses the target any write under a
// recovery.
//
// Production registration. RegisterProductionClaimOwners requires every
// owner, and the Inputs, PublicationEvidence and AdmissionEvidence owners
// (Plan 3) and View, Authorize, Evidence and LeaseSet (moves Plan 5) do not
// exist yet, so nothing registers this in production: it is tested and
// ready to plug in as MovesClaimPolicy(MovesRecordsFrom(store)). Until that
// registration, the landing keeps refusing at checking while
// CARLOS_HOST_CLAIM is on (ruling G16, movesland.go).

// ErrMovesRefused is what every MovesPolicyRefusal unwraps to.
var ErrMovesRefused = errors.New("host: the moves policy refuses the write")

// MovesPolicyRefusal is MovesClaimPolicy's typed refusal: which action was
// asked under which op, kind and phase, by which box, and why. Cause is
// the record read's error when that is the reason.
type MovesPolicyRefusal struct {
	Host, Op, Action, Kind, Phase, Box, Reason string
	Cause                                      error
}

func (e *MovesPolicyRefusal) Error() string {
	msg := fmt.Sprintf("host: the moves policy refuses %s for %s under %s/%q on %s: %s", e.Action, e.Host, e.Kind, e.Phase, e.Box, e.Reason)
	if e.Cause != nil {
		msg += ": " + e.Cause.Error()
	}
	return msg
}

func (e *MovesPolicyRefusal) Unwrap() []error {
	if e.Cause != nil {
		return []error{ErrMovesRefused, e.Cause}
	}
	return []error{ErrMovesRefused}
}

// MovesRecordReader reads the host's provisioning record: the move's
// pinned inputs live there. Production is MovesRecordsFrom(store).
type MovesRecordReader func(ctx context.Context, acct, app, host string) (*provision.Record, error)

// MovesRecordsFrom is provision.GetRecord over the deployment store.
func MovesRecordsFrom(s release.Store) MovesRecordReader {
	return func(ctx context.Context, acct, app, host string) (*provision.Record, error) {
		if s == nil {
			return nil, hostwrite.ErrUnavailable
		}
		return provision.GetRecord(ctx, s, acct, app, host)
	}
}

// heldConfigIdentityPrefix names the encoding HeldConfigIdentity uses, so
// a later form can be told apart from it.
const heldConfigIdentityPrefix = "etags-v1:"

// heldConfigPins is what HeldConfigIdentity encodes: the non-empty bundle
// etags and the overlay etag. encoding/json writes map keys sorted and
// escapes every delimiter, so two different sets never share an identity.
type heldConfigPins struct {
	ETags   map[string]string `json:"etags"`
	Overlay string            `json:"overlay"`
}

// HeldConfigIdentity is the identity a held-config write carries (§9.1):
// the bundle etags and the overlay etag it rendered from, in one canonical
// string, so the policy can compare them with the move's pins exactly. An
// empty etag is the same as an absent one. The empty configuration (an app
// with no default bundle and no overlay pins exactly that at readiness) has
// its own identity, the encoding of the empty set; it is never the zero
// identity.
func HeldConfigIdentity(bundleETags map[string]string, overlayETag string) hostwrite.Identity {
	pins := heldConfigPins{ETags: map[string]string{}, Overlay: overlayETag}
	for k, v := range bundleETags {
		if k != "" && v != "" {
			pins.ETags[k] = v
		}
	}
	raw, err := json.Marshal(pins)
	if err != nil {
		// Strings always marshal; a failure here would be a Go runtime
		// fault, and the zero identity is refused everywhere.
		return hostwrite.Identity{}
	}
	return hostwrite.Identity{Config: heldConfigIdentityPrefix + string(raw)}
}

// The held actions and the landing's rights hostbox.Policy.Check asks
// Policy.Moves about (wakepolicy.go).
const (
	movesActionHeldConfig  = "held-config"
	movesActionHeldRelease = "held-release"
	movesActionHeldTimers  = "held-timers"
	movesActionRestore     = "restore"
	movesActionSpawn       = "spawn"
	movesActionSeed        = "seed"
	movesActionWriter      = "writer"
	movesActionInstall     = "install"
)

// movesPhaseRights is the landing's table: the actions each phase permits
// (§7.6), keyed by the phase whose work makes the write. configured renders,
// gates the release and installs the database; its closing CAS writes
// restored (G8), so restored grants nothing. A phase or action not listed is
// refused.
//
// probing (Plan 3 Task 5, H4) is the probe's: the held render and the held
// release before every start (§7.7), and the spawn, which carries the launch
// identity bound to the pins. The admitted session's restore, launch record,
// launch and replication all run under that one spawn selection (claim Plan
// 9 Group D), so restore, seed and writer get no row. probed grants nothing
// to a fresh selection; committed's ordinary wake is Task 7's row.
var movesPhaseRights = map[string]map[string]bool{
	provision.LandLocal:      {movesActionHeldTimers: true},
	provision.LandConfigured: {movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionRestore: true, movesActionInstall: true},
	provision.LandProbing:    {movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionSpawn: true},
}

// movesSessionContinues reports whether the claim's phase may differ from
// the selection's (H3, claim Plan 9 D1): only the admitted session's own
// spawn selection, and only along the landing's session row
// (hostclaim.SessionObserves: probing, probed, committed), so the probe's
// replication and process survive the probed and committed CASes. The
// rights stay the ones of the phase the selection was made at.
func movesSessionContinues(c *hostclaim.Claim, s hostbox.Selection) bool {
	if c.Op.Phase == s.Phase {
		return true
	}
	return s.Action == movesActionSpawn && hostclaim.SessionObserves(c.Op.Kind, s.Phase, c.Op.Phase)
}

// movesPolicyActions is every action the policy knows. Anything else is
// refused before the table is consulted.
var movesPolicyActions = map[string]bool{
	movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionHeldTimers: true,
	movesActionRestore: true, movesActionSpawn: true, movesActionSeed: true, movesActionWriter: true, movesActionInstall: true,
}

// MovesClaimPolicy builds hostbox.Policy.Moves for the moves plans over a
// record reader: see the file comment for what it grants.
// MovesClaimPolicy is MovesClaimPolicyWithInputs with no committed input
// check: the committed wake's spawn is refused (production, until the claim
// binds one; H4).
func MovesClaimPolicy(read MovesRecordReader) func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error {
	return MovesClaimPolicyWithInputs(read, nil)
}

// MovesClaimPolicyWithInputs builds hostbox.Policy.Moves. committed is the
// committed wake's input check (Plan 3 Task 7, H4): a fresh spawn selection
// at landing/committed carries the ordinary launch identity, and committed
// says whether it is the launch this host's row would run, as the serving
// InputPolicy does for serving. Nil refuses every committed spawn.
func MovesClaimPolicyWithInputs(read MovesRecordReader, committed func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error) func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error {
	return func(ctx context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {
		refuse := func(reason string, cause error) error {
			r := &MovesPolicyRefusal{Host: s.Host, Op: s.Op, Action: s.Action, Box: s.Box, Reason: reason, Cause: cause}
			if c != nil {
				r.Kind, r.Phase = c.Op.Kind, c.Op.Phase
			}
			return r
		}
		if ctx == nil {
			return refuse("no context", nil)
		}
		if err := ctx.Err(); err != nil {
			return err
		}
		if c == nil || c.Holder == nil {
			return refuse("the claim has no holder", nil)
		}
		if !movesPolicyActions[s.Action] {
			return refuse("not a moves action", nil)
		}
		if s.Host == "" || s.Box == "" || s.Region == "" || s.Op == "" {
			return refuse("the selection names no host, box, region or op", nil)
		}
		if c.Host != s.Host {
			return refuse("the claim is another host's", nil)
		}
		if c.Op.ID != s.Op {
			return refuse("the claim is under another op", nil)
		}
		if c.Op.Kind != s.Kind || !movesSessionContinues(c, s) {
			return refuse("the claim's kind or phase is not the selection's", nil)
		}
		if s.Captured.Stamp == 0 || c.Holder.Stamp != s.Captured.Stamp || c.Holder.Incarnation != s.Captured.Incarnation {
			return refuse("the capture is not the holder's", nil)
		}
		switch c.Op.Kind {
		case hostclaim.KindLanding:
		case hostclaim.KindReverting, hostclaim.KindRestoring:
			return movesRecoveryPolicy(ctx, read, c, s, id, refuse)
		case hostclaim.KindMoving:
			return refuse("a moving op writes nothing held", nil)
		default:
			return refuse("not a move", nil)
		}
		if c.Op.Exec != s.Box {
			return refuse("this box is not the executor", nil)
		}
		if len(c.Op.Participants) != 2 || c.Op.Participants[1] != s.Box || c.Op.Intent["target"] != s.Box || c.Op.Intent["source"] == s.Box {
			return refuse("this box is not the move's target", nil)
		}
		if _, err := park.VerifiedHome(s.FleetHomeRegion, s.HomeRegion, s.Region, c.Op.Intent["home_region"]); err != nil {
			return refuse("this region is not the move's target region", err)
		}
		if !park.SameRegion(s.HomeRegion, c.Op.Intent["to"], s.Region) {
			return refuse("this region is not the move's target region", nil)
		}
		for _, lost := range c.Lost {
			if lost == s.Box {
				return refuse("this box is declared lost", nil)
			}
		}
		if s.Phase == hostclaim.PhaseLandingCommitted && s.Action == movesActionSpawn {
			return movesCommittedSpawn(ctx, read, committed, c, s, id, refuse)
		}
		if !movesPhaseRights[s.Phase][s.Action] {
			return refuse("the phase does not permit the action", nil)
		}
		probeSpawn := s.Phase == provision.LandProbing && s.Action == movesActionSpawn
		if probeSpawn {
			if !recoveryIdentityFits(s.Action, id) {
				return refuse("the identity does not fit the action", nil)
			}
		} else if !movesIdentityFits(s.Action, id) {
			return refuse("the identity does not fit the action", nil)
		}
		if s.Action != movesActionHeldConfig && s.Action != movesActionHeldRelease && !probeSpawn {
			return nil
		}
		// The held config and release are the move's pinned inputs (§9.1,
		// §9.2), read from the record this claim's holder names.
		if read == nil {
			return refuse("no record reader", nil)
		}
		if c.Holder.Account == "" || c.Holder.App == "" {
			return refuse("the holder names no account or app", nil)
		}
		rec, err := read(ctx, c.Holder.Account, c.Holder.App, c.Host)
		if err != nil {
			return refuse("the record cannot be read", err)
		}
		if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != s.Op || rec.Move.Incarnation != c.Holder.Incarnation {
			return refuse("the record is not this landing", nil)
		}
		switch {
		case probeSpawn:
			// The probe's launch is this landing's, unmarked, or the same
			// session continuing once the record is committed (H3).
			if (rec.Move.State != provision.MoveLanding || rec.Move.Reason != "") && rec.Move.State != provision.MoveCommitted {
				return refuse("the record is not this landing", nil)
			}
		case rec.Move.State != provision.MoveLanding:
			return refuse("the record is not this landing", nil)
		}
		in := rec.Move.Inputs
		if in == nil {
			return refuse("no pinned inputs", nil)
		}
		if probeSpawn {
			if in.Release == "" || id.Config != HeldConfigIdentity(in.BundleETags, in.OverlayETag).Config || id.Release != in.Release {
				return refuse("the launch is not the pinned one", nil)
			}
			return nil
		}
		switch s.Action {
		case movesActionHeldRelease:
			if in.Release == "" {
				return refuse("no pinned release", nil)
			}
			if id.Release != in.Release {
				return refuse("the release is not the pinned one", nil)
			}
		case movesActionHeldConfig:
			// Move.Inputs present is the pin, empty etags included: an app
			// without a default bundle or an overlay pins the empty
			// configuration, and the render that names the same empty set
			// is granted.
			if id.Config != HeldConfigIdentity(in.BundleETags, in.OverlayETag).Config {
				return refuse("the config is not the pinned one", nil)
			}
		}
		return nil
	}
}

// recoveryRow is one row of a recovery's table: the claim phase it runs
// under, whether it is the dirty return's install, and the actions it
// grants.
type recoveryRow struct {
	claimPhase string
	install    bool
	actions    map[string]bool
}

// restoringRights is the source's table during a rollback, keyed by the
// record's Rollback.SourcePhase (§7.8 source table) at the phase whose work
// makes the write: the installer restores and installs inside configured,
// before the CAS that writes restored (G8, J24), so restored grants nothing.
// The claim stays at claimed through the dirty return's install (Task 3) and
// mirrors probing (J1). A source phase not listed grants nothing.
var restoringRights = map[string]recoveryRow{
	provision.RbSourceLocal: {claimPhase: hostclaim.PhaseClaimed, install: true, actions: map[string]bool{movesActionHeldTimers: true}},
	provision.RbSourceConfigured: {claimPhase: hostclaim.PhaseClaimed, install: true, actions: map[string]bool{
		movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionRestore: true, movesActionInstall: true}},
	provision.RbSourceProbing: {claimPhase: hostclaim.PhaseProbing, actions: map[string]bool{movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionSpawn: true}},
}

// revertingRights is the source's table during a revert, keyed by the
// claim's rev_phase (§7.4), which op.phase mirrors from probing on (J1).
var revertingRights = map[string]recoveryRow{
	hostclaim.PhaseProbing: {claimPhase: hostclaim.PhaseProbing, actions: map[string]bool{movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionSpawn: true}},
}

// movesRecoveryPolicy is MovesClaimPolicy's reverting and restoring rows:
// see the file comment. The claim, the selection and the capture have
// already been matched.
func movesRecoveryPolicy(ctx context.Context, read MovesRecordReader, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity, refuse func(string, error) error) error {
	if c.Op.Exec != s.Box {
		return refuse("this box is not the executor", nil)
	}
	if len(c.Op.Participants) == 0 || c.Op.Participants[0] != s.Box || c.Op.Intent["source"] != s.Box || c.Op.Intent["target"] == s.Box {
		return refuse("this box is not the move's source", nil)
	}
	if _, err := park.VerifiedHome(s.FleetHomeRegion, s.HomeRegion, s.Region, c.Op.Intent["home_region"]); err != nil {
		return refuse("this region is the move's target region", err)
	}
	if park.SameRegion(s.HomeRegion, c.Op.Intent["to"], s.Region) {
		return refuse("this region is the move's target region", nil)
	}
	for _, lost := range c.Lost {
		if lost == s.Box {
			return refuse("this box is declared lost", nil)
		}
	}
	if read == nil {
		return refuse("no record reader", nil)
	}
	if c.Holder.Account == "" || c.Holder.App == "" {
		return refuse("the holder names no account or app", nil)
	}
	rec, err := read(ctx, c.Holder.Account, c.Holder.App, c.Host)
	if err != nil {
		return refuse("the record cannot be read", err)
	}
	if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != s.Op || rec.Move.Incarnation != c.Holder.Incarnation {
		return refuse("the record is not this move", nil)
	}
	if !park.PlacementMatches(s.HomeRegion, rec.Move.From, s.Region) {
		return refuse("this region is not the move's source region", nil)
	}
	var (
		row      recoveryRow
		ok       bool
		rb       *provision.MoveRollback
		recovery *provision.MoveInputs
	)
	switch c.Op.Kind {
	case hostclaim.KindRestoring:
		rb = rec.Move.Rollback
		if rec.Move.State != provision.MoveRollingBack || rb == nil || rb.ID == "" || rb.ID != c.Op.Intent["rollback"] {
			return refuse("the record is not this rollback", nil)
		}
		row, ok = restoringRights[rb.SourcePhase]
		recovery = rb.RecoveryInputs
	default: // hostclaim.KindReverting
		if rec.Move.State != provision.MoveReverting {
			return refuse("the record is not this revert", nil)
		}
		row, ok = revertingRights[c.Op.Journal[hostclaim.JournalRevPhase]]
		recovery = rec.Move.RecoveryInputs
	}
	if !ok || row.claimPhase != c.Op.Phase || !row.actions[s.Action] {
		return refuse("the phase does not permit the action", nil)
	}
	if row.install {
		if !rb.Dirty {
			return refuse("only a dirty return installs", nil)
		}
		stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
		if err != nil || stamp == 0 || c.Holder.Stamp != stamp {
			return refuse("the holder is not at the source's pre-move stamp", nil)
		}
	}
	if !recoveryIdentityFits(s.Action, id) {
		return refuse("the identity does not fit the action", nil)
	}
	switch s.Action {
	case movesActionHeldConfig, movesActionHeldRelease:
		if recovery == nil {
			// configured renders the inputs its closing CAS records, and a
			// clean return's render before its probe precedes the record. A
			// revert's probe records before it renders (M11), so a revert
			// has no render before the record.
			if rb != nil && (rb.SourcePhase == provision.RbSourceConfigured || !rb.Dirty) {
				return nil
			}
			return refuse("no recorded recovery inputs", nil)
		}
	case movesActionSpawn:
		if recovery == nil {
			return refuse("no recorded recovery inputs", nil)
		}
	default:
		return nil
	}
	switch s.Action {
	case movesActionHeldConfig:
		if id.Config != HeldConfigIdentity(recovery.BundleETags, recovery.OverlayETag).Config {
			return refuse("the config is not the recovery's", nil)
		}
	case movesActionHeldRelease:
		if recovery.Release == "" {
			return refuse("no recorded recovery release", nil)
		}
		if id.Release != recovery.Release {
			return refuse("the release is not the recovery's", nil)
		}
	case movesActionSpawn:
		if recovery.Release == "" {
			return refuse("no recorded recovery release", nil)
		}
		if id.Config != HeldConfigIdentity(recovery.BundleETags, recovery.OverlayETag).Config || id.Release != recovery.Release {
			return refuse("the launch is not the recovery's", nil)
		}
	}
	return nil
}

// recoveryIdentityFits is movesIdentityFits for a recovery, with Plan 3
// H4's launch identity: a spawn names both the config and the release it
// launches with.
func recoveryIdentityFits(action string, id hostwrite.Identity) bool {
	if action == movesActionSpawn {
		return id.Config != "" && id.Release != ""
	}
	return movesIdentityFits(action, id)
}

// movesIdentityFits: a held-config write names the config and no release,
// a held-release write the release and no config, and every other action
// carries no identity at all. Which config and release is the record's
// business, checked after this.
func movesIdentityFits(action string, id hostwrite.Identity) bool {
	switch action {
	case movesActionHeldConfig:
		return id.Config != "" && id.Release == ""
	case movesActionHeldRelease:
		return id.Release != "" && id.Config == ""
	default:
		return id == hostwrite.Identity{}
	}
}

// heldTimersReplacer is the landing's timers import (ruling G16):
// schedule.SelectedTimers over the registry, its Work selected as
// held-timers for the host, so under CARLOS_HOST_CLAIM the write runs
// inside the claim's selection and Policy.Moves is asked first. With the
// flag off hostwrite.Config.Select answers the legacy pass-through and the
// registry's own ReplaceTimers runs exactly as before. With the flag on and
// no selector it refuses (hostwrite.ErrUnavailable) and writes nothing.
func heldTimersReplacer(writes hostwrite.Config, reg *registry.Registry) func(ctx context.Context, host string, rows []registry.Timer) error {
	return func(ctx context.Context, host string, rows []registry.Timer) error {
		if ctx == nil || reg == nil {
			return hostwrite.ErrUnavailable
		}
		work, err := writes.Select(ctx, host, movesActionHeldTimers)
		if err != nil {
			return err
		}
		return schedule.SelectedTimers{Store: reg, Work: work, Context: ctx, Host: host}.ReplaceTimers(host, rows)
	}
}

// movesCommittedSpawn is the committed wake's spawn (Plan 3 Task 7): the
// ordinary launch identity, over a committed record of this move, accepted by
// the committed input check.
func movesCommittedSpawn(ctx context.Context, read MovesRecordReader, committed func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity, refuse func(string, error) error) error {
	if committed == nil {
		return refuse("the committed launch has no input check", nil)
	}
	if !recoveryIdentityFits(s.Action, id) {
		return refuse("the identity does not fit the action", nil)
	}
	if read == nil {
		return refuse("no record reader", nil)
	}
	rec, err := read(ctx, c.Holder.Account, c.Holder.App, c.Host)
	if err != nil {
		return refuse("the record cannot be read", err)
	}
	if rec == nil || rec.Host != c.Host || rec.Move == nil || rec.Move.ID != s.Op || rec.Move.Incarnation != c.Holder.Incarnation || rec.Move.State != provision.MoveCommitted {
		return refuse("the record is not this committed move", nil)
	}
	if err := committed(ctx, c, s, id); err != nil {
		return refuse("the launch is not the committed one", err)
	}
	return nil
}