package hostops

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"os"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/controlname"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/release"
)

// ErrControlKey is the data boundary's unconditional refusal of a canonical
// control object. It wraps ErrContract so existing callers treat it as an
// unavailable contract, never as pending work that may succeed on retry.
var ErrControlKey = fmt.Errorf("%w: canonical control object is writable only by the typed control writer", ErrContract)

const (
	MailGateKey       = "control/mail-resource-gate.json"
	registrationPurge = "purge"
)

// IsControlOnlyKey names every object the data boundary refuses to mutate,
// whatever the context carries. Malformed app shapes are refused by the
// classifier, not here, so this stays an exact-key predicate.
func IsControlOnlyKey(key string) bool {
	if key == MailGateKey || (strings.HasPrefix(key, WakeResetPrefix) || strings.HasPrefix(key, WakeResetArchivePrefix)) {
		return true
	}
	p := strings.Split(key, "/")
	if len(p) != 4 || p[0] != "apps" || !controlname.ValidAccount(p[1]) || !controlname.ValidApp(p[2]) {
		return false
	}
	switch p[3] {
	case "admission.json", "purge.json", "purge-request.json":
		return true
	}
	return false
}

// ControlWriter is the admission engine's private writer. Its raw store is
// a named unexported field; there is no accessor, no generic Put/Update and
// no mutate callback. Every exported method is one typed transition that
// validates its own precondition against a fresh read.
type ControlWriter struct {
	homeRegion string
	raw        release.Store
	pub        ed25519.PublicKey
	now        func() time.Time
}

// NewControlWriter is called once, by ConfigureClaimRuntime. A source census
// (TestClaimConsoleMutationInvariant step 5) rejects any other caller.
func NewControlWriter(raw release.Store, pub ed25519.PublicKey, now func() time.Time, homeRegion ...string) (*ControlWriter, error) {
	if raw == nil || len(pub) != ed25519.PublicKeySize || now == nil {
		return nil, ErrContract
	}
	home, err := consoleHomeRegion(homeRegion)
	if err != nil {
		return nil, err
	}
	return &ControlWriter{raw: raw, pub: pub, now: now, homeRegion: home}, nil
}

func (w *ControlWriter) ready() error {
	if w == nil || w.raw == nil || w.now == nil {
		return ErrContract
	}
	return nil
}

// updateAdmission is the single CAS primitive. Lower-case and receiver-only:
// it is reachable solely from the typed methods in this package.
func (w *ControlWriter) updateAdmission(ctx context.Context, before AdmissionObserved, mutate func(*AppAdmission) error) (AdmissionObserved, error) {
	if err := w.ready(); err != nil {
		return AdmissionObserved{}, err
	}
	next, err := nextAdmission(before, mutate)
	if err != nil {
		return AdmissionObserved{}, err
	}
	body, err := json.Marshal(next)
	if err != nil {
		return AdmissionObserved{}, err
	}
	etag, err := w.raw.Put(ctx, before.Key, body, before.ETag)
	if err != nil {
		return AdmissionObserved{}, err
	}
	return AdmissionObserved{Key: before.Key, ETag: etag, Value: next}, nil
}

// casRetries bounds retryAdmission. Registrations renew every 20s and each
// effect takes two CASes, so a lost race is routine; eight fresh attempts
// cover every realistic burst without turning a wedge into a spin.
const casRetries = 8

// retryAdmission re-reads and re-applies mutate after a lost CAS. Use it only
// for transitions whose mutate re-validates everything it depends on and is
// idempotent against its own earlier success (settle, renew, bind, phase,
// reopen, restore receipt, drain removal). Initial transitions that choose a
// fresh identity or bind a saved claim observation (RegisterStart,
// BeginExclusive, ReservePurge, the generation-only claim CAS) stay single
// CAS: their caller decides what a conflict means. before may be zero, in
// which case the first attempt reads.
func (w *ControlWriter) retryAdmission(ctx context.Context, account, app, appID string, before AdmissionObserved, mutate func(*AppAdmission) error) (AdmissionObserved, error) {
	var err error
	if before.ETag == "" {
		if before, err = w.read(ctx, account, app, appID); err != nil {
			return AdmissionObserved{}, err
		}
	}
	for attempt := 1; ; attempt++ {
		next, err := w.updateAdmission(ctx, before, mutate)
		if err == nil || !errors.Is(err, release.ErrCASConflict) || attempt >= casRetries {
			return next, err
		}
		select {
		case <-ctx.Done():
			return AdmissionObserved{}, errors.Join(err, ctx.Err())
		case <-time.After(time.Duration(attempt) * 15 * time.Millisecond):
		}
		if before, err = w.read(ctx, account, app, appID); err != nil {
			return AdmissionObserved{}, err
		}
	}
}

