rvbox/internal/client/spool/events.go

package spool

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"fmt"
	"time"
	"unicode/utf8"

	"github.com/klauspost/compress/zstd"
	rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
	"github.com/rvbox/rvbox/internal/domain"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type Command struct {
	IssueUUID       domain.UUID
	ImmutableSHA256 [32]byte
	Revision        uint64
	Phase           uint32
	Terminal        bool
	// ExecutionSpec is the deterministic protobuf payload received from the
	// server. It is retained as raw protobuf bytes so the runtime can validate
	// and execute exactly the admitted request after a restart.
	ExecutionSpec []byte
	// Script carries the immutable descriptor reservation for a script-backed
	// command. Its bytes arrive later through AppendScriptChunk; creating the
	// descriptor row in the same transaction as acceptance prevents an
	// accepted command from being left without upload state after a crash.
	Script *ScriptDescriptor
}

type Acceptance struct {
	Duplicate bool
	Command   Command
}

type EventInput struct {
	Kind        uint32
	Compression uint32
	RawBytes    uint64
	Payload     []byte
	Output      bool
	UseCloseout bool
	CreatedAt   time.Time
}

const (
	// These match the CommandEvent oneof field numbers, allowing the runtime to
	// reconstruct the envelope without a second event-kind translation table.
	EventKindOutput           uint32 = 5
	EventKindOutputTruncation uint32 = 10
)

type Event struct {
	IssueUUID    domain.UUID
	LocalOrdinal uint64
	EventSeq     uint64
	Kind         uint32
	Compression  uint32
	RawBytes     uint64
	Payload      []byte
	CreatedAt    time.Time
}

