package console

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfence"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
)

// The console's move pass (moves Plan 5 Task 7, spec §7.5, §7.10, §7.11,
// §13; Paul's ruling M4). It is the moves body of claim Plan 10's
// Recovery.Moves hook: the recovery pass (ControlWriter.RecoveryPass) calls
// movesRecovery once per claim of a move kind, with a MoveRecoveryGrant bound
// to that claim, its move registration and the exact record it observed.
// ConfigureClaimRuntime binds it only with the moves switch on.
//
// Each call reads the claim and the record again through the console's store,
// gives up if either moved on since the grant's observation (the next pass
// observes afresh), decides at most one action (decideMoveRecovery) and takes
// it. The grant is spent by its first write, so one observation is never two
// writes. In the order the decision tries them:
//
//  1. An orphan: a claim of a move kind whose record carries neither the move
//     nor its outcome. A moving claim before the seal (the request door's
//     record write was lost) is left for ten minutes, then its start effect
//     is settled without the request's handle (ResolveStart, K3), and on a
//     later pass, with the start settled not applied, released
//     (release-not-applied). Any other orphan is paged, once, and left: an
//     operation past the seal with no record is an incident.
//  2. A source terminated before the flip (K13): Convert with the source's
//     termination fence, which writes the record's relocation decision on
//     one pass and converts the claim to relocating on the next.
//  3. The six-hour abort (K18): a move still preparing moves.AbortAfter
//     after its request is aborted through ConsoleMove(ConsoleMoveAbort);
//     the source's own abort (movesabort.go) takes it from there.
//  4. The claim half the record decided (§7.5, Review Focus 5): flip,
//     commit or rollback, with the target's fence between the flip and the
//     rest (below). Anything else is left alone.
//  5. The deadline: a landing past the earlier of its cut and landing
//     deadlines is rolled back by the record (J11, J9's recorded reason).
//  6. A lost target released with nothing of its box or region owed:
//     target_parked_at, through the
//     record-sequenced recovery grant (StampLostMoveTargetParked)
//     (p10 Task G).
//  7. A close whose stamps are in and whose op owes nothing: the record's
//     close (close, rolled-back-close or revert-close). The boxes close on
//     their own in the pass the close becomes due; the console writes it
//     only when it has been due for moveCloseGrace, and pages that it did.
//     An abort is never closed here (fix round 1, ruling 1): only the
//     source closes it, after it retired its own snapshots, receipts,
//     ledger and journal (movesabort.go), and nothing on the record shows
//     that it did. An abort due past the lane's horizon (the claim's
//     quiet_after for the target region) plus moveCloseGrace is paged
//     once, and left to the source.
//  8. The release: the record closed (LastMove names the op) and the claim
//     not yet released. This is the second half of every record-first close
//     (ruling M4); after a box crashed between the halves, only this pass
//     writes it, because no box visits a record without a move.
//
// The target's fence (moves after claim Plan 12, Task 5, ruling Q8). A
// landing claim before its commit, or a restoring claim, whose target box was
// terminated has its regional worker's termination fence at
// control/box-fences/<target enrollment>.json. The grant's
// ReleaseFencedTarget takes one step of it per call, record first: on a
// landing it rolls the record back with the target lost under the fence
// (RollbackTargetFenced), or on a rollback marks the target lost; then the
// claim's release, which rolls a landing claim back and consumes the fence in
// one write; then the record's loss commit (released, the fence as its
// evidence). The fence comes after the flip (the grant refuses a moving
// claim) and ahead of the claim's rollback half and the deadline's rollback,
// because the grant writes both halves itself. Once the claim carries the
// consumed fence, the reference is the claim's (ConsumedMoveTargetFence),
// never the bucket's; and once the record carries it as both its loss and its
// evidence, the pass stops calling it. A dirty target (one that released with
// writes the source does not have) cannot be finished from a dead box: the
// grant refuses it after it verified the fence, the pass pages once that an
// operator must declare the target lost, and leaves it for J4. The page rests
// only on the grant's verified-dirty outcome, never on a bare ErrPending. Its fence
// comes after the claim's rollback half, which the grant does not take for
// it. The console logs the fence's record writes, and its claim rollback, as
// its own move writes, from the grant's outcome (the write's own evidence).
//
// Pages (§13) go through alerting.Page on the sinks cmd/console arms, best
// effort: a rollback waiting past its release deadline (an incident, again
// only a full hour after the last attempt), a move preparing for over an
// hour, an orphan past the seal (an incident), an overdue close the console
// wrote, an overdue abort the console leaves to its source, and a dirty
// fenced target left for J4 (an incident) (each once).
// The rate limit is durable: the last attempt's time is kept per finding in
// the console's store (move_pages, fix round 1 ruling 3), written whether or
// not delivery worked, so a restarted console keeps the spacing.
//
// A relocating claim is region HA's, and the pass does nothing for it.

const (
	// moveOrphanAfter is how old, by Op.StartedAt, a moving claim with no
	// record move must be before the pass settles its start.
	moveOrphanAfter = 10 * time.Minute
	// movePreparingPage is how long a move may prepare before it pages
	// (§13: 60 minutes with no promotable round).
	movePreparingPage = time.Hour
	// moveCloseGrace is how long a close must have been due before the
	// console writes it instead of the boxes.
	moveCloseGrace = 10 * time.Minute
)

type moveRecoveryAction string

const (
	moveActNone         moveRecoveryAction = ""
	moveActResolveStart moveRecoveryAction = "resolve-start"
	moveActConvert      moveRecoveryAction = "convert"
	moveActAbort        moveRecoveryAction = "abort"
	moveActClaim        moveRecoveryAction = "claim"
	moveActRecord       moveRecoveryAction = "record"
	moveActStampLost    moveRecoveryAction = "stamp-lost"
	// moveActFence is one step of the target's termination fence, through
	// the grant's ReleaseFencedTarget (Q8).
	moveActFence moveRecoveryAction = "fence"
	moveActClose moveRecoveryAction = "close"
	// moveActAbortOverdue writes nothing: an abort due to close, which only
	// its source may close, is checked for being overdue and paged.
	moveActAbortOverdue moveRecoveryAction = "abort-overdue"
)