func (w *ControlWriter) read(ctx context.Context, account, app, appID string) (AdmissionObserved, error) {
	if err := w.ready(); err != nil {
		return AdmissionObserved{}, err
	}
	return ReadAdmission(ctx, w.raw, account, app, appID)
}

func (w *ControlWriter) purgeAbsent(ctx context.Context, account, app string) error {
	key, err := PurgeKey(account, app)
	if err != nil {
		return err
	}
	if _, _, err = w.raw.Get(ctx, key); errors.Is(err, os.ErrNotExist) {
		return nil
	} else if err != nil {
		return err
	}
	return ErrPending
}

// ---- registrations --------------------------------------------------------

// StartHandle is the in-memory capability for one registration this process
// created. Knowing a registration's owner tuple from the bucket is not
// enough to issue effects under it; the handle is.
type StartHandle struct {
	w                       *ControlWriter
	moveStart               moveStartSlot
	account, app, appID, id string
	owner                   RegistrationOwner
	epoch                   uint64
	// kind and incarnation are the registration's: a reset-wake-retry
	// grant is bound to them (review R2).
	kind, incarnation string
}

func (h *StartHandle) ID() string               { return h.id }
func (h *StartHandle) AppID() string            { return h.appID }
func (h *StartHandle) Epoch() uint64            { return h.epoch }
func (h *StartHandle) Owner() RegistrationOwner { return h.owner }

func (w *ControlWriter) RegisterStart(ctx context.Context, account, app, appID, id string, r Registration, now time.Time) (*StartHandle, error) {
	if r.Kind == registrationPurge || r.Kind == RegistrationIngest {
		return nil, ErrIdentity
	}
	if err := w.purgeAbsent(ctx, account, app); err != nil {
		return nil, err
	}
	before, err := w.read(ctx, account, app, appID)
	if err != nil {
		return nil, err
	}
	if r.AppID != appID || r.Epoch != before.Value.Epoch {
		return nil, ErrIdentity
	}
	if _, err = w.updateAdmission(ctx, before, func(a *AppAdmission) error { return registerStart(a, id, r, now) }); err != nil {
		return nil, err
	}
	return &StartHandle{w: w, account: account, app: app, appID: appID, id: id, owner: r.Owner, epoch: r.Epoch, kind: r.Kind, incarnation: r.Incarnation}, nil
}

// Bind and Renew retry a lost CAS: bindStart is idempotent for the same
// (created, incarnation) and both re-check owner, epoch and liveness.
func (h *StartHandle) Bind(ctx context.Context, created time.Time, incarnation string, now time.Time) error {
	_, err := h.w.retryAdmission(ctx, h.account, h.app, h.appID, AdmissionObserved{}, func(a *AppAdmission) error {
		return bindStart(a, h.id, h.owner, created, incarnation, now)
	})
	return err
}

func (h *StartHandle) Renew(ctx context.Context, now time.Time) error {
	_, err := h.w.retryAdmission(ctx, h.account, h.app, h.appID, AdmissionObserved{}, func(a *AppAdmission) error {
		return renewStart(a, h.id, h.owner, now)
	})
	return err
}

type startHandleKey struct{}

// WithStartHandle carries the request's own registration capability to
// ProductionAuthority.RecordEffects. Only a writer-minted handle exists.
func WithStartHandle(ctx context.Context, h *StartHandle) context.Context {
	return context.WithValue(ctx, startHandleKey{}, h)
}

// startHandleFor returns the context's handle only for the exact
// registration and incarnation asked for.
func startHandleFor(ctx context.Context, account, app, appID, id string) (*StartHandle, bool) {
	h, ok := ctx.Value(startHandleKey{}).(*StartHandle)
	if !ok || h == nil || h.account != account || h.app != app || h.appID != appID || h.id != id {
		return nil, false
	}
	return h, true
}