func (store *Store) AcceptCommand(ctx context.Context, command Command, acceptedAt time.Time) (Acceptance, error) {
	if !validUUID(command.IssueUUID) || command.Revision == 0 || command.Phase == 0 || command.Phase > 11 || command.Terminal != isTerminalPhase(command.Phase) || acceptedAt.IsZero() {
		return Acceptance{}, errors.New("invalid command acceptance")
	}
	var storedSpec []byte
	var specCharge, scriptCharge uint64
	if len(command.ExecutionSpec) > 0 {
		if uint64(len(command.ExecutionSpec)) > store.maxExecutionSpecBytes {
			return Acceptance{}, errors.New("execution specification exceeds client limit")
		}
		var err error
		storedSpec, err = compressExecutionSpec(command.ExecutionSpec)
		if err != nil {
			return Acceptance{}, err
		}
		specCharge, err = EstimateCharge(ChargeInput{EncodedBytes: uint64(len(storedSpec)), SQLiteRows: 1, IndexEntries: 1})
		if err != nil {
			return Acceptance{}, err
		}
	}
	var storedScript []byte
	if command.Script != nil {
		if err := validateAcceptedScript(command.ExecutionSpec, *command.Script, store.maxScriptBytes); err != nil {
			return Acceptance{}, err
		}
		var err error
		storedScript, err = compressScript(nil)
		if err != nil {
			return Acceptance{}, err
		}
		scriptCharge, err = EstimateCharge(ChargeInput{EncodedBytes: uint64(len(storedScript)), SQLiteRows: 1, IndexEntries: 1})
		if err != nil {
			return Acceptance{}, err
		}
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return Acceptance{}, err
	}
	defer tx.Rollback()
	existing, found, err := commandByUUID(ctx, tx, command.IssueUUID, store.maxExecutionSpecBytes)
	if err != nil {
		return Acceptance{}, err
	}
	if found {
		if existing.ImmutableSHA256 != command.ImmutableSHA256 {
			return Acceptance{}, ErrCommandConflict
		}
		if len(command.ExecutionSpec) > 0 && !bytes.Equal(existing.ExecutionSpec, command.ExecutionSpec) {
			return Acceptance{}, ErrCommandConflict
		}
		return Acceptance{Duplicate: true, Command: existing}, nil
	}
	var tombstoneHash []byte
	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, command.IssueUUID[:]).Scan(&tombstoneHash)
	if err == nil {
		if string(tombstoneHash) != string(command.ImmutableSHA256[:]) {
			return Acceptance{}, ErrCommandConflict
		}
		return Acceptance{}, ErrAlreadyExecuted
	}
	if err != sql.ErrNoRows {
		return Acceptance{}, err
	}
	baseCharge, err := EstimateCharge(ChargeInput{SQLiteRows: 1, IndexEntries: 2})
	if err != nil {
		return Acceptance{}, err
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return Acceptance{}, err
	}
	charge, overflow := addChecked(baseCharge, specCharge)
	if !overflow {
		charge, overflow = addChecked(charge, scriptCharge)
	}
	if overflow {
		return Acceptance{}, &CapacityError{Tier: CapacityTierHardMaximum, Requested: ^uint64(0), Available: store.quotaLimits.HardAllocationBytes}
	}
	decision, err := CheckReservation(store.quotaLimits, ReservationState{ClientTotalCharged: clientTotal, CloseoutRemaining: store.quotaLimits.CloseoutReserveBytes}, ReservationRequest{ChargedBytes: charge})
	if err != nil {
		return Acceptance{}, err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO commands(issue_uuid, immutable_sha256, command_revision, phase, terminal, base_charged_bytes, total_charged_bytes, closeout_remaining_bytes, accepted_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, command.IssueUUID[:], command.ImmutableSHA256[:], command.Revision, command.Phase, boolInt(command.Terminal), baseCharge, decision.CommandTotalCharged, decision.CloseoutRemaining, acceptedAt.UnixNano())
	if err != nil {
		return Acceptance{}, err
	}
	if err := updateClientTotalCharge(ctx, tx, decision.ClientTotalCharged); err != nil {
		return Acceptance{}, err
	}
	if len(storedSpec) > 0 {
		digest := immutableDigest(storedSpec)
		if _, err := tx.ExecContext(ctx, `INSERT INTO command_specs(issue_uuid, raw_bytes, stored_bytes, compression, payload, payload_sha256, charged_bytes) VALUES (?, ?, ?, 2, ?, ?, ?)`, command.IssueUUID[:], len(command.ExecutionSpec), len(storedSpec), storedSpec, digest[:], specCharge); err != nil {
			return Acceptance{}, err
		}
	}
	if command.Script != nil {
		if _, err := tx.ExecContext(ctx, `INSERT INTO scripts(issue_uuid, declared_raw_bytes, declared_sha256, stored_bytes, compression, stored_data, charged_bytes) VALUES (?, ?, ?, ?, 2, ?, ?)`, command.IssueUUID[:], command.Script.SizeBytes, command.Script.SHA256[:], len(storedScript), storedScript, scriptCharge); err != nil {
			return Acceptance{}, err
		}
	}
	if err := tx.Commit(); err != nil {
		return Acceptance{}, err
	}
	return Acceptance{Command: command}, nil
}

func validateAcceptedScript(encodedSpec []byte, descriptor ScriptDescriptor, maximum uint64) error {
	if descriptor.SizeBytes > maximum {
		return ErrScriptBounds
	}
	var spec rvboxv1.ExecutionSpec
	if err := proto.Unmarshal(encodedSpec, &spec); err != nil {
		return ErrScriptConflict
	}
	declared := spec.GetScript()
	if declared == nil || declared.GetSizeBytes() != descriptor.SizeBytes || len(declared.GetSha256()) != sha256.Size || !bytes.Equal(declared.GetSha256(), descriptor.SHA256[:]) {
		return ErrScriptConflict
	}
	return nil
}

// GetCommand returns the durable command metadata and its immutable execution
// specification. The returned protobuf bytes are a copy and can be decoded or
// modified by the runtime without changing the spool's source of truth.
func (store *Store) GetCommand(ctx context.Context, issueUUID domain.UUID) (Command, error) {
	if !validUUID(issueUUID) {
		return Command{}, ErrUnknownCommand
	}
	command, found, err := commandByUUID(ctx, store.db, issueUUID, store.maxExecutionSpecBytes)
	if err != nil {
		return Command{}, err
	}
	if !found {
		return Command{}, ErrUnknownCommand
	}
	command.ExecutionSpec = bytes.Clone(command.ExecutionSpec)
	return command, nil
}