type movePageKind string

const (
	movePageReleaseDeadline movePageKind = "release-deadline"
	movePagePreparing       movePageKind = "preparing"
	movePageOrphan          movePageKind = "orphan"
	movePageCloseOverdue    movePageKind = "close-overdue"
	movePageAbortOverdue    movePageKind = "abort-overdue"
	// movePageTargetFencedDirty is a dirty target whose fence the grant
	// refused: left for J4 (Q8).
	movePageTargetFencedDirty movePageKind = "target-fenced-dirty"
)

// moveRecoveryInput is what one observation saw. Start is the op's move-start
// effect, read only for a moving claim with no record move; Fence is the
// source's termination fence, read only for a moving claim. TargetFence is
// the target's (Q8), read only for a landing claim before its commit or a
// restoring claim: the claim's consumed fence when it carries one
// (TargetFenceConsumed), otherwise the bucket's.
type moveRecoveryInput struct {
	Now                 time.Time
	Claim               *hostclaim.Claim
	Rec                 provision.Record
	Start               *hostops.Effect
	Fence               *hostclaim.FenceRef
	TargetFence         *hostclaim.FenceRef
	TargetFenceConsumed bool
}

// moveRecoveryDecision is at most one action, with the grant action it
// names, and the pages the observation calls for.
type moveRecoveryDecision struct {
	Action moveRecoveryAction
	Claim  hostops.MoveRecoveryClaimAction
	Record hostops.MoveRecoveryRecordAction
	Pages  []movePageKind
}

// decideMoveRecovery is the pass's whole policy, on values (see the file
// comment for the order).
func decideMoveRecovery(in moveRecoveryInput) moveRecoveryDecision {
	var d moveRecoveryDecision
	c := in.Claim
	if c == nil || c.Holder == nil || c.Op.ID == "" || c.Op.Kind == hostclaim.KindRelocating {
		return d
	}
	op, rec, m := c.Op, in.Rec, in.Rec.Move
	sealed := op.Kind == hostclaim.KindMoving && op.Phase == hostclaim.PhaseSealed
	if m == nil || m.Legacy() || m.ID != op.ID {
		if m != nil {
			return d // another move's record: not this op's to settle
		}
		if rec.LastMove != nil && rec.LastMove.ID == op.ID {
			d.Action, d.Claim = moveActClaim, hostops.RecoveryMoveRelease
			return d
		}
		if op.Kind != hostclaim.KindMoving || op.Phase != hostclaim.PhaseMoving {
			d.Pages = append(d.Pages, movePageOrphan)
			return d
		}
		if op.StartedAt.IsZero() || in.Now.Before(op.StartedAt.Add(moveOrphanAfter)) || in.Start == nil {
			return d
		}
		switch {
		case in.Start.State == "uncertain":
			d.Action = moveActResolveStart
		case in.Start.State == "complete" && in.Start.ResultReference == "not-applied":
			d.Action, d.Claim = moveActClaim, hostops.RecoveryMoveReleaseNotApplied
		}
		return d
	}
	if m.Incarnation != c.Holder.Incarnation {
		return d
	}
	rb := m.Rollback
	if m.State == provision.MoveRollingBack && rb != nil && rb.ReleaseDeadline != nil && in.Now.After(*rb.ReleaseDeadline) && rb.TargetPhase == provision.RbTargetReleasing {
		d.Pages = append(d.Pages, movePageReleaseDeadline)
	}
	if m.State == provision.MovePreparing && !m.RequestedAt.IsZero() && !in.Now.Before(moves.PreparingSince(m).Add(movePreparingPage)) {
		d.Pages = append(d.Pages, movePagePreparing)
	}
	owed := owedUnderOp(c, op.ID, nil)
	fence, fenceDirty := targetFenceStep(in)

	switch {
	case op.Kind == hostclaim.KindMoving && in.Fence != nil && (m.State == provision.MovePreparing || m.State == provision.MoveReady || m.State == provision.MoveCutting || m.State == provision.MoveRelocating):
		d.Action = moveActConvert
	case op.Kind == hostclaim.KindMoving && op.Phase == hostclaim.PhaseMoving && moves.AbortDue(rec, in.Now):
		d.Action = moveActAbort
	case m.State == provision.MoveLanding && m.FlippedAt != nil && rec.Region == m.To && sealed:
		d.Action, d.Claim = moveActClaim, hostops.RecoveryMoveFlip
	case fence && !fenceDirty:
		d.Action = moveActFence
	case m.State == provision.MoveCommitted && m.CommittedAt != nil && m.FlippedAt != nil && rec.Region == m.To && op.Kind == hostclaim.KindLanding && op.Phase == hostclaim.PhaseLandingProbed:
		d.Action, d.Claim = moveActClaim, hostops.RecoveryMoveCommit
	case m.State == provision.MoveRollingBack && rb != nil && rec.Region == m.From && (op.Kind == hostclaim.KindLanding && op.Phase != hostclaim.PhaseCommitted || sealed):
		d.Action, d.Claim = moveActClaim, hostops.RecoveryMoveRollback
	case m.State == provision.MoveLanding && op.Kind == hostclaim.KindLanding && op.Phase != hostclaim.PhaseCommitted && moveDeadlinePassed(m, in.Now):
		d.Action, d.Record = moveActRecord, hostops.RecoveryMoveDeadlineRollback
	case fence:
		// A dirty target: after the claim's rollback half, which the grant
		// does not take for it. The grant verifies the fence and refuses it;
		// the pass pages (movesRecovery).
		d.Action = moveActFence
	case m.State == provision.MoveRollingBack && rb != nil && rb.TargetPhase == provision.RbTargetReleased && !rb.Dirty && rb.TargetParkedAt == nil &&
		hostclaim.LostMoveTargetParked(c, park.PlacementRegion(m.To)):
		d.Action = moveActStampLost
	case owed > 0:
	case m.State == provision.MoveCommitted && m.OpenedAt != nil && m.SourceParkedAt != nil && rec.Region == m.To && op.Kind == hostclaim.KindLanding && op.Phase == hostclaim.PhaseCommitted:
		d.Action, d.Record = moveActClose, hostops.RecoveryMoveClose
	case m.State == provision.MoveRollingBack && rb != nil && rb.RecoveredAt != nil && rb.OpenedAt != nil && rb.TargetParkedAt != nil && rec.Region == m.From && op.Kind == hostclaim.KindRestoring && op.Phase == hostclaim.PhaseOpened:
		d.Action, d.Record = moveActClose, hostops.RecoveryMoveRolledBackClose
	case m.State == provision.MoveReverting && m.ReopenedAt != nil && rec.Region == m.From && op.Kind == hostclaim.KindReverting && op.Phase == hostclaim.PhaseOpened:
		d.Action, d.Record = moveActClose, hostops.RecoveryMoveRevertClose
	case m.State == provision.MoveAborting && op.Kind == hostclaim.KindMoving && op.Phase == hostclaim.PhaseAborting:
		d.Action = moveActAbortOverdue
	}
	return d
}