func (h *StartHandle) Runner(now func() time.Time) EffectRunner {
	return EffectRunner{w: h.w, account: h.account, app: h.app, appID: h.appID, operation: h.id, owner: h.owner, now: now}
}

// ---- exclusive operations -------------------------------------------------

// OperationHandle is the capability for one exclusive app operation. It is
// minted only by BeginExclusive, ResumeExclusive and BeginPurge.
type OperationHandle struct {
	w                   *ControlWriter
	account, app, appID string
	id, kind            string
	owner               RegistrationOwner
}

func (h *OperationHandle) ID() string               { return h.id }
func (h *OperationHandle) Kind() string             { return h.kind }
func (h *OperationHandle) AppID() string            { return h.appID }
func (h *OperationHandle) Account() string          { return h.account }
func (h *OperationHandle) App() string              { return h.app }
func (h *OperationHandle) Owner() RegistrationOwner { return h.owner }

func (h *OperationHandle) Runner(now func() time.Time) EffectRunner {
	return EffectRunner{w: h.w, account: h.account, app: h.app, appID: h.appID, operation: h.id, owner: h.owner, now: now, exclusive: true}
}

func (w *ControlWriter) BeginExclusive(ctx context.Context, account, app, appID string, op AppOperation, owner RegistrationOwner) (*OperationHandle, error) {
	switch op.Kind {
	case OpDeleting, OpPlacing, OpAppMutation:
	default:
		return nil, ErrIdentity
	}
	if !validRegistrationOwner(owner) || op.AppID != appID {
		return nil, ErrIdentity
	}
	if err := w.purgeAbsent(ctx, account, app); err != nil {
		return nil, err
	}
	before, err := w.read(ctx, account, app, appID)
	if err != nil {
		return nil, err
	}
	if op.Kind == OpAppMutation {
		// A joined mutation additionally waits for every registration and
		// effect to settle: it conservatively serializes app mutations.
		if before.Value.Op == nil && (len(before.Value.Registrations) != 0 || !effectsSettled(&before.Value, "")) {
			return nil, ErrPending
		}
	}
	if _, err = w.updateAdmission(ctx, before, func(a *AppAdmission) error { return exclusiveAdmission(a, op) }); err != nil {
		return nil, err
	}
	return &OperationHandle{w: w, account: account, app: app, appID: appID, id: op.ID, kind: op.Kind, owner: owner}, nil
}

func (w *ControlWriter) ResumeExclusive(ctx context.Context, account, app, appID, opID, kind string, owner RegistrationOwner) (*OperationHandle, error) {
	switch kind {
	case OpDeleting, OpPlacing, OpPurging:
	default:
		return nil, ErrIdentity
	}
	if !validRegistrationOwner(owner) || opID == "" {
		return nil, ErrIdentity
	}
	a, err := w.read(ctx, account, app, appID)
	if err != nil {
		return nil, err
	}
	if a.Value.State != "closed" || a.Value.Op == nil || a.Value.Op.ID != opID || a.Value.Op.Kind != kind {
		return nil, ErrPending
	}
	return &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: kind, owner: owner}, nil
}

func (h *OperationHandle) current(ctx context.Context) (AdmissionObserved, error) {
	a, err := h.w.read(ctx, h.account, h.app, h.appID)
	if err != nil {
		return AdmissionObserved{}, err
	}
	if a.Value.State != "closed" || a.Value.Op == nil || a.Value.Op.ID != h.id || a.Value.Op.Kind != h.kind {
		return AdmissionObserved{}, ErrPending
	}
	return a, nil
}

func (h *OperationHandle) mine(a *AppAdmission) error {
	if a.State != "closed" || a.Op == nil || a.Op.ID != h.id || a.Op.Kind != h.kind || a.Op.AppID != h.appID {
		return ErrPending
	}
	return nil
}

// Reserved intent keys only the typed methods below write.
var reservedIntent = []string{"action", "drained.", "account", "app"}