// AppendEvent assigns command-local order only. EventSeq remains zero until
// AssignSendWindow makes the durable event eligible for a wire send.
func (store *Store) AppendEvent(ctx context.Context, issueUUID domain.UUID, input EventInput) (Event, error) {
	if !validUUID(issueUUID) || input.Kind == 0 || input.Compression < 1 || input.Compression > 2 || input.CreatedAt.IsZero() {
		return Event{}, errors.New("invalid client event")
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return Event{}, err
	}
	defer tx.Rollback()
	var nextOrdinal, outputCharged, totalCharged, closeout uint64
	err = tx.QueryRowContext(ctx, `SELECT next_local_ordinal, output_charged_bytes, total_charged_bytes, closeout_remaining_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&nextOrdinal, &outputCharged, &totalCharged, &closeout)
	if err == sql.ErrNoRows {
		return Event{}, ErrUnknownCommand
	}
	if err != nil {
		return Event{}, err
	}
	charge, err := EstimateCharge(ChargeInput{EncodedBytes: uint64(len(input.Payload)), SQLiteRows: 1, IndexEntries: 2})
	if err != nil {
		return Event{}, err
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return Event{}, err
	}
	decision, err := CheckReservation(store.quotaLimits, ReservationState{CommandOutputCharged: outputCharged, CommandTotalCharged: totalCharged, ClientTotalCharged: clientTotal, CloseoutRemaining: closeout}, ReservationRequest{ChargedBytes: charge, Output: input.Output, UseCloseout: input.UseCloseout})
	if err != nil {
		return Event{}, err
	}
	ordinal := nextOrdinal
	digest := immutableDigest(input.Payload)
	_, err = tx.ExecContext(ctx, `INSERT INTO events(issue_uuid, local_ordinal, event_kind, compression, raw_bytes, charged_bytes, output, payload, payload_sha256, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, issueUUID[:], ordinal, input.Kind, input.Compression, input.RawBytes, charge, boolInt(input.Output), input.Payload, digest[:], input.CreatedAt.UnixNano())
	if err != nil {
		return Event{}, err
	}
	if _, err = tx.ExecContext(ctx, `UPDATE commands SET next_local_ordinal = ?, output_charged_bytes = ?, total_charged_bytes = ?, closeout_remaining_bytes = ? WHERE issue_uuid = ?`, ordinal+1, decision.CommandOutputCharged, decision.CommandTotalCharged, decision.CloseoutRemaining, issueUUID[:]); err != nil {
		return Event{}, err
	}
	if err := updateClientTotalCharge(ctx, tx, decision.ClientTotalCharged); err != nil {
		return Event{}, err
	}
	if err := tx.Commit(); err != nil {
		return Event{}, err
	}
	return Event{IssueUUID: issueUUID, LocalOrdinal: ordinal, Kind: input.Kind, Compression: input.Compression, RawBytes: input.RawBytes, Payload: append([]byte(nil), input.Payload...), CreatedAt: input.CreatedAt}, nil
}

// AssignSendWindow atomically gives previously-unassigned events contiguous
// per-command wire sequences. Once assigned, they are immutable until Ack.
func (store *Store) AssignSendWindow(ctx context.Context, issueUUID domain.UUID, maximumEvents, maximumStoredBytes uint64) ([]Event, error) {
	if maximumEvents == 0 || maximumStoredBytes == 0 {
		return nil, nil
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	var next, lastAck, assignedBytes uint64
	err = tx.QueryRowContext(ctx, `SELECT next_event_seq, last_server_ack,
COALESCE((SELECT sum(length(payload)) FROM events
  WHERE issue_uuid = commands.issue_uuid AND event_seq IS NOT NULL AND event_seq > commands.last_server_ack), 0)
FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next, &lastAck, &assignedBytes)
	if err == sql.ErrNoRows {
		return nil, ErrUnknownCommand
	}
	if err != nil {
		return nil, err
	}
	// Assigned rows remain pinned until cumulative acknowledgement.  Never
	// assign another row while that durable send window is full; otherwise a
	// burst of local events could grow the pinned set without bound even though
	// callers pass a per-command byte limit on every invocation.
	if assignedBytes >= maximumStoredBytes {
		return eventsBySequence(ctx, tx, issueUUID, 0)
	}
	maximumStoredBytes -= assignedBytes
	rows, err := tx.QueryContext(ctx, `SELECT local_ordinal, length(payload) FROM events WHERE issue_uuid = ? AND event_seq IS NULL ORDER BY local_ordinal LIMIT ?`, issueUUID[:], maximumEvents)
	if err != nil {
		return nil, err
	}
	var ordinals []uint64
	for rows.Next() {
		var ordinal, storedBytes uint64
		if err := rows.Scan(&ordinal, &storedBytes); err != nil {
			_ = rows.Close()
			return nil, err
		}
		if storedBytes > maximumStoredBytes {
			break
		}
		maximumStoredBytes -= storedBytes
		ordinals = append(ordinals, ordinal)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	for _, ordinal := range ordinals {
		if _, err := tx.ExecContext(ctx, `UPDATE events SET event_seq = ? WHERE issue_uuid = ? AND local_ordinal = ? AND event_seq IS NULL`, next, issueUUID[:], ordinal); err != nil {
			return nil, err
		}
		next++
	}
	if len(ordinals) > 0 {
		if _, err := tx.ExecContext(ctx, `UPDATE commands SET next_event_seq = ? WHERE issue_uuid = ?`, next, issueUUID[:]); err != nil {
			return nil, err
		}
	}
	events, err := eventsBySequence(ctx, tx, issueUUID, 0)
	if err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return events, nil
}

func (store *Store) PendingEvents(ctx context.Context, issueUUID domain.UUID) ([]Event, error) {
	return eventsBySequence(ctx, store.db, issueUUID, 0)
}

// Ack applies a cumulative server acknowledgement. Acknowledging an event that
// was never assigned is a protocol/storage boundary violation, never a hint to
// skip data.
func (store *Store) Ack(ctx context.Context, issueUUID domain.UUID, eventSeq uint64) error {
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	var lastAck, next, outputCharged, totalCharged uint64
	err = tx.QueryRowContext(ctx, `SELECT last_server_ack, next_event_seq, output_charged_bytes, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&lastAck, &next, &outputCharged, &totalCharged)
	if err == sql.ErrNoRows {
		return ErrUnknownCommand
	}
	if err != nil {
		return err
	}
	if eventSeq <= lastAck {
		return tx.Commit()
	}
	if eventSeq >= next {
		return ErrInvalidEventAck
	}
	var released, releasedOutput uint64
	if err := tx.QueryRowContext(ctx, `SELECT COALESCE(sum(charged_bytes), 0), COALESCE(sum(CASE WHEN output = 1 THEN charged_bytes ELSE 0 END), 0) FROM events WHERE issue_uuid = ? AND event_seq <= ?`, issueUUID[:], eventSeq).Scan(&released, &releasedOutput); err != nil {
		return err
	}
	if released > totalCharged || releasedOutput > outputCharged {
		return fmt.Errorf("client spool charge counter mismatch")
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return err
	}
	if released > clientTotal {
		return fmt.Errorf("client spool aggregate charge counter mismatch")
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE issue_uuid = ? AND event_seq <= ?`, issueUUID[:], eventSeq); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `UPDATE commands SET last_server_ack = ?, output_charged_bytes = ?, total_charged_bytes = ? WHERE issue_uuid = ?`, eventSeq, outputCharged-releasedOutput, totalCharged-released, issueUUID[:]); err != nil {
		return err
	}
	if err := updateClientTotalCharge(ctx, tx, clientTotal-released); err != nil {
		return err
	}
	return tx.Commit()
}

func (store *Store) MarkTerminal(ctx context.Context, issueUUID domain.UUID, phase uint32) error {
	if phase < 5 || phase > 11 {
		return errors.New("invalid terminal command phase")
	}
	result, err := store.db.ExecContext(ctx, `UPDATE commands SET phase = ?, terminal = 1 WHERE issue_uuid = ?`, phase, issueUUID[:])
	if err != nil {
		return err
	}
	count, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if count == 0 {
		return ErrUnknownCommand
	}
	return nil
}

// LaunchEvidence is the durable pre/post-authorization record used to fence
// uncertain OS launches across daemon restarts. PID is evidence only and is
// never sufficient for recovery-time signalling without a native creation
// identity check.
type LaunchEvidence struct {
	Phase   domain.LaunchPhase
	Context string
	PID     uint32
}

func (store *Store) SetLaunchPhase(ctx context.Context, issueUUID domain.UUID, phase domain.LaunchPhase, contextName string, pid uint32) error {
	if !validUUID(issueUUID) || phase > domain.LaunchPhaseAuthorized || len(contextName) > 128 || !utf8.ValidString(contextName) {
		return errors.New("invalid launch barrier")
	}
	if phase == domain.LaunchPhaseNone {
		contextName, pid = "", 0
	}
	result, err := store.db.ExecContext(ctx, `UPDATE commands SET launch_phase = ?, launch_context = ?, launch_pid = ? WHERE issue_uuid = ? AND terminal = 0`, uint32(phase), contextName, pid, issueUUID[:])
	if err != nil {
		return err
	}
	count, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if count == 0 {
		return ErrUnknownCommand
	}
	return nil
}

// RecoverLaunchUncertainty converts every command whose authorization barrier
// crossed before a restart into one durable interrupted terminal event. The
// Windows Job kill-on-close guarantee makes this safe: a surviving process is
// never redispatched, and recovery never signals a PID by itself.
func (store *Store) RecoverLaunchUncertainty(ctx context.Context, now time.Time) ([]domain.UUID, error) {
	if now.IsZero() {
		return nil, errors.New("launch recovery time is required")
	}
	rows, err := store.db.QueryContext(ctx, `SELECT issue_uuid, command_revision FROM commands WHERE launch_phase = 2 AND terminal = 0 ORDER BY accepted_at, issue_uuid`)
	if err != nil {
		return nil, err
	}
	type pending struct {
		issue    domain.UUID
		revision uint64
	}
	var pendingRows []pending
	for rows.Next() {
		var encoded []byte
		var revision uint64
		if err := rows.Scan(&encoded, &revision); err != nil {
			_ = rows.Close()
			return nil, err
		}
		var issue domain.UUID
		if len(encoded) != len(issue) {
			_ = rows.Close()
			return nil, ErrScriptState
		}
		copy(issue[:], encoded)
		if !validUUID(issue) || revision == 0 {
			_ = rows.Close()
			return nil, ErrScriptState
		}
		pendingRows = append(pendingRows, pending{issue: issue, revision: revision})
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	interrupted := make([]domain.UUID, 0, len(pendingRows))
	for _, item := range pendingRows {
		if _, err := store.AppendLifecycle(ctx, item.issue, uint32(rvboxv1.CommandLifecycle_COMMAND_INTERRUPTED), item.revision, "uncertain launch recovered after daemon restart", now); err != nil {
			return interrupted, err
		}
		interrupted = append(interrupted, item.issue)
	}
	return interrupted, nil
}

// AppendLifecycle atomically advances the durable command phase and appends
// its lifecycle event. Keeping these writes in one transaction prevents a
// crash between a terminal marker and its public event from creating a state
// that can be replayed as a second execution.
func (store *Store) AppendLifecycle(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time) (Event, error) {
	return store.AppendLifecycleWithIdentity(ctx, issueUUID, phase, revision, detail, observedAt, nil)
}

// AppendLifecycleWithIdentity is the lifecycle boundary used by the client
// runtime when the Windows supervisor has captured an immutable selection
// record.  Keeping the legacy wrapper above preserves the small store API for
// recovery and tests that have no platform identity to attach.
func (store *Store) AppendLifecycleWithIdentity(ctx context.Context, issueUUID domain.UUID, phase uint32, revision uint64, detail string, observedAt time.Time, identity *rvboxv1.WindowsExecutionIdentity) (Event, error) {
	if !validUUID(issueUUID) || phase == 0 || phase > 11 || observedAt.IsZero() || revision == 0 {
		return Event{}, errors.New("invalid lifecycle event")
	}
	if len(detail) > 4096 || !utf8.ValidString(detail) {
		return Event{}, errors.New("lifecycle detail is invalid or too large")
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return Event{}, err
	}
	defer tx.Rollback()
	var current uint32
	var storedRevision uint64
	var nextOrdinal, outputCharged, totalCharged, closeout uint64
	if err := tx.QueryRowContext(ctx, `SELECT phase, command_revision, next_local_ordinal, output_charged_bytes, total_charged_bytes, closeout_remaining_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&current, &storedRevision, &nextOrdinal, &outputCharged, &totalCharged, &closeout); err == sql.ErrNoRows {
		return Event{}, ErrUnknownCommand
	} else if err != nil {
		return Event{}, err
	}
	if storedRevision != revision {
		return Event{}, ErrCommandConflict
	}
	if current == phase {
		return Event{}, nil
	}
	if !domain.CanTransition(rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase)) {
		return Event{}, fmt.Errorf("invalid lifecycle transition %s -> %s", rvboxv1.CommandLifecycle(current), rvboxv1.CommandLifecycle(phase))
	}
	lifecycle := &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle(phase), CommandRevision: revision, Detail: detail, WindowsExecutionIdentity: identity}
	payload, err := proto.MarshalOptions{Deterministic: true}.Marshal(&rvboxv1.CommandEvent{IssueUuid: issueUUID.String(), ObservedAt: timestamppb.New(observedAt), Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: lifecycle}})
	if err != nil {
		return Event{}, err
	}
	charge, err := EstimateCharge(ChargeInput{EncodedBytes: uint64(len(payload)), SQLiteRows: 1, IndexEntries: 2})
	if err != nil {
		return Event{}, err
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return Event{}, err
	}
	decision, err := CheckReservation(store.quotaLimits, ReservationState{CommandOutputCharged: outputCharged, CommandTotalCharged: totalCharged, ClientTotalCharged: clientTotal, CloseoutRemaining: closeout}, ReservationRequest{ChargedBytes: charge, UseCloseout: isTerminalPhase(phase)})
	if err != nil {
		return Event{}, err
	}
	digest := immutableDigest(payload)
	if _, err := tx.ExecContext(ctx, `INSERT INTO events(issue_uuid, local_ordinal, event_kind, compression, raw_bytes, charged_bytes, output, payload, payload_sha256, created_at) VALUES (?, ?, 4, 1, ?, ?, 0, ?, ?, ?)`, issueUUID[:], nextOrdinal, len(payload), charge, payload, digest[:], observedAt.UnixNano()); err != nil {
		return Event{}, err
	}
	terminal := isTerminalPhase(phase)
	if _, err := tx.ExecContext(ctx, `UPDATE commands SET phase = ?, terminal = ?, launch_phase = CASE WHEN ? = 1 THEN 0 ELSE launch_phase END, launch_context = CASE WHEN ? = 1 THEN '' ELSE launch_context END, launch_pid = CASE WHEN ? = 1 THEN 0 ELSE launch_pid END, next_local_ordinal = ?, total_charged_bytes = ?, closeout_remaining_bytes = ? WHERE issue_uuid = ?`, phase, boolInt(terminal), boolInt(terminal), boolInt(terminal), boolInt(terminal), nextOrdinal+1, decision.CommandTotalCharged, decision.CloseoutRemaining, issueUUID[:]); err != nil {
		return Event{}, err
	}
	if err := updateClientTotalCharge(ctx, tx, decision.ClientTotalCharged); err != nil {
		return Event{}, err
	}
	if err := tx.Commit(); err != nil {
		return Event{}, err
	}
	return Event{IssueUUID: issueUUID, LocalOrdinal: nextOrdinal, Kind: 4, Compression: 1, RawBytes: uint64(len(payload)), Payload: append([]byte(nil), payload...), CreatedAt: observedAt}, nil
}