// targetFenceStep reports whether the target's fence (in.TargetFence) has a
// step left for the grant's ReleaseFencedTarget, and whether the step is a
// dirty target's, which the grant refuses and the pass leaves for J4 (Q8).
// The record's regions are compared as the grant compares them, code with
// code.
//
// Consumed (the claim's release is done), the one step left is the record's
// loss commit: the target lost under this fence. Once the record carries the
// fence as its loss and its evidence, or the target released on its own
// evidence, the fence is spent. Not yet consumed, a step is one the grant
// would take: a landing's rollback; a rollback whose target still holds
// something the source needs (releasing, returning, or released dirty) is
// marked lost; the target lost under this fence, the claim's release; or a
// target that released clean while the claim still waits on it
// (restoring/releasing under the target), the claim's release. A target
// lost under another loss (J4's declaration) is not the fence's.
func targetFenceStep(in moveRecoveryInput) (step, dirty bool) {
	c, rec, ref := in.Claim, in.Rec, in.TargetFence
	if ref == nil || c == nil || rec.Move == nil {
		return false, false
	}
	op, m := c.Op, rec.Move
	rb := m.Rollback
	if in.TargetFenceConsumed {
		return op.Kind == hostclaim.KindRestoring && m.State == provision.MoveRollingBack && rb != nil && rec.Region == m.From &&
			rb.TargetPhase == provision.RbTargetLost && rb.LossRef == ref.Key && !rb.Dirty, false
	}
	if !(op.Kind == hostclaim.KindLanding && op.Phase != hostclaim.PhaseCommitted) && op.Kind != hostclaim.KindRestoring {
		return false, false
	}
	if m.State == provision.MoveLanding {
		return op.Kind == hostclaim.KindLanding && m.FlippedAt != nil && rec.Region == m.To, false
	}
	if m.State != provision.MoveRollingBack || rb == nil || rec.Region != m.From {
		return false, false
	}
	switch rb.TargetPhase {
	case provision.RbTargetReleasing, provision.RbTargetReturning:
		step = true
	case provision.RbTargetReleased:
		step = rb.Dirty || op.Kind == hostclaim.KindRestoring && op.Phase == hostclaim.PhaseReleasing && op.Exec == op.Intent["target"]
	case provision.RbTargetLost:
		step = rb.LossRef == ref.Key
	}
	return step, step && rb.Dirty
}

// moveDeadlinePassed is the grant's deadline rule: the earlier of the cut and
// landing deadlines (J11, provision.LandingDeadlineOf, the box's own rule) is
// behind now.
func moveDeadlinePassed(m *provision.MoveState, now time.Time) bool {
	deadline, _ := provision.LandingDeadlineOf(m)
	return deadline != nil && now.After(*deadline)
}

// owedUnderOp counts the claim's obligations under op id, those match
// accepts when it is set.
func owedUnderOp(c *hostclaim.Claim, id string, match func(hostclaim.Obligation) bool) int {
	n := 0
	for _, o := range c.Obligations {
		if o.Op == id && (match == nil || match(o)) {
			n++
		}
	}
	return n
}

func (s *Server) movesNow() time.Time {
	if s.movesClock != nil {
		return s.movesClock().UTC()
	}
	return time.Now().UTC()
}