// SetPhase advances this operation's own phase and intent. It cannot touch
// identity, registrations, effects, last_purge or last_restore.
func (h *OperationHandle) SetPhase(ctx context.Context, from, to string, set map[string]string) error {
	if to == "" {
		return ErrIdentity
	}
	for k := range set {
		for _, r := range reservedIntent {
			if k == r || (strings.HasSuffix(r, ".") && strings.HasPrefix(k, r)) {
				return ErrIdentity
			}
		}
	}
	a, err := h.current(ctx)
	if err != nil {
		return err
	}
	_, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
		if err := h.mine(v); err != nil {
			return err
		}
		if v.Op.Phase != from {
			if v.Op.Phase == to && intentHas(v.Op.Intent, set) {
				return nil
			}
			return ErrPending
		}
		v.Op.Phase = to
		if v.Op.Intent == nil {
			v.Op.Intent = map[string]string{}
		}
		for k, val := range set {
			v.Op.Intent[k] = val
		}
		return nil
	})
	return err
}

func intentHas(have, want map[string]string) bool {
	for k, v := range want {
		if have[k] != v {
			return false
		}
	}
	return true
}

// finish reopens admission. Spec §5.9: every reopen advances the epoch, so
// no permit, registration or receipt check bound to the closed epoch can
// act in the reopened one. bumpEpoch is kept for call-site clarity and is
// always true.
func (h *OperationHandle) finish(ctx context.Context, check func(*AppAdmission) error, bumpEpoch bool) error {
	bumpEpoch = true
	a, err := h.current(ctx)
	if err != nil {
		return err
	}
	// Retried: h.mine fails once the reopen landed (the retry then reports
	// ErrPending, never a second epoch bump).
	_, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
		if err := h.mine(v); err != nil {
			return err
		}
		if len(v.Registrations) != 0 || !effectsSettled(v, "") {
			return ErrPending
		}
		if check != nil {
			if err := check(v); err != nil {
				return err
			}
		}
		v.Op = nil
		v.State = "open"
		if bumpEpoch {
			v.Epoch++
		}
		return nil
	})
	return err
}

// FinishDelete reopens with a new epoch: no old registration can renew into
// the next incarnation's admission.
func (h *OperationHandle) FinishDelete(ctx context.Context) error {
	if h.kind != OpDeleting {
		return ErrIdentity
	}
	return h.finish(ctx, nil, true)
}

func (h *OperationHandle) FinishPlacement(ctx context.Context) error {
	if h.kind != OpPlacing {
		return ErrIdentity
	}
	return h.finish(ctx, func(v *AppAdmission) error {
		if v.Op.Phase != "releasing" {
			return ErrPending
		}
		return nil
	}, false)
}

func (h *OperationHandle) FinishMutation(ctx context.Context) error {
	if h.kind != OpAppMutation {
		return ErrIdentity
	}
	return h.finish(ctx, func(v *AppAdmission) error {
		if r := v.LastRestore; r != nil && r.ID == h.id && r.State != "complete" {
			return ErrPending
		}
		return nil
	}, false)
}

// ---- restore receipts (review P2-3) ----------------------------------------

// PrepareRestore persists the receipt, bound to the exact tombstone ETag,
// BEFORE the conditional tombstone delete.
func (h *OperationHandle) PrepareRestore(ctx context.Context, tombstoneETag string, deletedAt int64) error {
	if h.kind != OpAppMutation || tombstoneETag == "" || deletedAt <= 0 {
		return ErrIdentity
	}
	a, err := h.current(ctx)
	if err != nil {
		return err
	}
	if a.Value.Op.Intent["action"] != "restore" {
		return ErrIdentity
	}
	_, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
		if err := h.mine(v); err != nil {
			return err
		}
		if r := v.LastRestore; r != nil && r.ID == h.id {
			if r.TombstoneETag == tombstoneETag && r.DeletedAt == deletedAt {
				return nil
			}
			return ErrIdentity
		}
		v.LastRestore = &RestoreMark{ID: h.id, AppID: h.appID, Epoch: v.Epoch, TombstoneETag: tombstoneETag, DeletedAt: deletedAt, State: "prepared"}
		return nil
	})
	return err
}