// CleanupTerminal moves a fully acknowledged terminal command into the compact
// tombstone ledger and removes all command-owned spool data in one transaction.
func (store *Store) CleanupTerminal(ctx context.Context, issueUUID domain.UUID, acknowledgedAt time.Time) error {
	if acknowledgedAt.IsZero() {
		return errors.New("terminal acknowledgement time is required")
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	var hash []byte
	var terminal int
	var revision uint64
	var phase uint32
	var totalCharged uint64
	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &revision, &phase, &terminal, &totalCharged)
	if err == sql.ErrNoRows {
		return ErrUnknownCommand
	}
	if err != nil {
		return err
	}
	if terminal == 0 {
		return errors.New("command is not terminal")
	}
	var pending uint64
	if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM events WHERE issue_uuid = ?`, issueUUID[:]).Scan(&pending); err != nil {
		return err
	}
	if pending != 0 {
		return ErrEventsPending
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return err
	}
	if totalCharged > clientTotal {
		return fmt.Errorf("client spool aggregate charge counter mismatch")
	}
	if _, err := tx.ExecContext(ctx, `INSERT INTO command_tombstones(issue_uuid, immutable_sha256, command_revision, terminal_lifecycle, acknowledged_at) VALUES (?, ?, ?, ?, ?)`, issueUUID[:], hash, revision, phase, acknowledgedAt.UnixNano()); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM commands WHERE issue_uuid = ?`, issueUUID[:]); err != nil {
		return err
	}
	if err := updateClientTotalCharge(ctx, tx, clientTotal-totalCharged); err != nil {
		return err
	}
	trim, err := tombstonesToTrim(ctx, tx, store.tombstoneLimit)
	if err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM command_tombstones WHERE issue_uuid IN (SELECT issue_uuid FROM command_tombstones ORDER BY acknowledged_at, issue_uuid LIMIT ?)`, trim); err != nil {
		return err
	}
	return tx.Commit()
}

// DiscardTerminal removes a server-confirmed terminal command even when its
// local event rows have not been acknowledged. The compact tombstone is kept so
// a late duplicate dispatch cannot execute it again. It is intentionally only
// for a ReconcileResult that authoritatively tells the client to discard local
// terminal state; ordinary terminal cleanup must use CleanupTerminal.
func (store *Store) DiscardTerminal(ctx context.Context, issueUUID domain.UUID, acknowledgedAt time.Time) error {
	if !validUUID(issueUUID) || acknowledgedAt.IsZero() {
		return errors.New("invalid terminal discard")
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var hash []byte
	var revision uint64
	var phase uint32
	var terminal int
	var totalCharged uint64
	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &revision, &phase, &terminal, &totalCharged)
	if errors.Is(err, sql.ErrNoRows) {
		var tombstoneHash []byte
		err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, issueUUID[:]).Scan(&tombstoneHash)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrUnknownCommand
		}
		if err != nil {
			return err
		}
		if len(tombstoneHash) != 32 {
			return fmt.Errorf("invalid terminal tombstone")
		}
		return tx.Commit()
	}
	if err != nil {
		return err
	}
	if terminal == 0 {
		return errors.New("command is not terminal")
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return err
	}
	if totalCharged > clientTotal {
		return fmt.Errorf("client spool aggregate charge counter mismatch")
	}
	var existingHash []byte
	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, issueUUID[:]).Scan(&existingHash)
	if err == nil && (len(existingHash) != len(hash) || string(existingHash) != string(hash)) {
		return ErrCommandConflict
	}
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if errors.Is(err, sql.ErrNoRows) {
		if _, err := tx.ExecContext(ctx, `INSERT INTO command_tombstones(issue_uuid, immutable_sha256, command_revision, terminal_lifecycle, acknowledged_at) VALUES (?, ?, ?, ?, ?)`, issueUUID[:], hash, revision, phase, acknowledgedAt.UnixNano()); err != nil {
			return err
		}
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM commands WHERE issue_uuid = ?`, issueUUID[:]); err != nil {
		return err
	}
	if err := updateClientTotalCharge(ctx, tx, clientTotal-totalCharged); err != nil {
		return err
	}
	trim, err := tombstonesToTrim(ctx, tx, store.tombstoneLimit)
	if err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM command_tombstones WHERE issue_uuid IN (SELECT issue_uuid FROM command_tombstones ORDER BY acknowledged_at, issue_uuid LIMIT ?)`, trim); err != nil {
		return err
	}
	return tx.Commit()
}

type queryer interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}

func eventsBySequence(ctx context.Context, query queryer, issueUUID domain.UUID, after uint64) ([]Event, error) {
	rows, err := query.QueryContext(ctx, `SELECT local_ordinal, event_seq, event_kind, compression, raw_bytes, payload, created_at FROM events WHERE issue_uuid = ? AND event_seq IS NOT NULL AND event_seq > ? ORDER BY event_seq`, issueUUID[:], after)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var events []Event
	for rows.Next() {
		var event Event
		var created int64
		event.IssueUUID = issueUUID
		if err := rows.Scan(&event.LocalOrdinal, &event.EventSeq, &event.Kind, &event.Compression, &event.RawBytes, &event.Payload, &created); err != nil {
			return nil, err
		}
		event.CreatedAt = time.Unix(0, created).UTC()
		events = append(events, event)
	}
	return events, rows.Err()
}

func commandByUUID(ctx context.Context, query interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}, issueUUID domain.UUID, maxSpecBytes uint64) (Command, bool, error) {
	var command Command
	var hash []byte
	var terminal int
	var stored, digest []byte
	var rawBytes, storedBytes, compression sql.NullInt64
	err := query.QueryRowContext(ctx, `SELECT c.immutable_sha256, c.command_revision, c.phase, c.terminal,
	 s.raw_bytes, s.stored_bytes, s.compression, s.payload, s.payload_sha256
	 FROM commands c LEFT JOIN command_specs s ON s.issue_uuid = c.issue_uuid WHERE c.issue_uuid = ?`, issueUUID[:]).Scan(&hash, &command.Revision, &command.Phase, &terminal, &rawBytes, &storedBytes, &compression, &stored, &digest)
	if err == sql.ErrNoRows {
		return Command{}, false, nil
	}
	if err != nil {
		return Command{}, false, err
	}
	if len(hash) != len(command.ImmutableSHA256) {
		return Command{}, false, fmt.Errorf("invalid stored immutable command hash")
	}
	copy(command.ImmutableSHA256[:], hash)
	command.IssueUUID = issueUUID
	command.Terminal = terminal != 0
	if rawBytes.Valid {
		computed := immutableDigest(stored)
		if rawBytes.Int64 <= 0 || storedBytes.Int64 <= 0 || compression.Int64 != 2 || len(digest) != sha256.Size || uint64(rawBytes.Int64) > maxSpecBytes || uint64(storedBytes.Int64) != uint64(len(stored)) || !bytes.Equal(computed[:], digest) {
			return Command{}, false, fmt.Errorf("invalid stored execution specification")
		}
		decoded, err := decompressExecutionSpec(stored, uint64(rawBytes.Int64), maxSpecBytes)
		if err != nil {
			return Command{}, false, err
		}
		command.ExecutionSpec = decoded
	}
	return command, true, nil
}

func compressExecutionSpec(raw []byte) ([]byte, error) {
	encoder, err := zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1))
	if err != nil {
		return nil, err
	}
	defer encoder.Close()
	return encoder.EncodeAll(raw, nil), nil
}

func decompressExecutionSpec(stored []byte, rawBytes, maximum uint64) ([]byte, error) {
	if rawBytes == 0 || rawBytes > maximum {
		return nil, errors.New("invalid stored execution specification size")
	}
	decoder, err := zstd.NewReader(bytes.NewReader(stored), zstd.WithDecoderConcurrency(1), zstd.WithDecoderMaxMemory(maximum+1))
	if err != nil {
		return nil, errors.New("invalid stored execution specification")
	}
	defer decoder.Close()
	decoded, err := decoder.DecodeAll(stored, nil)
	if err != nil || uint64(len(decoded)) != rawBytes || uint64(len(decoded)) > maximum {
		return nil, errors.New("invalid stored execution specification")
	}
	return decoded, nil
}

func tombstonesToTrim(ctx context.Context, tx *sql.Tx, limit uint64) (uint64, error) {
	var count uint64
	if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM command_tombstones`).Scan(&count); err != nil {
		return 0, err
	}
	if count <= limit {
		return 0, nil
	}
	return count - limit, nil
}

func boolInt(value bool) int {
	if value {
		return 1
	}
	return 0
}

func clientTotalCharge(ctx context.Context, tx *sql.Tx) (uint64, error) {
	var total uint64
	if err := tx.QueryRowContext(ctx, `SELECT client_total_charged_bytes FROM spool_counters WHERE singleton = 1`).Scan(&total); err != nil {
		return 0, err
	}
	return total, nil
}

func updateClientTotalCharge(ctx context.Context, tx *sql.Tx, total uint64) error {
	_, err := tx.ExecContext(ctx, `UPDATE spool_counters SET client_total_charged_bytes = ? WHERE singleton = 1`, total)
	return err
}

func validUUID(value domain.UUID) bool {
	_, err := domain.ParseUUIDv7(value.String())
	return err == nil
}

func isTerminalPhase(phase uint32) bool { return phase >= 5 && phase <= 11 }