// movesRecovery is the Recovery.Moves hook (see the file comment).
func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant) error {
	a := s.cfg.ClaimAuthority
	store := s.cfg.Releases
	if a == nil || a.Control == nil || store == nil || g == nil {
		return hostops.ErrContract
	}
	o := g.Observation()
	obs, err := hostclaim.Read(ctx, store, o.Host)
	if err != nil {
		return err
	}
	if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.ETag != o.ClaimETag || obs.Claim.Op.ID != o.OpID {
		return nil // moved on since the observation: the next pass observes again
	}
	c := obs.Claim
	r, err := hostops.ReadRecord(ctx, store, c.Holder.Key)
	if err != nil {
		return err
	}
	if r.ETag != o.RecordETag {
		return nil
	}
	rec, err := provision.DecodeRecord(r.Body)
	if err != nil {
		return err
	}
	now := s.movesNow()
	in := moveRecoveryInput{Now: now, Claim: c, Rec: rec}
	ours := rec.Move != nil && rec.Move.ID == o.OpID
	if c.Op.Kind == hostclaim.KindMoving && c.Op.Phase == hostclaim.PhaseMoving && !ours {
		adm, err := hostops.ReadAdmission(ctx, store, o.Account, o.App, o.AppID)
		if err != nil {
			return err
		}
		if e, ok := adm.Value.Effects[o.OpID]; ok {
			in.Start = &e
		}
	}
	if c.Op.Kind == hostclaim.KindMoving && ours {
		if in.Fence, err = s.sourceFence(ctx, c); err != nil {
			return err
		}
	}
	if ours && (c.Op.Kind == hostclaim.KindLanding && c.Op.Phase != hostclaim.PhaseCommitted || c.Op.Kind == hostclaim.KindRestoring) {
		// Once consumed, the fence is the claim's, never the bucket's (Q8).
		if ref, ok := hostclaim.ConsumedMoveTargetFence(c); ok {
			in.TargetFence, in.TargetFenceConsumed = &ref, true
		} else if in.TargetFence, err = s.bucketFence(ctx, c.Op.Intent["target_enrollment"]); err != nil {
			return err
		}
	}
	d := decideMoveRecovery(in)
	for _, kind := range d.Pages {
		s.pageMove(ctx, kind, o, c, rec, now)
	}
	switch d.Action {
	case moveActResolveStart:
		out, err := g.ResolveStart(ctx)
		if errors.Is(err, hostops.ErrMoveStartFenceUnproven) {
			log.Printf("console: moves: %s: the start of move %s is uncertain and its record has not advanced past it; left", o.Host, o.OpID)
			return nil
		}
		if err == nil {
			log.Printf("console: moves: %s: the start of move %s settled %s", o.Host, o.OpID, out)
		}
		return err
	case moveActConvert:
		return g.Convert(ctx, *in.Fence)
	case moveActAbort:
		co, err := hostops.ObserveConsoleMove(ctx, store, o.Account, o.App, o.Host)
		if err != nil {
			return err
		}
		if co != o {
			return nil
		}
		if _, err := a.Control.ConsoleMove(ctx, co, hostops.ConsoleMoveAbort); err != nil {
			return err
		}
		log.Printf("console: moves: %s: move %s is still preparing after %s; aborted", o.Host, o.OpID, moves.AbortAfter)
		if t, ok := abortTransition(rec); ok {
			s.queueMoveLog(o, t, now)
		}
		return nil
	case moveActClaim:
		if d.Claim == hostops.RecoveryMoveRelease {
			// claim_released is published only from a release that
			// happened: queued unconfirmed now, confirmed once the release
			// succeeds (a crash between the two is settled by the drainer,
			// from the claim).
			s.queueMoveLogAs(o, claimTransition(o, rec, "claim_released", "claim-released"), now, false)
			if f := s.moveReleaseFault; f != nil {
				if err := f(); err != nil {
					return err
				}
			}
		}
		if err := g.Claim(ctx, d.Claim); err != nil {
			return err
		}
		switch d.Claim {
		case hostops.RecoveryMoveRelease:
			if err := s.store.ConfirmMoveLog(o.Host, o.OpID, "claim-released"); err != nil {
				log.Printf("console: moves: %s: the release of move %s is not confirmed in its log outbox (the drainer settles it from the claim): %v", o.Host, o.OpID, err)
			}
		case hostops.RecoveryMoveRollback:
			s.queueMoveLog(o, claimTransition(o, rec, "claim_rollback", "claim-rollback"), now)
		}
		return nil
	case moveActRecord:
		if err := g.Record(ctx, d.Record); err != nil {
			return err
		}
		if t, ok := g.Transition(); ok {
			s.queueMoveLog(o, t, now)
		}
		return nil
	case moveActStampLost:
		return g.StampLostMoveTargetParked(ctx)
	case moveActFence:
		if f := s.moveFenceCall; f != nil {
			f(*in.TargetFence)
		}
		out, err := g.ReleaseFencedTarget(ctx, *in.TargetFence)
		if out.VerifiedDirty && errors.Is(err, hostops.ErrPending) {
			// Only the grant's verified-dirty outcome pages (fix round 1,
			// item 1): it is given after the fence verified. Any other
			// ErrPending (a withdrawn registration, a spent grant) is not
			// evidence of a terminated target. J4, the operator's
			// declaration, takes it from here.
			s.pageMove(ctx, movePageTargetFencedDirty, o, c, rec, now)
			return nil
		}
		if err != nil {
			return err
		}
		// The step's own writes, from the grant's evidence (fix round 1,
		// item 2), never from a later read of the record.
		if out.Transition != nil {
			s.queueMoveLog(o, *out.Transition, now)
		}
		if out.ClaimRolledBack {
			s.queueMoveLog(o, claimTransition(o, rec, "claim_rollback", "claim-rollback"), now)
		}
		return nil
	case moveActClose:
		if !s.moveCloseOverdue(o, d.Record, now) {
			return nil
		}
		if err := g.Record(ctx, d.Record); err != nil {
			return err
		}
		log.Printf("console: moves: %s: move %s closed by the console (%s); the release follows", o.Host, o.OpID, d.Record)
		if t, ok := g.Transition(); ok {
			s.queueMoveLog(o, t, now)
		}
		s.pageMove(ctx, movePageCloseOverdue, o, c, rec, now)
		return nil
	case moveActAbortOverdue:
		if s.moveAbortOverdue(o, park.QuietUntil(c.QuietAfter, s.cfg.dbClaimHomeRegion(), park.PlacementRegion(rec.Move.To)), now) {
			s.pageMove(ctx, movePageAbortOverdue, o, c, rec, now)
		}
		return nil
	}
	return nil
}

// sourceFence is the moving source's termination fence, when one exists:
// the reference to control/box-fences/<source enrollment>.json and its exact
// bytes. Convert verifies it; nothing here trusts it.
func (s *Server) sourceFence(ctx context.Context, c *hostclaim.Claim) (*hostclaim.FenceRef, error) {
	return s.bucketFence(ctx, c.Op.Intent["source_enrollment"])
}