// CompleteRestore marks the receipt complete once the exact tombstone
// version is gone. A different current tombstone is a refusal.
func (h *OperationHandle) CompleteRestore(ctx context.Context) error {
	if h.kind != OpAppMutation {
		return ErrIdentity
	}
	return h.w.completeRestore(ctx, h.account, h.app, h.appID, h.id, false)
}

func (w *ControlWriter) completeRestore(ctx context.Context, account, app, appID, opID string, finishDelete bool) error {
	a, err := w.read(ctx, account, app, appID)
	if err != nil {
		return err
	}
	r := a.Value.LastRestore
	if a.Value.Op == nil || a.Value.Op.ID != opID || a.Value.Op.Kind != OpAppMutation || r == nil || r.ID != opID || r.AppID != appID || r.Epoch != a.Value.Epoch {
		return ErrPending
	}
	if r.State == "complete" {
		return nil
	}
	key := release.AppTombstoneKey(account, app)
	_, tag, gerr := w.raw.Get(ctx, key)
	switch {
	case errors.Is(gerr, os.ErrNotExist):
	case gerr != nil:
		return gerr
	case tag == r.TombstoneETag && finishDelete:
		// Recovery finishes the user's restore rather than abandoning it:
		// the exact-ETag delete is idempotent and cannot erase a successor.
		if err = release.DeleteIfMatch(ctx, w.raw, key, r.TombstoneETag); err != nil {
			return err
		}
	case tag == r.TombstoneETag:
		return ErrPending
	default:
		// No trash can run while this operation holds the app.
		return ErrIdentity
	}
	// Retried: a concurrent settle or renewal must not leave a deleted
	// marker behind an incomplete receipt.
	_, err = w.retryAdmission(ctx, account, app, appID, a, func(v *AppAdmission) error {
		if v.Op == nil || v.Op.ID != opID || v.LastRestore == nil || v.LastRestore.ID != opID || v.LastRestore.TombstoneETag != r.TombstoneETag {
			return ErrPending
		}
		v.LastRestore.State = "complete"
		return nil
	})
	return err
}

// recoveryOwner stands in for an issuer when recovery itself records the
// outcome of an effect the issuer never prepared. It is not a registration
// owner any handle can hold, so no runner can advance or reuse the effect.
func recoveryOwner(opID string) RegistrationOwner {
	return RegistrationOwner{Console: "recovery", Enrollment: "recovery", Controller: opID}
}

// recoveredMutation reports that opID's app-mutation already ended with a
// settled outcome (another console's recovery, or the issuer, reopened it).
func recoveredMutation(v AppAdmission, opID string) bool {
	if v.Op != nil && v.Op.ID == opID {
		return false
	}
	if r := v.LastRestore; r != nil && r.ID == opID {
		e, ok := v.Effects[opID]
		return r.State == "complete" && ok && e.State == "complete"
	}
	e, ok := v.Effects[opID]
	return ok && e.OperationID == opID && e.State == "complete" && effectsSettled(&v, opID)
}

// RecoverMutation is the only automatic way to reopen an abandoned
// app-mutation. It never re-runs a backend call. Three shapes:
//
//   - restore with a saved receipt (any effect state, including no effect:
//     the receipt precedes Issue): the bucket decides. The exact-ETag delete
//     is finished, the op's effect is recorded complete (created when the
//     issuer never prepared it, which also fences a late Issue: it finds a
//     complete effect and gets EffectResult, never a permit), then reopen.
//     A receipt-less restore stays pending here; AbandonRestore owns it.
//   - mail: reopens only once the shared gate no longer holds this op's
//     gate ID AND every effect of the op is complete (a merely prepared one
//     is superseded). An issued or uncertain effect keeps the op; the gate
//     reconciler (ReconcileMailGate) settles those first.
//   - anything else (generic): reopens when nothing of the op is issued or
//     uncertain, superseding its merely prepared effects in the same CAS
//     (abandonUnissuedMutation). Unless the op's own effect is already
//     complete, the op must be at least UnissuedMutationRecoverAfter old,
//     so recovery does not race a live issuer between BeginExclusive and
//     its issued CAS. That age is liveness only; the safety argument is
//     the epoch-advancing reopen, which EffectRunner.Issue's issued CAS
//     re-checks (authorized), so a late issuer is fenced.
//
// An issued or uncertain generic effect is never settled here, by age or by
// readback: it is an activation dependency that needs a quiescence protocol.
//
// It is idempotent: an op that some other path already reopened with a
// settled outcome returns nil.
func (w *ControlWriter) RecoverMutation(ctx context.Context, account, app, appID, opID string) error {
	a, err := w.read(ctx, account, app, appID)
	if err != nil {
		return err
	}
	if a.Value.Op == nil || a.Value.Op.ID != opID {
		if recoveredMutation(a.Value, opID) {
			return nil
		}
		return ErrPending
	}
	if a.Value.State != "closed" || a.Value.Op.Kind != OpAppMutation || len(a.Value.Registrations) != 0 {
		return ErrPending
	}
	switch a.Value.Op.Intent["action"] {
	case "restore":
		if r := a.Value.LastRestore; r != nil && r.ID == opID {
			return w.recoverRestoreReceipt(ctx, account, app, appID, opID)
		}
		// No receipt: nothing may have been issued yet, or the issuer is
		// between BeginExclusive and PrepareRestore. AbandonRestore decides.
		return ErrPending
	case "mail":
		return w.recoverMailMutation(ctx, account, app, appID, opID, a.Value.Op.Intent["gate"])
	}
	return w.abandonUnissuedMutation(ctx, account, app, appID, opID, UnissuedMutationRecoverAfter)
}

