rvbox/internal/client/spool/events.go

package spool

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

	"github.com/rvbox/rvbox/internal/domain"
)

type Command struct {
	IssueUUID       domain.UUID
	ImmutableSHA256 [32]byte
	Revision        uint64
	Phase           uint32
	Terminal        bool
}

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")
	}
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return Acceptance{}, err
	}
	defer tx.Rollback()
	existing, found, err := commandByUUID(ctx, tx, command.IssueUUID)
	if err != nil {
		return Acceptance{}, err
	}
	if found {
		if existing.ImmutableSHA256 != command.ImmutableSHA256 {
			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
	}
	charge, err := EstimateCharge(ChargeInput{SQLiteRows: 1, IndexEntries: 2})
	if err != nil {
		return Acceptance{}, err
	}
	clientTotal, err := clientTotalCharge(ctx, tx)
	if err != nil {
		return Acceptance{}, err
	}
	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), charge, 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 err := tx.Commit(); err != nil {
		return Acceptance{}, err
	}
	return Acceptance{Command: 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 uint64
	err = tx.QueryRowContext(ctx, `SELECT next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next)
	if err == sql.ErrNoRows {
		return nil, ErrUnknownCommand
	}
	if err != nil {
		return nil, err
	}
	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
}

// 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()
}

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) (Command, bool, error) {
	var command Command
	var hash []byte
	var terminal int
	err := query.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &command.Revision, &command.Phase, &terminal)
	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
	return command, true, 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 }