// bucketFence is the termination fence of the box enrolled as enrollment,
// when one exists in the bucket: its reference and its exact bytes' digest.
// The grant verifies it; nothing here trusts it.
func (s *Server) bucketFence(ctx context.Context, enrollment string) (*hostclaim.FenceRef, error) {
	if enrollment == "" {
		return nil, nil
	}
	key := hostfence.FenceKey(enrollment)
	body, _, err := s.cfg.Releases.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	sum := sha256.Sum256(body)
	return &hostclaim.FenceRef{Enrollment: enrollment, Key: key, SHA256: hex.EncodeToString(sum[:])}, nil
}

// moveDueSince is when the pass first saw this close due, remembering now
// if it had not. The memory is the process's: a restarted console waits the
// grace again, which only delays the console's close or page. Entries a day
// old are dropped.
func (s *Server) moveDueSince(key string, now time.Time) time.Time {
	s.moveCloseMu.Lock()
	defer s.moveCloseMu.Unlock()
	if s.moveCloseSeen == nil {
		s.moveCloseSeen = map[string]time.Time{}
	}
	for k, at := range s.moveCloseSeen {
		if now.Sub(at) > 24*time.Hour {
			delete(s.moveCloseSeen, k)
		}
	}
	first, ok := s.moveCloseSeen[key]
	if !ok {
		s.moveCloseSeen[key] = now
		return now
	}
	return first
}

// moveCloseOverdue reports whether this close has been due for
// moveCloseGrace; when it has, the memory is cleared for the close.
func (s *Server) moveCloseOverdue(o hostops.ConsoleMoveObservation, action hostops.MoveRecoveryRecordAction, now time.Time) bool {
	key := o.Host + "|" + o.OpID + "|" + string(action)
	if now.Sub(s.moveDueSince(key, now)) < moveCloseGrace {
		return false
	}
	s.moveCloseMu.Lock()
	delete(s.moveCloseSeen, key)
	s.moveCloseMu.Unlock()
	return true
}

// moveAbortOverdue reports whether an abort with nothing owed has stayed
// unclosed past its lane's horizon (quiet, the claim's quiet_after for the
// target region, when the claim has one) plus moveCloseGrace, and has been
// seen due for moveCloseGrace itself. The page it leads to is once per move
// (pageMove), so the memory is kept.
func (s *Server) moveAbortOverdue(o hostops.ConsoleMoveObservation, quiet, now time.Time) bool {
	since := s.moveDueSince(o.Host+"|"+o.OpID+"|abort", now)
	if now.Sub(since) < moveCloseGrace {
		return false
	}
	return quiet.IsZero() || !now.Before(quiet.Add(moveCloseGrace))
}

// pageMove sends one page for a finding (its kind, host and move) unless
// the finding was paged before: ever, for a once-per-move page, or within
// the last full hour, for the release deadline. The attempt's time is
// recorded whether or not delivery worked (pages are best effort).
func (s *Server) pageMove(ctx context.Context, kind movePageKind, o hostops.ConsoleMoveObservation, c *hostclaim.Claim, rec provision.Record, now time.Time) {
	rule := string(kind) + " " + o.Host + " " + o.OpID
	level := alerting.LevelWarning
	every := time.Duration(0) // once per move
	var subject, body string
	switch kind {
	case movePageReleaseDeadline:
		level, every = alerting.LevelIncident, time.Hour
		subject, body = moveWaitingPage(o.Host, o.OpID, rec.Move.To)
	case movePagePreparing:
		subject, body = movePreparingPageText(o.Host, o.OpID, rec.Move.To)
	case movePageOrphan:
		level = alerting.LevelIncident
		subject, body = moveOrphanPage(o.Host, o.OpID)
	case movePageCloseOverdue:
		subject, body = moveCloseOverduePage(o.Host, o.OpID)
	case movePageAbortOverdue:
		subject, body = moveAbortOverduePage(o.Host, o.OpID, rec.Move.To, c.Op.Intent["source"])
	case movePageTargetFencedDirty:
		level = alerting.LevelIncident
		subject, body = moveTargetFencedDirtyPage(o.Host, o.OpID, c.Op.Intent["target"], moves.RegionText(rec.Move.To), moves.RegionText(rec.Move.From))
	default:
		return
	}
	if s.store == nil {
		return
	}
	last, paged, err := s.store.MovePageAttempt(rule)
	if err != nil {
		log.Printf("console: moves: %s: the page record for %s: %v", o.Host, rule, err)
		return
	}
	if paged && (every == 0 || now.Sub(last) < every) {
		return
	}
	if err := s.store.SetMovePageAttempt(rule, now); err != nil {
		log.Printf("console: moves: %s: record the page %s: %v", o.Host, rule, err)
		return
	}
	if err := alerting.Page(ctx, level, subject, body); err != nil {
		log.Printf("console: moves: %s: page %s: %v", o.Host, kind, err)
	}
}

// moveWaitingPage is a rollback waiting past its release deadline (§7.8
// waiting): the subject (MV-N325) and the body (MV-N326).
func moveWaitingPage(host, id, to string) (subject, body string) {
	subject = fmt.Sprintf("carlos move waiting for %s to release: %s", to, host)
	body = fmt.Sprintf("The move %s of %s is rolling back, and %s has not released it by its deadline. %s stays offline until a box in %s confirms it has stopped, or that box is declared lost.",
		id, host, to, host, to)
	return subject, body
}

// movePreparingPageText is a move preparing for over an hour (§13): the
// subject (MV-N327) and the body (MV-N328).
func movePreparingPageText(host, id, to string) (subject, body string) {
	subject = fmt.Sprintf("carlos move preparing for over an hour: %s", host)
	body = fmt.Sprintf("The move %s of %s to %s has been preparing for over an hour. It aborts itself six hours after the request if %s is still not ready.",
		id, host, to, to)
	return subject, body
}