// MutationActionStarted is the intent key RunAppMutation sets at
// BeginExclusive (unix seconds). It is the same key restore uses.
const MutationActionStarted = RestoreActionStarted

// UnissuedMutationRecoverAfter is how long the sweep leaves a generic
// app-mutation with no settled effect to its issuer. An op without the
// started stamp (a lost-response begin from an older build) is eligible at
// once: the reopen fences it either way.
const UnissuedMutationRecoverAfter = time.Minute

// abandonUnissued is the issuer's own recovery after Issue failed: f never
// ran, so no age bound applies.
func (h *OperationHandle) abandonUnissued(ctx context.Context) error {
	if h.kind != OpAppMutation {
		return ErrIdentity
	}
	return h.w.abandonUnissuedMutation(ctx, h.account, h.app, h.appID, h.id, 0)
}

// abandonUnissuedMutation reopens a generic app-mutation (not restore, not
// mail: those keep their specialized recovery) in one CAS when every effect
// is complete or one of this op's merely prepared effects, which it
// supersedes. An issued or uncertain effect of any operation, a
// registration, or a restore receipt refuses. The reopen advances the epoch,
// so the op's issuer can no longer issue (EffectRunner.authorized), and a
// superseded effect fails its prepared->issued CAS.
func (w *ControlWriter) abandonUnissuedMutation(ctx context.Context, account, app, appID, opID string, minAge time.Duration) error {
	if err := w.ready(); err != nil {
		return err
	}
	now := w.now()
	_, err := w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
		if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || v.Op.AppID != appID || len(v.Registrations) != 0 {
			return ErrPending
		}
		switch v.Op.Intent["action"] {
		case "restore", "mail":
			return ErrPending
		}
		if r := v.LastRestore; r != nil && r.ID == opID {
			return ErrPending
		}
		own, ok := v.Effects[opID]
		if minAge > 0 && !(ok && own.OperationID == opID && own.AppID == appID && own.State == "complete") {
			sec, perr := strconv.ParseInt(v.Op.Intent[MutationActionStarted], 10, 64)
			if perr == nil && sec > 0 && now.Before(time.Unix(sec, 0).Add(minAge)) {
				return ErrPending
			}
		}
		for id, e := range v.Effects {
			switch {
			case e.State == "complete":
			case e.OperationID == opID && e.AppID == appID && e.State == "prepared":
				e.State, e.ResultReference = "complete", "superseded"
				v.Effects[id] = e
			default:
				return ErrPending // issued, uncertain, or another operation's
			}
		}
		if v.Epoch == math.MaxUint64 {
			return ErrPending
		}
		v.Op, v.State = nil, "open"
		v.Epoch++
		return nil
	})
	return err
}