// moveOrphanPage is a claim held by a move the record does not carry (K3):
// the subject (MV-N329) and the body (MV-N330).
func moveOrphanPage(host, id string) (subject, body string) {
	subject = fmt.Sprintf("carlos move without a record: %s", host)
	body = fmt.Sprintf("The host claim of %s is held by the move %s, and the instance record carries no move. Nothing releases it on its own. An operator must check the record.",
		host, id)
	return subject, body
}

// moveCloseOverduePage is a close the console wrote because no box did: the
// subject (MV-N331) and the body (MV-N332).
func moveCloseOverduePage(host, id string) (subject, body string) {
	subject = fmt.Sprintf("carlos move close overdue: %s", host)
	body = fmt.Sprintf("The move %s of %s was ready to close for more than 10 minutes and no box closed it. The console has closed it.", id, host)
	return subject, body
}

// moveAbortOverduePage is an abort that owes nothing and that its source has
// not closed (fix round 1, ruling 1): the subject (MV-N333) and the body
// (MV-N334).
func moveAbortOverduePage(host, id, to, box string) (subject, body string) {
	subject = fmt.Sprintf("carlos move abort not closed: %s", host)
	body = fmt.Sprintf("The abort of the move %s of %s has nothing left to park in %s, and %s has not closed it. %s closes it once it has removed the move's snapshots from its own bucket. Check %s.",
		id, host, to, box, box, box)
	return subject, body
}

// moveTargetFencedDirtyPage is a target terminated before it returned the
// writes it made, whose fence the grant refuses (Q8, left for J4): the
// subject (MV-N450) and the body (MV-N451). to and from are as a sentence
// names them (moves.RegionText).
func moveTargetFencedDirtyPage(host, id, box, to, from string) (subject, body string) {
	subject = fmt.Sprintf("carlos move target terminated before returning its writes: %s", host)
	body = fmt.Sprintf("%s in %s was terminated after it wrote data it had not returned to %s. the move %s of %s cannot finish on its own. run carlos move resolve --host %s --declare-lost %s to move %s back to %s.",
		box, to, from, id, host, host, box, host, from)
	return subject, body
}

// The move log of the console's own move writes (Plan 6 Task 3; fix round 3,
// ruling 1): the closes, the landing deadline's rollback, the six-hour abort,
// and the claim's rollback and release. Each entry is named as a box's write
// of the same transition would be (moves.TransitionOf), from the write's own
// evidence (the grant's Transition, or the record the pass acted on), and
// carries the console as its actor; the console has no member sink.
//
// The release's own entry (claim_released) is published only from a release
// that happened (ruling 2): queued unconfirmed before g.Claim and confirmed
// after it succeeds; the drainer settles a row left unconfirmed (a failed
// release, or a crash before the confirm) from the claim. A declared loss of
// a move's target (declared_lost) is the console's too, from either door
// that performs the declaration, in the same two phases (final review I4,
// declareMoveLossLogged); the drainer settles its unconfirmed row from the
// loss the claim records (settleDeclaredLost).
//
// An entry is queued in the console's outbox (move_log_outbox) with its
// exact bytes and its report prefix, and the recovery pass's drainer
// (drainMoveLog) delivers it, once a pass, whatever its move has done since:
// independent of any grant (spent by the time the entry exists) and of the
// claim's release, which never waits for the log. A row whose write fails
// is retried with backoff, indefinitely, and warned of once in the journal
// when it is an hour old. A confirmed row the writer refuses outright is
// held, never deleted: kept, not retried and warned of once, until its
// incarnation is retired or purged (PurgeMoveLogIncarnation). A release of a start never applied is not logged:
// the record never named that move.

// moveLogWarnAfter is the age at which an undelivered entry is warned of.
const moveLogWarnAfter = time.Hour

// moveLogDrainBatch bounds one pass's deliveries.
const moveLogDrainBatch = 256

// moveLogBackoff is the wait after a row's attempts-th failure: 30 seconds,
// doubling, at most an hour.
func moveLogBackoff(attempts int) time.Duration {
	d := 30 * time.Second
	for i := 1; i < attempts && d < time.Hour; i++ {
		d *= 2
	}
	return min(d, time.Hour)
}

// moveLogEntry is the console's entry for t.
func moveLogEntry(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) ([]byte, error) {
	return json.Marshal(moves.LogEvent{V: 1, Op: t.Op, Host: o.Host, Event: t.Event, Identity: t.Identity, Actor: "console", At: now.UTC(), Fields: t.Fields})
}

// queueMoveLog queues the entry for t, for the drainer.
func (s *Server) queueMoveLog(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) {
	s.queueMoveLogAs(o, t, now, true)
}

// queueMoveLogAs queues the entry for t, confirmed or not: the drainer
// delivers a confirmed entry, and settles an unconfirmed one from the claim
// (settleUnconfirmed). The row is bound to the incarnation t's evidence
// carried, which must be the grant's move and instance.
func (s *Server) queueMoveLogAs(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time, confirmed bool) {
	if err := s.queueMoveLogErr(o, t, now, confirmed); err != nil {
		log.Printf("console: moves: %s: the log of move %s at %s is lost: %v", o.Host, o.OpID, t.Identity, err)
	}
}

// queueMoveLogErr is queueMoveLogAs, answering whether the row is queued:
// for a caller that must not act unless it is (declareMoveLossLogged).
func (s *Server) queueMoveLogErr(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time, confirmed bool) error {
	if t.Op != o.OpID || t.Incarnation != o.Incarnation {
		return fmt.Errorf("the entry names move %s of incarnation %q, not move %s of %q", t.Op, t.Incarnation, o.OpID, o.Incarnation)
	}
	body, err := moveLogEntry(o, t, now)
	if err != nil {
		return err
	}
	return s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, t.Incarnation, body, confirmed, now)
}

// claimRootKey is the root public key the claim runtime signs with, which
// the move loss challenges and records verify under; nil without the
// runtime.
func (s *Server) claimRootKey() ed25519.PublicKey {
	if a := s.cfg.ClaimAuthority; a != nil {
		return a.Signer.RootPublicKey()
	}
	return nil
}

// confirmMoveLog confirms t's entry once the write it names succeeded. The
// row is queued again, confirmed, in case a pass dropped the unconfirmed one
// in between (settleDeclaredLost drops a row whose loss is not committed
// yet), and then marked confirmed if it was there already.
func (s *Server) confirmMoveLog(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) {
	s.queueMoveLogAs(o, t, now, true)
	if err := s.store.ConfirmMoveLog(o.Host, o.OpID, t.Identity); err != nil {
		log.Printf("console: moves: %s: the log of move %s at %s is not confirmed in its log outbox (the drainer settles it): %v", o.Host, o.OpID, t.Identity, err)
	}
}

// declareMoveLossLogged runs one declaration of a move's lost target
// (declare, the configured authority's DeclareMoveLoss) with its move log
// entry, declared_lost (final review I4): queued unconfirmed before the
// declaration, from the record as the door reads it, and confirmed once the
// declaration succeeds. The declaration moves the target on from lost to
// released within the request, so the box's pass may never see it lost; the
// console that performed it is the entry's producer. A record that cannot be
// read, or a row that cannot be queued, refuses the declaration: an
// operator's action fails closed rather than leave no retry for its entry
// (final re-review 1, I4).
func (s *Server) declareMoveLossLogged(ctx context.Context, account, app, host string, sub hostfence.Submission, declare func() error) error {
	var (
		o      hostops.ConsoleMoveObservation
		t      moves.Transition
		logged bool
	)
	rec, err := provision.GetRecord(ctx, s.cfg.Releases, account, app, host)
	if err != nil {
		// Fail closed (final re-review 1, I4): a declaration whose entry
		// cannot be queued is not made.
		return fmt.Errorf("moves: the declared loss cannot be queued in the move log: the record cannot be read: %w", err)
	}
	// Only a declaration whose signed consequence is the loss of this move's
	// target is declared_lost; a destroyed-local-only one changes no move
	// state and logs nothing (final re-review 1, N1).
	if op, inc, ok := hostops.DeclaresMoveTargetLoss(s.claimRootKey(), sub); ok && rec.Move != nil && op == rec.Move.ID && inc == rec.Move.Incarnation {
		if t, logged = moves.DeclaredLost(rec.Move); logged {
			o = hostops.ConsoleMoveObservation{Host: host, OpID: rec.Move.ID, Incarnation: rec.Move.Incarnation}
			if err := s.queueMoveLogErr(o, t, s.movesNow(), false); err != nil {
				return fmt.Errorf("moves: the declared loss cannot be queued in the move log: %w", err)
			}
		}
	}
	if err := declare(); err != nil {
		return err
	}
	if logged {
		s.confirmMoveLog(o, t, s.movesNow())
	}
	return nil
}

// drainMoveLog is the Recovery.MoveLogs hook: every due row of the outbox,
// delivered through the pass's writer.
func (s *Server) drainMoveLog(ctx context.Context, lw *hostops.MoveLogWriter) error {
	now := s.movesNow()
	rows, err := s.store.MoveLogDue(now, moveLogDrainBatch)
	if err != nil {
		return fmt.Errorf("moves: the log outbox: %w", err)
	}
	for _, r := range rows {
		if err := ctx.Err(); err != nil {
			return err
		}
		if !r.Confirmed {
			deliver, wait := s.settleUnconfirmed(ctx, r)
			if wait != nil {
				s.retryMoveLog(r, now, wait)
			}
			if !deliver {
				continue
			}
		}
		s.deliverMoveLog(ctx, lw, r, now)
	}
	return nil
}

// settleUnconfirmed settles a row queued for a release that was not yet
// confirmed, from the host's claim: released for the row's move, it is
// confirmed (deliver); still the move's, or the claim unreadable, it waits
// (wait says why, and it takes a failed delivery's backoff and warning, so
// it rotates out of the batch); anything else, it is dropped with a journal
// line.
func (s *Server) settleUnconfirmed(ctx context.Context, r MoveLogPendingEntry) (deliver bool, wait error) {
	if r.Name == moves.DeclaredLostIdentity {
		return s.settleDeclaredLost(ctx, r)
	}
	o, err := hostclaim.Read(ctx, s.cfg.Releases, r.Host)
	if err != nil {
		return false, fmt.Errorf("unconfirmed, and the claim cannot be read: %w", err)
	}
	switch {
	case !o.Absent && o.Claim != nil && hostclaim.RequireReleasedMove(r.Op)(o.Claim) == nil:
		if err := s.store.ConfirmMoveLog(r.Host, r.Op, r.Name); err != nil {
			return false, fmt.Errorf("unconfirmed: %w", err)
		}
		return true, nil
	case !o.Absent && o.Claim != nil && o.Claim.Op.ID == r.Op:
		return false, errors.New("unconfirmed: the claim still holds the move")
	}
	log.Printf("console: moves: %s: the log of move %s at %s was never confirmed and the claim no longer holds the move; dropped", r.Host, r.Op, r.Name)
	if err := s.store.SettleMoveLog(r.Host, r.Op, r.Name); err != nil {
		log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err)
	}
	return false, nil
}