func (w *ControlWriter) recoverRestoreReceipt(ctx context.Context, account, app, appID, opID string) error {
	if err := w.sqlChildrenSettled(ctx, account, app, appID, opID); err != nil {
		return err
	}
	if err := w.completeRestore(ctx, account, app, appID, opID, true); err != nil {
		return err
	}
	if _, err := w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
		r := v.LastRestore
		if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || r == nil || r.ID != opID || r.State != "complete" {
			return ErrPending
		}
		for id, e := range v.Effects {
			if e.OperationID != opID || e.State == "complete" || id == opID {
				continue
			}
			if e.State != "prepared" {
				// An issued/uncertain effect under the restore op that is
				// not the receipt-bound one: its outcome is not decidable
				// from the bucket.
				return ErrPending
			}
			e.State, e.ResultReference = "complete", "superseded"
			v.Effects[id] = e
		}
		e, ok := v.Effects[opID]
		switch {
		case !ok:
			v.Effects[opID] = Effect{ID: opID, OperationID: opID, AppID: appID, Resource: OpAppMutation, Action: "restore",
				Inputs: map[string]string{"tombstone_etag": r.TombstoneETag}, Owner: recoveryOwner(opID),
				State: "complete", ResultReference: "restore-reconciled"}
		case e.OperationID != opID || e.AppID != appID:
			return ErrIdentity
		case e.State != "complete":
			// prepared (never called), issued or uncertain: the call's only
			// external effect is the receipt-bound delete, now decided.
			e.State, e.ResultReference = "complete", "restore-reconciled"
			v.Effects[opID] = e
		}
		return nil
	}); err != nil {
		return err
	}
	h := &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: OpAppMutation}
	return h.finish(ctx, func(v *AppAdmission) error {
		if r := v.LastRestore; r == nil || r.ID != opID || r.State != "complete" {
			return ErrPending
		}
		return nil
	}, true)
}

// recoverMailMutation ends a mail member operation whose gate was released
// or reconciled. Never while the gate still holds this gate ID: ops end only
// after the gate's release CAS.
func (w *ControlWriter) recoverMailMutation(ctx context.Context, account, app, appID, opID, gateID string) error {
	if err := w.sqlChildrenSettled(ctx, account, app, appID, opID); err != nil {
		return err
	}
	if gateID == "" {
		return ErrIdentity
	}
	g, _, err := w.readMailGate(ctx)
	if err != nil {
		return err
	}
	if g.State == "held" && g.ID == gateID {
		return ErrPending
	}
	if _, err = w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
		if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || v.Op.Intent["gate"] != gateID {
			return ErrPending
		}
		for id, e := range v.Effects {
			if e.OperationID != opID || e.State == "complete" {
				continue
			}
			if e.State != "prepared" {
				return ErrPending // issued or uncertain: ReconcileMailGate first
			}
			// A prepared effect never called: its prepared->issued CAS now
			// fails, and the gate it needed is gone anyway.
			e.State, e.ResultReference = "complete", "superseded"
			v.Effects[id] = e
		}
		return nil
	}); err != nil {
		return err
	}
	h := &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: OpAppMutation}
	return h.finish(ctx, nil, true)
}

// ---- drains and settlement --------------------------------------------------

type drainOutcome int

const (
	drainKeep drainOutcome = iota
	drainRemove
)