// settleDeclaredLost settles an unconfirmed declared_lost row (final review
// I4; final re-review 1) from the claim. A committed loss of the move's
// target, in the claim's loss_refs (which the declaration's claim CAS writes
// as it commits, and nothing removes), confirms the row (deliver). While the
// move can still be declared lost (the claim's op is still the move, not
// released to serving) the row is kept, however long, with the failed
// delivery's backoff and its one warning. It is dropped only once the claim
// shows the move closed or released with no such loss, or the instance's
// incarnation gone (no claim, or another holder); the incarnation's purge
// deletes it too (PurgeMoveLogIncarnation). The drop deletes only a row
// still unconfirmed, so a declaration that confirmed its row after the claim
// was read here keeps it. A claim or loss record that cannot be read waits.
func (s *Server) settleDeclaredLost(ctx context.Context, r MoveLogPendingEntry) (deliver bool, wait error) {
	o, err := hostclaim.Read(ctx, s.cfg.Releases, r.Host)
	if err != nil {
		return false, fmt.Errorf("unconfirmed, and the claim cannot be read: %w", err)
	}
	if f := s.moveLogSettleHook; f != nil {
		f()
	}
	drop := func(why string) (bool, error) {
		log.Printf("console: moves: %s: the log of move %s at %s was never confirmed and %s; dropped", r.Host, r.Op, r.Name, why)
		if err := s.store.DropUnconfirmedMoveLog(r.Host, r.Op, r.Name); err != nil {
			log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err)
		}
		return false, nil
	}
	if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.Incarnation != r.Incarnation {
		return drop("the claim no longer holds the instance it was queued for")
	}
	if len(o.Claim.LossRefs) > 0 {
		pub := s.claimRootKey()
		if len(pub) != ed25519.PublicKeySize {
			return false, errors.New("unconfirmed, and there is no key to read the claim's loss records")
		}
		for _, id := range o.Claim.LossRefs {
			l, err := hostclaim.ReadLossRecord(ctx, s.cfg.Releases, pub, id)
			if errors.Is(err, hostclaim.ErrLossRecordAbsent) || errors.Is(err, hostclaim.ErrLossRecord) {
				continue
			}
			if err != nil {
				return false, fmt.Errorf("unconfirmed, and a loss record cannot be read: %w", err)
			}
			// Only a loss of the move's target, signed as such, is the
			// row's evidence: a destroyed-local-only loss of the same op is
			// not (final re-review 1, N1).
			op, inc, target := hostops.MoveTargetLossOf(l.Record)
			if !target || op != r.Op || inc != r.Incarnation || l.Record.Host != r.Host || l.Record.OpID != r.Op {
				continue
			}
			if err := s.store.ConfirmMoveLog(r.Host, r.Op, r.Name); err != nil {
				return false, fmt.Errorf("unconfirmed: %w", err)
			}
			return true, nil
		}
	}
	if o.Claim.Op.ID == r.Op && o.Claim.Op.Kind != hostclaim.KindServing {
		return false, errors.New("unconfirmed: no loss of the move's target has committed, and the move can still be declared lost")
	}
	return drop("the claim shows the move closed or released with no loss of its target")
}

// retryMoveLog moves a row's next attempt on after one that did not settle
// it (why), and warns of it once in the journal when it is an hour old.
func (s *Server) retryMoveLog(r MoveLogPendingEntry, now time.Time, why error) {
	attempts, warned := r.Attempts+1, r.Warned
	if !warned && now.Sub(r.QueuedAt) >= moveLogWarnAfter {
		log.Printf("console: moves: warning: %s: the log of move %s at %s has been unwritten for over an hour: %v", r.Host, r.Op, r.Name, why)
		warned = true
	}
	if err := s.store.RetryMoveLog(r.Host, r.Op, r.Name, attempts, now.Add(moveLogBackoff(attempts)), warned); err != nil {
		log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err)
	}
}

// deliverMoveLog writes one row, and settles it or schedules its retry.
func (s *Server) deliverMoveLog(ctx context.Context, lw *hostops.MoveLogWriter, r MoveLogPendingEntry, now time.Time) {
	settle := func() {
		if err := s.store.SettleMoveLog(r.Host, r.Op, r.Name); err != nil {
			log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err)
		}
	}
	_, err := lw.WriteMoveLog(ctx, hostops.MoveLogEntry{Host: r.Host, Op: r.Op, Incarnation: r.Incarnation, Name: r.Name, Body: r.Body})
	switch {
	case err == nil:
		settle()
	case errors.Is(err, hostops.ErrMoveLogTaken):
		log.Printf("console: moves: %s: the log of move %s at %s was written by another writer first", r.Host, r.Op, r.Name)
		settle()
	case errors.Is(err, hostops.ErrIdentity):
		// The writer refuses it (its claim's incarnation or record no
		// longer vouches for it: a later move's close, a recreation, a
		// retirement outside the console's door). The row is never deleted
		// for that: it is held, its bytes kept and no longer retried, until
		// its incarnation's purge (final fix, ruling 1).
		log.Printf("console: moves: warning: %s: the log of move %s at %s is held: the writer refuses it (%v); its entry is kept until its incarnation is purged", r.Host, r.Op, r.Name, err)
		if herr := s.store.HoldMoveLog(r.Host, r.Op, r.Name); herr != nil {
			log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, herr)
		}
	default:
		s.retryMoveLog(r, now, err)
	}
}

// abortTransition is the six-hour abort's entry: the record the pass acted
// on, as the abort leaves it (aborting).
func abortTransition(rec provision.Record) (moves.Transition, bool) {
	if rec.Move == nil {
		return moves.Transition{}, false
	}
	after, m := rec, *rec.Move
	m.State = provision.MoveAborting
	after.Move = &m
	return moves.TransitionOf(rec, after)
}

// claimTransition is the entry for a claim write of the console's: the move's
// regions and incarnation from the record the pass acted on (a closed
// move's, from the grant's observation of the same instance).
func claimTransition(o hostops.ConsoleMoveObservation, rec provision.Record, event, identity string) moves.Transition {
	t := moves.Transition{Op: o.OpID, Event: event, Identity: identity, Incarnation: o.Incarnation}
	if m := rec.Move; m != nil {
		t.Incarnation = m.Incarnation
		t.Fields = map[string]string{"from": m.From, "to": m.To}
	}
	return t
}