// judgeRegistration decides whether one saved registration can no longer
// produce an effect. It may write a generation-only claim CAS that
// invalidates the saved observation. recoverCreate is the caller's own
// controller hook for a start that already owns the host.
func (w *ControlWriter) judgeRegistration(ctx context.Context, a *AppAdmission, account, app, id string, reg Registration, now time.Time, recoverCreate func(context.Context, string) error) (drainOutcome, SavedClaim, error) {
	if !effectsSettled(a, id) {
		return drainKeep, SavedClaim{}, nil
	}
	if reg.Kind == RegistrationIngest {
		if len(a.Effects) > 0 {
			for _, e := range a.Effects {
				if e.OperationID == id {
					return drainRemove, SavedClaim{}, nil
				}
			}
		}
		if RegistrationDrained(reg, now) {
			return drainRemove, SavedClaim{}, nil
		}
		return drainKeep, SavedClaim{}, nil
	}
	if reg.Kind == registrationPurge || reg.Kind == RegistrationWakeRetryReset {
		if RegistrationDrained(reg, now) {
			return drainRemove, SavedClaim{}, nil
		}
		return drainKeep, SavedClaim{}, nil
	}
	claim, err := hostclaim.Read(ctx, w.raw, reg.Host)
	if err != nil {
		return drainKeep, SavedClaim{}, err
	}
	if !claim.Absent && claim.Claim.Op.ID == id {
		if claim.Claim.Op.Kind == hostclaim.KindCreating && recoverCreate != nil {
			if err = recoverCreate(ctx, reg.Host); err != nil && !errors.Is(err, ErrPending) {
				return drainKeep, SavedClaim{}, err
			}
		}
		// An accepted start owns the host until its own recovery settles.
		return drainKeep, SavedClaim{}, nil
	}
	h := claim.Claim
	completed := !claim.Absent && h.Holder != nil && h.Holder.Account == account && h.Holder.App == app && h.Holder.Incarnation == reg.Incarnation && h.Op.Kind == hostclaim.KindServing
	cancelled := !claim.Absent && reg.Incarnation != "" && slices.Contains(h.RetiredIncarnations, reg.Incarnation)
	if completed || cancelled {
		return drainRemove, SavedObservation(claim), nil
	}
	if !RegistrationDrained(reg, now) {
		return drainKeep, SavedClaim{}, nil
	}
	if SavedObservation(claim) == reg.Claim {
		// Generation-only CAS; for prior absence this writes a free claim.
		// Never Delete(claim).
		if claim, err = hostclaim.Update(ctx, w.raw, claim, func(*hostclaim.Claim) error { return nil }); err != nil {
			return drainKeep, SavedClaim{}, err
		}
	}
	if SavedObservation(claim) == reg.Claim {
		return drainKeep, SavedClaim{}, nil
	}
	return drainRemove, SavedObservation(claim), nil
}

// Drain removes every saved registration that can no longer act, journaling
// each witness into the operation intent in the same CAS. ErrPending while
// any registration or effect is still live.
func (h *OperationHandle) Drain(ctx context.Context, now time.Time, recoverCreate func(context.Context, string) error) error {
	if h.kind != OpDeleting && h.kind != OpPlacing {
		return ErrIdentity
	}
	a, err := h.current(ctx)
	if err != nil {
		return err
	}
	for id, reg := range a.Value.Registrations {
		out, witness, err := h.w.judgeRegistration(ctx, &a.Value, h.account, h.app, id, reg, now, recoverCreate)
		if err != nil {
			return err
		}
		if out != drainRemove {
			return ErrPending
		}
		evidence, err := json.Marshal(struct {
			Registration Registration `json:"registration"`
			Witness      SavedClaim   `json:"witness"`
		}{reg, witness})
		if err != nil {
			return err
		}
		fresh, err := h.current(ctx)
		if err != nil {
			return err
		}
		if a, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, fresh, func(v *AppAdmission) error {
			if err := h.mine(v); err != nil {
				return err
			}
			saved, ok := v.Registrations[id]
			if !ok {
				return nil
			}
			if saved != reg {
				return ErrPending
			}
			if v.Op.Intent == nil {
				v.Op.Intent = map[string]string{}
			}
			v.Op.Intent["drained."+id] = string(evidence)
			delete(v.Registrations, id)
			return nil
		}); err != nil {
			return err
		}
	}
	if len(a.Value.Registrations) != 0 || !effectsSettled(&a.Value, "") {
		return ErrPending
	}
	return nil
}

// SettleRegistration removes one finished registration while no exclusive
// operation exists (the recovery sweep).
func (w *ControlWriter) SettleRegistration(ctx context.Context, account, app, appID, id string, now time.Time) error {
	a, err := w.read(ctx, account, app, appID)
	if err != nil {
		return err
	}
	reg, ok := a.Value.Registrations[id]
	if !ok {
		return nil
	}
	if a.Value.Op != nil {
		return ErrPending
	}
	out, _, err := w.judgeRegistration(ctx, &a.Value, account, app, id, reg, now, nil)
	if err != nil {
		return err
	}
	if out != drainRemove {
		return ErrPending
	}
	_, err = w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
		if v.Op != nil || v.Registrations[id] != reg || !effectsSettled(v, id) {
			return ErrPending
		}
		delete(v.Registrations, id)
		return nil
	})
	return err
}