import { createRequire } from 'module';
import { randomUUID } from 'crypto';
import { mkdirSync } from 'fs';
import { dirname } from 'path';
import { loadNodeSqliteModule } from '../load-node-sqlite.js';
import type { MailboxRepository } from '../mailbox/mailbox-repository.js';
import type {
  AckMailboxClaimRequest,
  AckMailboxClaimResult,
  ClaimMailboxItemsRequest,
  EnqueueMailboxItem,
  EnqueueMailboxResult,
  MailboxClaim,
  MailboxDeliveryMode,
  MailboxItem,
  MailboxPeekFilter,
  NackMailboxClaimRequest,
  NackMailboxClaimResult,
  SuppressMailboxItemsFilter,
} from '../mailbox/types.js';
import { DeliveryLeaseLostError, type EventRepository } from './event-repository.js';
import {
  cloneRecord,
  initialDeliveryFor,
  type AcquireLeaseRequest,
  type BacklogSummary,
  type CanonicalEvent,
  type CommittedEventListFilter,
  type CommitEventRequest,
  type CommitEventResult,
  type ConsumerLease,
  type CurrentEventScope,
  type DetailStateTransition,
  type EventAccountScope,
  type EventDeliveryRecord,
  type EventListFilter,
  type EventSegmentRecord,
  type EventWithDelivery,
  type LatestEventScopeOptions,
  type MonitorDeliveryAttempt,
  type MonitorStateTransition,
  type PlaySessionRecord,
  type SourceCheckpointRecord,
} from './types.js';

interface StatementLike {
  run(...params: unknown[]): { changes?: number | bigint; lastInsertRowid?: number | bigint };
  get(...params: unknown[]): Record<string, unknown> | undefined;
  all(...params: unknown[]): Array<Record<string, unknown>>;
}

interface DatabaseLike {
  exec(sql: string): void;
  prepare(sql: string): StatementLike;
  close(): void;
}

interface SqliteModuleLike {
  DatabaseSync?: new (path: string) => DatabaseLike;
}

export interface SqliteUnavailable {
  code: 'SQLITE_UNAVAILABLE' | 'SQLITE_OPEN_FAILED';
  message: string;
  cause?: string;
}

export type SqliteRepositoryOpenResult =
  | { ok: true; repository: SqliteEventRepository }
  | { ok: false; unavailable: SqliteUnavailable };

const require = createRequire(import.meta.url);

function loadSqlite(): { ok: true; module: SqliteModuleLike } | { ok: false; unavailable: SqliteUnavailable } {
  try {
    const module = loadNodeSqliteModule<SqliteModuleLike>(require);
    if (typeof module.DatabaseSync !== 'function') {
      return {
        ok: false,
        unavailable: {
          code: 'SQLITE_UNAVAILABLE',
          message: 'This Node.js runtime does not expose node:sqlite DatabaseSync.',
        },
      };
    }
    return { ok: true, module };
  } catch (error) {
    return {
      ok: false,
      unavailable: {
        code: 'SQLITE_UNAVAILABLE',
        message: 'SQLite event storage is unavailable in this Node.js runtime. Use MemoryEventRepository or a Node build with node:sqlite.',
        cause: error instanceof Error ? error.message : String(error),
      },
    };
  }
}

export function detectSqliteAvailability(): SqliteUnavailable | null {
  const result = loadSqlite();
  return result.ok ? null : result.unavailable;
}

function booleanInt(value: boolean): number {
  return value ? 1 : 0;
}

function numberValue(value: unknown): number {
  return typeof value === 'bigint' ? Number(value) : Number(value ?? 0);
}

function optionalString(value: unknown): string | undefined {
  return typeof value === 'string' ? value : undefined;
}

function sessionFromRow(row: Record<string, unknown>): PlaySessionRecord {
  return {
    sessionId: String(row.session_id),
    adapterId: String(row.adapter_id),
    accountId: String(row.account_id),
    resumeKey: optionalString(row.resume_key),
    mode: optionalString(row.mode),
    status: row.status === 'sealed' ? 'sealed' : 'open',
    isCurrent: numberValue(row.is_current) === 1,
    openedAt: String(row.opened_at),
    sealedAt: optionalString(row.sealed_at),
  };
}

function segmentFromRow(row: Record<string, unknown>): EventSegmentRecord {
  return {
    segmentId: String(row.segment_id),
    sessionId: String(row.session_id),
    sequence: numberValue(row.segment_seq),
    kind: String(row.kind),
    scopeRef: optionalString(row.scope_ref),
    status: row.status === 'closed' ? 'closed' : 'open',
    isCurrent: numberValue(row.is_current) === 1,
    openedAt: String(row.opened_at),
    closedAt: optionalString(row.closed_at),
  };
}

function eventFromRow(row: Record<string, unknown>): CanonicalEvent {
  const payload = JSON.parse(String(row.payload_json)) as Record<string, unknown>;
  return {
    commitSeq: numberValue(row.commit_seq),
    ...(row.epoch_seq === null || row.epoch_seq === undefined
      ? {}
      : { epochSeq: numberValue(row.epoch_seq) }),
    eventId: String(row.event_id),
    sessionId: String(row.session_id),
    segmentId: String(row.segment_id),
    type: String(row.type),
    producer: row.producer === 'local_synth' ? 'local_synth' : 'backend_push',
    ...(optionalString(row.producer_plugin_id) === undefined
      ? {}
      : { producerPluginId: optionalString(row.producer_plugin_id) }),
    ...(row.producer_output_index === null || row.producer_output_index === undefined
      ? {}
      : { producerOutputIndex: numberValue(row.producer_output_index) }),
    deliveryClass: row.delivery_class === 'monitor_and_events'
      ? 'monitor_and_events'
      : row.delivery_class === 'events_only'
        ? 'events_only'
        : 'internal_only',
    occurredAt: String(row.occurred_at),
    receivedAt: String(row.received_at),
    payload,
    sourceId: String(row.source_id),
    sourceEventId: optionalString(row.source_event_id),
    sourceSequence: optionalString(row.source_sequence),
    connectionId: optionalString(row.connection_id),
    replayed: numberValue(row.replayed) === 1,
  };
}

function deliveryFromRow(row: Record<string, unknown>): EventDeliveryRecord {
  return {
    eventId: String(row.event_id),
    consumerId: String(row.consumer_id),
    monitorState: row.monitor_state as EventDeliveryRecord['monitorState'],
    monitorAttemptCount: numberValue(row.monitor_attempt_count),
    monitorLastAttemptAt: optionalString(row.monitor_last_attempt_at),
    monitorDeliveredAt: optionalString(row.monitor_delivered_at),
    detailState: row.detail_state as EventDeliveryRecord['detailState'],
    detailDeliveredAt: optionalString(row.detail_delivered_at),
    detailDeliveryVia: row.detail_delivery_via as EventDeliveryRecord['detailDeliveryVia'],
  };
}

function checkpointFromRow(row: Record<string, unknown>): SourceCheckpointRecord {
  return {
    sourceId: String(row.source_id),
    checkpoint: String(row.checkpoint),
    sourceSequence: optionalString(row.source_sequence),
    eventId: String(row.event_id),
    committedAt: String(row.committed_at),
  };
}

function mailboxItemFromRow<T = unknown>(row: Record<string, unknown>): MailboxItem<T> {
  return {
    itemId: String(row.item_id),
    schemaVersion: numberValue(row.schema_version),
    formatterVersion: String(row.formatter_version),
    consumerId: String(row.consumer_id),
    sessionId: String(row.session_id),
    segmentId: String(row.segment_id),
    eventId: String(row.event_id),
    eventCommitSeq: numberValue(row.event_commit_seq),
    // New runtime code only creates short notifications. The SQLite schema
    // still accepts historical full rows so existing ledgers open in place.
    mode: 'short',
    coalesceKey: String(row.coalesce_key),
    payload: JSON.parse(String(row.payload_json)) as T,
    state: row.state as MailboxItem<T>['state'],
    attemptCount: numberValue(row.attempt_count),
    availableAtMs: numberValue(row.available_at),
    createdAt: String(row.created_at),
    updatedAt: String(row.updated_at),
    lastAttemptAt: optionalString(row.last_attempt_at),
    claimOwner: optionalString(row.claim_owner),
    claimToken: optionalString(row.claim_token),
    claimExpiresAtMs: row.claim_expires_at === null || row.claim_expires_at === undefined
      ? undefined
      : numberValue(row.claim_expires_at),
    acceptedAt: optionalString(row.accepted_at),
    acceptanceLevel: row.acceptance_level === 'host_accepted'
      ? 'host_accepted'
      : row.acceptance_level === 'sink_emitted'
        ? 'sink_emitted'
        : undefined,
    receiptId: optionalString(row.receipt_id),
    lastErrorCode: optionalString(row.last_error_code),
  };
}

function placeholders(count: number): string {
  return new Array(count).fill('?').join(', ');
}

export class SqliteEventRepository implements EventRepository, MailboxRepository {
  static open(path: string): SqliteRepositoryOpenResult {
    const loaded = loadSqlite();
    if (!loaded.ok) return loaded;
    let db: DatabaseLike | undefined;
    try {
      if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true });
      const DatabaseSync = loaded.module.DatabaseSync;
      if (!DatabaseSync) {
        return {
          ok: false,
          unavailable: {
            code: 'SQLITE_UNAVAILABLE',
            message: 'This Node.js runtime does not expose node:sqlite DatabaseSync.',
          },
        };
      }
      db = new DatabaseSync(path);
      const repository = new SqliteEventRepository(db);
      repository.configure();
      repository.migrate();
      return { ok: true, repository };
    } catch (error) {
      try { db?.close(); } catch {}
      return {
        ok: false,
        unavailable: {
          code: 'SQLITE_OPEN_FAILED',
          message: `Unable to open the SQLite event repository at ${path}.`,
          cause: error instanceof Error ? error.message : String(error),
        },
      };
    }
  }

  private constructor(private readonly db: DatabaseLike) {}

  close(): void {
    this.db.close();
  }

  schemaVersion(): number {
    const row = this.db.prepare('SELECT MAX(version) AS version FROM schema_migrations').get();
    return numberValue(row?.version);
  }

  journalMode(): string {
    const row = this.db.prepare('PRAGMA journal_mode').get();
    return String(row?.journal_mode ?? 'unknown').toLowerCase();
  }

  saveSession(session: PlaySessionRecord): void {
    this.transaction(() => {
      const existing = this.db.prepare(`
        SELECT is_current FROM play_sessions WHERE session_id = ?
      `).get(session.sessionId);
      if (numberValue(existing?.is_current) === 1 && !session.isCurrent) {
        this.retireMailboxScope(
          'session_id',
          session.sessionId,
          session.sealedAt ?? session.openedAt,
        );
      }
      if (session.isCurrent) {
        this.db.prepare(`
          UPDATE mailbox_items
             SET state = 'suppressed', updated_at = ?
           WHERE state = 'pending'
             AND session_id IN (
               SELECT session_id
                 FROM play_sessions
                WHERE adapter_id = ?
                  AND account_id = ?
                  AND session_id <> ?
                  AND is_current = 1
             )
        `).run(session.openedAt, session.adapterId, session.accountId, session.sessionId);
        this.db.prepare(`
          UPDATE event_deliveries
             SET detail_state = CASE
                   WHEN detail_state = 'pending' THEN 'deferred'
                   ELSE detail_state
                 END,
                 monitor_state = CASE
                   WHEN monitor_state = 'pending'
                    AND NOT EXISTS (
                      SELECT 1
                        FROM mailbox_items m
                       WHERE m.event_id = event_deliveries.event_id
                         AND m.consumer_id = event_deliveries.consumer_id
                         AND m.state = 'claimed'
                    ) THEN 'suppressed'
                   ELSE monitor_state
                 END
           WHERE event_id IN (
             SELECT e.event_id
               FROM events e
               JOIN play_sessions p ON p.session_id = e.session_id
              WHERE p.adapter_id = ?
                AND p.account_id = ?
                AND p.session_id <> ?
                AND p.is_current = 1
           )
        `).run(session.adapterId, session.accountId, session.sessionId);
        this.db.prepare(`
          UPDATE play_sessions
             SET is_current = 0
           WHERE adapter_id = ? AND account_id = ? AND session_id <> ? AND is_current = 1
        `).run(session.adapterId, session.accountId, session.sessionId);
      }
      this.db.prepare(`
        INSERT INTO play_sessions(
          session_id, adapter_id, account_id, resume_key, mode, status,
          is_current, opened_at, sealed_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(session_id) DO UPDATE SET
          adapter_id = excluded.adapter_id,
          account_id = excluded.account_id,
          resume_key = excluded.resume_key,
          mode = excluded.mode,
          status = excluded.status,
          is_current = excluded.is_current,
          opened_at = excluded.opened_at,
          sealed_at = excluded.sealed_at
      `).run(
        session.sessionId,
        session.adapterId,
        session.accountId,
        session.resumeKey ?? null,
        session.mode ?? null,
        session.status,
        booleanInt(session.isCurrent),
        session.openedAt,
        session.sealedAt ?? null,
      );
    });
  }

  saveSegment(segment: EventSegmentRecord): void {
    this.transaction(() => {
      const existing = this.db.prepare(`
        SELECT is_current FROM event_segments WHERE segment_id = ?
      `).get(segment.segmentId);
      if (numberValue(existing?.is_current) === 1 && !segment.isCurrent) {
        this.db.prepare(`
          UPDATE mailbox_items
             SET available_at = 0, updated_at = ?
           WHERE state = 'pending' AND segment_id = ?
        `).run(segment.closedAt ?? segment.openedAt, segment.segmentId);
        this.db.prepare(`
          UPDATE event_deliveries
             SET detail_state = 'deferred'
           WHERE detail_state = 'pending'
             AND event_id IN (SELECT event_id FROM events WHERE segment_id = ?)
        `).run(segment.segmentId);
      }
      if (segment.isCurrent) {
        this.db.prepare(`
          UPDATE mailbox_items
             SET available_at = 0, updated_at = ?
           WHERE state = 'pending'
             AND segment_id IN (
               SELECT segment_id
                 FROM event_segments
                WHERE session_id = ?
                  AND is_current = 1
                  AND segment_id <> ?
             )
        `).run(segment.openedAt, segment.sessionId, segment.segmentId);
        this.db.prepare(`
          UPDATE event_deliveries
             SET detail_state = 'deferred'
           WHERE detail_state = 'pending'
             AND event_id IN (
               SELECT e.event_id
                 FROM events e
                 JOIN event_segments s ON s.segment_id = e.segment_id
                WHERE s.session_id = ?
                  AND s.is_current = 1
                  AND s.segment_id <> ?
             )
        `).run(segment.sessionId, segment.segmentId);
        this.db.prepare(`
          UPDATE event_segments
             SET is_current = 0
           WHERE session_id = ? AND segment_id <> ? AND is_current = 1
        `).run(segment.sessionId, segment.segmentId);
      }
      this.db.prepare(`
        INSERT INTO event_segments(
          segment_id, session_id, segment_seq, kind, scope_ref, status,
          is_current, opened_at, closed_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(segment_id) DO UPDATE SET
          session_id = excluded.session_id,
          segment_seq = excluded.segment_seq,
          kind = excluded.kind,
          scope_ref = excluded.scope_ref,
          status = excluded.status,
          is_current = excluded.is_current,
          opened_at = excluded.opened_at,
          closed_at = excluded.closed_at
      `).run(
        segment.segmentId,
        segment.sessionId,
        segment.sequence,
        segment.kind,
        segment.scopeRef ?? null,
        segment.status,
        booleanInt(segment.isCurrent),
        segment.openedAt,
        segment.closedAt ?? null,
      );
    });
  }

  currentScope(account?: EventAccountScope): CurrentEventScope | null {
    const where = account ? 'WHERE is_current = 1 AND adapter_id = ? AND account_id = ?' : 'WHERE is_current = 1';
    const params = account ? [account.adapterId, account.accountId] : [];
    const sessions = this.db.prepare(`
      SELECT * FROM play_sessions ${where} ORDER BY opened_at DESC LIMIT 2
    `).all(...params);
    if (sessions.length > 1) {
      throw new Error('Multiple current event sessions exist; provide adapterId and accountId.');
    }
    if (sessions.length === 0) return null;
    const session = sessionFromRow(sessions[0]);
    const segmentRow = this.db.prepare(`
      SELECT * FROM event_segments
       WHERE session_id = ? AND is_current = 1
       ORDER BY segment_seq DESC LIMIT 1
    `).get(session.sessionId);
    if (!segmentRow) return null;
    return { session, segment: segmentFromRow(segmentRow) };
  }

  latestScope(account: EventAccountScope, options: LatestEventScopeOptions = {}): CurrentEventScope | null {
    const sessionRow = this.db.prepare(`
      SELECT * FROM play_sessions
       WHERE adapter_id = ? AND account_id = ?
       ORDER BY is_current DESC, opened_at DESC
       LIMIT 1
    `).get(account.adapterId, account.accountId);
    if (!sessionRow) return null;
    const session = sessionFromRow(sessionRow);
    const clauses = ['session_id = ?'];
    const params: unknown[] = [session.sessionId];
    if (options.segmentKinds?.length) {
      clauses.push(`kind IN (${options.segmentKinds.map(() => '?').join(', ')})`);
      params.push(...options.segmentKinds);
    }
    const segmentRow = this.db.prepare(`
      SELECT * FROM event_segments
       WHERE ${clauses.join(' AND ')}
       ORDER BY is_current DESC, segment_seq DESC
       LIMIT 1
    `).get(...params);
    return segmentRow ? { session, segment: segmentFromRow(segmentRow) } : null;
  }

  commitEvent(request: CommitEventRequest): CommitEventResult {
    return this.transaction(() => {
      const duplicateById = request.event.sourceEventId
        ? this.findByRemoteId(request.event.sourceId, request.event.sourceEventId)
        : null;
      const duplicateBySequence = request.event.sourceSequence
        ? this.findByRemoteSequence(
          request.event.sourceId,
          request.event.sessionId,
          request.event.segmentId,
          request.event.sourceSequence,
        )
        : null;
      if (duplicateById && duplicateBySequence && duplicateById.eventId !== duplicateBySequence.eventId) {
        throw new Error('Conflicting source event ID and source sequence identities.');
      }
      const duplicate = duplicateById ?? duplicateBySequence;
      if (duplicate) {
        for (const consumerId of new Set(request.consumerIds)) {
          this.insertInitialDelivery(duplicate, consumerId);
        }
        return { inserted: false, event: duplicate };
      }
      if (this.eventById(request.event.eventId)) {
        throw new Error(`Event ID already exists: ${request.event.eventId}`);
      }

      const result = this.db.prepare(`
        INSERT INTO events(
          event_id, epoch_seq, session_id, segment_id, type, producer,
          producer_plugin_id, producer_output_index, delivery_class,
          occurred_at, received_at, payload_json, source_id, source_event_id,
          source_sequence, connection_id, replayed
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
      `).run(
        request.event.eventId,
        request.event.epochSeq ?? null,
        request.event.sessionId,
        request.event.segmentId,
        request.event.type,
        request.event.producer,
        request.event.producerPluginId ?? null,
        request.event.producerOutputIndex ?? null,
        request.event.deliveryClass,
        request.event.occurredAt,
        request.event.receivedAt,
        JSON.stringify(request.event.payload),
        request.event.sourceId,
        request.event.sourceEventId ?? null,
        request.event.sourceSequence ?? null,
        request.event.connectionId ?? null,
        booleanInt(request.event.replayed),
      );
      const event: CanonicalEvent = {
        ...cloneRecord(request.event),
        commitSeq: numberValue(result.lastInsertRowid),
      };
      for (const consumerId of new Set(request.consumerIds)) {
        this.insertInitialDelivery(event, consumerId);
      }
      if (request.checkpoint) this.upsertCheckpoint(request.checkpoint);
      return { inserted: true, event: cloneRecord(event) };
    });
  }

  eventById(eventId: string): CanonicalEvent | null {
    const row = this.db.prepare('SELECT * FROM events WHERE event_id = ?').get(eventId);
    return row ? eventFromRow(row) : null;
  }

  latestEventByType(sessionId: string, type: string, segmentId?: string): CanonicalEvent | null {
    const row = segmentId
      ? this.db.prepare(`
          SELECT * FROM events
           WHERE session_id = ? AND segment_id = ? AND type = ?
           ORDER BY commit_seq DESC LIMIT 1
        `).get(sessionId, segmentId, type)
      : this.db.prepare(`
          SELECT * FROM events WHERE session_id = ? AND type = ? ORDER BY commit_seq DESC LIMIT 1
        `).get(sessionId, type);
    return row ? eventFromRow(row) : null;
  }

  highWaterCommitSeq(sessionId: string, segmentId?: string): number {
    const row = segmentId
      ? this.db.prepare('SELECT MAX(commit_seq) AS high_water FROM events WHERE session_id = ? AND segment_id = ?').get(sessionId, segmentId)
      : this.db.prepare('SELECT MAX(commit_seq) AS high_water FROM events WHERE session_id = ?').get(sessionId);
    return numberValue(row?.high_water);
  }

  listCommittedEvents(filter: CommittedEventListFilter): CanonicalEvent[] {
    const clauses = ['session_id = ?'];
    const params: unknown[] = [filter.sessionId];
    if (filter.segmentId) {
      clauses.push('segment_id = ?');
      params.push(filter.segmentId);
    }
    if (filter.types?.length) {
      clauses.push(`type IN (${filter.types.map(() => '?').join(', ')})`);
      params.push(...filter.types);
    }
    if (filter.epochSeq !== undefined) {
      clauses.push('epoch_seq = ?');
      params.push(filter.epochSeq);
    }
    if (filter.afterCommitSeq !== undefined) {
      clauses.push('commit_seq > ?');
      params.push(filter.afterCommitSeq);
    }
    if (filter.maxCommitSeq !== undefined) {
      clauses.push('commit_seq <= ?');
      params.push(filter.maxCommitSeq);
    }
    if (filter.maxEpochSeq !== undefined) {
      clauses.push('(epoch_seq IS NULL OR epoch_seq <= ?)');
      params.push(filter.maxEpochSeq);
    }
    const order = filter.order === 'desc' ? 'DESC' : 'ASC';
    const limit = filter.limit === undefined
      ? ''
      : ' LIMIT ?';
    if (filter.limit !== undefined) params.push(Math.max(0, Math.trunc(filter.limit)));
    return this.db.prepare(`
      SELECT * FROM events
       WHERE ${clauses.join(' AND ')}
       ORDER BY commit_seq ${order}${limit}
    `).all(...params).map((row) => eventFromRow(row));
  }

  ensureDeliveries(sessionId: string, consumerId: string): void {
    this.db.prepare(`
      INSERT OR IGNORE INTO event_deliveries(
        event_id, consumer_id, monitor_state, detail_state
      )
      SELECT
        event_id,
        ?,
        CASE
          WHEN delivery_class <> 'monitor_and_events' THEN 'not_applicable'
          WHEN p.is_current = 1 THEN 'pending'
          ELSE 'suppressed'
        END,
        CASE
          WHEN delivery_class = 'internal_only' THEN 'not_applicable'
          WHEN p.is_current = 1 AND s.is_current = 1 THEN 'pending'
          ELSE 'deferred'
        END
      FROM events e
      JOIN event_segments s ON s.segment_id = e.segment_id
      JOIN play_sessions p ON p.session_id = e.session_id
      WHERE e.session_id = ?
    `).run(consumerId, sessionId);
  }

  delivery(eventId: string, consumerId: string): EventDeliveryRecord | null {
    const event = this.eventById(eventId);
    if (!event) return null;
    this.insertInitialDelivery(event, consumerId);
    const row = this.db.prepare(`
      SELECT * FROM event_deliveries WHERE event_id = ? AND consumer_id = ?
    `).get(eventId, consumerId);
    return row ? deliveryFromRow(row) : null;
  }

  listEvents(filter: EventListFilter, consumerId: string): EventWithDelivery[] {
    this.ensureDeliveries(filter.sessionId, consumerId);
    const clauses = ['e.session_id = ?', 'd.consumer_id = ?'];
    const params: unknown[] = [filter.sessionId, consumerId];
    if (filter.segmentId) {
      clauses.push('e.segment_id = ?');
      params.push(filter.segmentId);
    }
    if (filter.excludeSegmentId) {
      clauses.push('e.segment_id <> ?');
      params.push(filter.excludeSegmentId);
    }
    this.addInFilter(clauses, params, 'e.type', filter.types);
    this.addInFilter(clauses, params, 'e.delivery_class', filter.deliveryClasses);
    this.addInFilter(clauses, params, 'd.detail_state', filter.detailStates);
    this.addInFilter(clauses, params, 'd.monitor_state', filter.monitorStates);
    if (filter.maxCommitSeq !== undefined) {
      clauses.push('e.commit_seq <= ?');
      params.push(filter.maxCommitSeq);
    }
    if (filter.maxEpochSeq !== undefined) {
      clauses.push('(e.epoch_seq IS NULL OR e.epoch_seq <= ?)');
      params.push(filter.maxEpochSeq);
    }
    const order = filter.order === 'desc' ? 'DESC' : 'ASC';
    const rows = this.db.prepare(`
      SELECT
        e.*,
        d.consumer_id,
        d.monitor_state,
        d.monitor_attempt_count,
        d.monitor_last_attempt_at,
        d.monitor_delivered_at,
        d.detail_state,
        d.detail_delivered_at,
        d.detail_delivery_via
      FROM events e
      JOIN event_deliveries d ON d.event_id = e.event_id
      WHERE ${clauses.join(' AND ')}
      ORDER BY e.commit_seq ${order}
    `).all(...params);
    return rows.map((row) => ({ event: eventFromRow(row), delivery: deliveryFromRow(row) }));
  }

  backlogSummary(sessionId: string, consumerId: string): BacklogSummary {
    const rows = this.listEvents({ sessionId, detailStates: ['deferred'], order: 'asc' }, consumerId);
    return {
      unreadCount: rows.length,
      segmentCount: new Set(rows.map((row) => row.event.segmentId)).size,
      oldestAt: rows[0]?.event.occurredAt,
      newestAt: rows.at(-1)?.event.occurredAt,
      available: rows.length > 0,
    };
  }

  transitionDetails(transition: DetailStateTransition): number {
    this.assertLeaseIfPresent(transition.lease, transition.nowMs);
    if (transition.eventIds.length === 0 || transition.from.length === 0) return 0;
    if (transition.to === 'delivered' && (!transition.deliveredAt || !transition.deliveryVia)) {
      throw new Error('Delivered event details require deliveredAt and deliveryVia.');
    }
    const params: unknown[] = [
      transition.to,
      transition.to === 'delivered' ? transition.deliveredAt : null,
      transition.to === 'delivered' ? transition.deliveryVia : null,
      transition.consumerId,
      ...transition.eventIds,
      ...transition.from,
    ];
    const result = this.db.prepare(`
      UPDATE event_deliveries
         SET detail_state = ?, detail_delivered_at = ?, detail_delivery_via = ?
       WHERE consumer_id = ?
         AND event_id IN (${placeholders(transition.eventIds.length)})
         AND detail_state IN (${placeholders(transition.from.length)})
    `).run(...params);
    return numberValue(result.changes);
  }

  transitionMonitor(transition: MonitorStateTransition): number {
    this.assertLeaseIfPresent(transition.lease, transition.nowMs);
    if (transition.eventIds.length === 0 || transition.from.length === 0) return 0;
    if (transition.to === 'notified' && !transition.deliveredAt) {
      throw new Error('Notified Monitor events require deliveredAt.');
    }
    const params: unknown[] = [
      transition.to,
      transition.to === 'notified' ? transition.deliveredAt : null,
      transition.consumerId,
      ...transition.eventIds,
      ...transition.from,
    ];
    const result = this.db.prepare(`
      UPDATE event_deliveries
         SET monitor_state = ?, monitor_delivered_at = ?
       WHERE consumer_id = ?
         AND event_id IN (${placeholders(transition.eventIds.length)})
         AND monitor_state IN (${placeholders(transition.from.length)})
    `).run(...params);
    return numberValue(result.changes);
  }

  recordMonitorAttempt(attempt: MonitorDeliveryAttempt): boolean {
    this.assertLeaseIfPresent(attempt.lease, attempt.nowMs);
    const result = this.db.prepare(`
      UPDATE event_deliveries
         SET monitor_attempt_count = monitor_attempt_count + 1,
             monitor_last_attempt_at = ?
       WHERE event_id = ?
         AND consumer_id = ?
         AND monitor_state = 'pending'
    `).run(attempt.attemptedAt, attempt.eventId, attempt.consumerId);
    return numberValue(result.changes) === 1;
  }

  acquireLease(request: AcquireLeaseRequest): ConsumerLease | null {
    const nowMs = request.nowMs ?? Date.now();
    return this.transaction(() => {
      const row = this.db.prepare(`
        SELECT * FROM consumer_leases WHERE consumer_id = ? AND channel = ?
      `).get(request.consumerId, request.channel);
      if (row && numberValue(row.expires_at) > nowMs && row.owner_id !== request.ownerId) return null;
      const lease: ConsumerLease = {
        consumerId: request.consumerId,
        channel: request.channel,
        ownerId: request.ownerId,
        expiresAt: nowMs + Math.max(1, request.ttlMs),
      };
      this.db.prepare(`
        INSERT INTO consumer_leases(consumer_id, channel, owner_id, expires_at)
        VALUES (?, ?, ?, ?)
        ON CONFLICT(consumer_id, channel) DO UPDATE SET
          owner_id = excluded.owner_id,
          expires_at = excluded.expires_at
      `).run(lease.consumerId, lease.channel, lease.ownerId, lease.expiresAt);
      return lease;
    });
  }

  leaseIsValid(lease: ConsumerLease, nowMs = Date.now()): boolean {
    const row = this.db.prepare(`
      SELECT owner_id, expires_at FROM consumer_leases WHERE consumer_id = ? AND channel = ?
    `).get(lease.consumerId, lease.channel);
    return !!row && row.owner_id === lease.ownerId && numberValue(row.expires_at) > nowMs;
  }

  releaseLease(lease: ConsumerLease): boolean {
    const result = this.db.prepare(`
      DELETE FROM consumer_leases WHERE consumer_id = ? AND channel = ? AND owner_id = ?
    `).run(lease.consumerId, lease.channel, lease.ownerId);
    return numberValue(result.changes) === 1;
  }

  enqueueMailboxItem<T = unknown>(request: EnqueueMailboxItem<T>): EnqueueMailboxResult<T> {
    return this.transaction(() => {
      const existing = this.db.prepare(`
        SELECT * FROM mailbox_items
         WHERE consumer_id = ? AND event_id = ? AND mode = ?
         LIMIT 1
      `).get(request.consumerId, request.eventId, request.mode);
      if (existing) return { inserted: false, item: mailboxItemFromRow<T>(existing) };
      if (this.db.prepare('SELECT item_id FROM mailbox_items WHERE item_id = ?').get(request.itemId)) {
        throw new Error(`Mailbox item ID already exists: ${request.itemId}`);
      }

      const event = this.eventById(request.eventId);
      if (!event) throw new Error(`Unknown Mailbox event: ${request.eventId}`);
      if (
        event.sessionId !== request.sessionId
        || event.segmentId !== request.segmentId
        || event.commitSeq !== request.eventCommitSeq
      ) {
        throw new Error(`Mailbox scope does not match Event ${request.eventId}.`);
      }
      if (event.deliveryClass !== 'monitor_and_events') {
        throw new Error(`Short Mailbox delivery is not applicable to Event ${request.eventId}.`);
      }

      const createdAt = request.createdAt ?? new Date().toISOString();
      const state = this.mailboxSessionIsCurrent(request.sessionId, request.segmentId)
        ? 'pending'
        : 'suppressed';
      this.insertInitialDelivery(event, request.consumerId);
      this.db.prepare(`
        INSERT INTO mailbox_items(
          item_id, schema_version, formatter_version, consumer_id, session_id,
          segment_id, event_id, event_commit_seq, mode, coalesce_key,
          payload_json, state, attempt_count, available_at, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)
      `).run(
        request.itemId,
        request.schemaVersion ?? 1,
        request.formatterVersion,
        request.consumerId,
        request.sessionId,
        request.segmentId,
        request.eventId,
        request.eventCommitSeq,
        request.mode,
        request.coalesceKey,
        JSON.stringify(request.payload) ?? 'null',
        state,
        request.availableAtMs ?? 0,
        createdAt,
        createdAt,
      );
      const inserted = this.db.prepare('SELECT * FROM mailbox_items WHERE item_id = ?')
        .get(request.itemId);
      if (!inserted) throw new Error(`Unable to read inserted Mailbox item ${request.itemId}.`);
      return { inserted: true, item: mailboxItemFromRow<T>(inserted) };
    });
  }

  mailboxItem<T = unknown>(itemId: string): MailboxItem<T> | null {
    const row = this.db.prepare('SELECT * FROM mailbox_items WHERE item_id = ?').get(itemId);
    return row ? mailboxItemFromRow<T>(row) : null;
  }

  mailboxItemForEvent<T = unknown>(
    consumerId: string,
    eventId: string,
    mode: MailboxDeliveryMode,
  ): MailboxItem<T> | null {
    const row = this.db.prepare(`
      SELECT * FROM mailbox_items
       WHERE consumer_id = ? AND event_id = ? AND mode = ?
       LIMIT 1
    `).get(consumerId, eventId, mode);
    return row ? mailboxItemFromRow<T>(row) : null;
  }

  peekMailbox<T = unknown>(filter: MailboxPeekFilter): MailboxItem<T>[] {
    const clauses = ['consumer_id = ?'];
    const params: unknown[] = [filter.consumerId];
    if (filter.sessionId) {
      clauses.push('session_id = ?');
      params.push(filter.sessionId);
    }
    if (filter.segmentId) {
      clauses.push('segment_id = ?');
      params.push(filter.segmentId);
    }
    this.addInFilter(clauses, params, 'event_id', filter.eventIds);
    this.addInFilter(clauses, params, 'mode', filter.modes);
    this.addInFilter(clauses, params, 'state', filter.states);
    const order = filter.order === 'desc' ? 'DESC' : 'ASC';
    const limit = filter.limit === undefined ? '' : ' LIMIT ?';
    if (filter.limit !== undefined) params.push(Math.max(0, Math.trunc(filter.limit)));
    return this.db.prepare(`
      SELECT * FROM mailbox_items
       WHERE ${clauses.join(' AND ')}
       ORDER BY event_commit_seq ${order}, item_id ${order}${limit}
    `).all(...params).map((row) => mailboxItemFromRow<T>(row));
  }

  claimMailbox<T = unknown>(request: ClaimMailboxItemsRequest): MailboxClaim<T> | null {
    const nowMs = request.nowMs ?? Date.now();
    return this.transaction(() => {
      this.suppressExpiredMailboxClaimsOutsideCurrentSession(request.consumerId, nowMs);
      if (!this.mailboxSessionIsCurrent(request.sessionId, request.segmentId)) return null;

      const rawRows = this.db.prepare(`
        SELECT m.*, d.monitor_state, d.detail_state
          FROM mailbox_items m
          JOIN event_deliveries d
            ON d.event_id = m.event_id AND d.consumer_id = m.consumer_id
         WHERE m.consumer_id = ?
           AND m.session_id = ?
           AND m.segment_id = ?
           AND m.mode = ?
           AND m.state IN ('pending', 'claimed')
         ORDER BY m.event_commit_seq ASC, m.item_id ASC
      `).all(request.consumerId, request.sessionId, request.segmentId, request.mode);

      const activeRows: Array<Record<string, unknown>> = [];
      const reconciledAt = new Date(nowMs).toISOString();
      for (const row of rawRows) {
        const state = String(row.state);
        const expired = state === 'claimed' && numberValue(row.claim_expires_at) <= nowMs;
        const matchesDelivery = request.mode === 'short'
          ? row.monitor_state === 'pending'
          : row.detail_state === 'pending' || row.detail_state === 'deferred';
        if ((state === 'pending' || expired) && !matchesDelivery) {
          this.db.prepare(`
            UPDATE mailbox_items
               SET state = 'suppressed', claim_owner = NULL, claim_token = NULL,
                   claim_expires_at = NULL, updated_at = ?
             WHERE item_id = ?
          `).run(reconciledAt, row.item_id);
          continue;
        }
        activeRows.push(row);
      }

      const first = activeRows[0];
      if (!first || !this.sqliteMailboxRowIsClaimable(first, nowMs)) return null;
      const maxItems = Math.max(1, Math.trunc(request.maxItems ?? 1));
      const selected: Array<Record<string, unknown>> = [];
      for (const row of activeRows) {
        if (selected.length >= maxItems) break;
        if (
          row.coalesce_key !== first.coalesce_key
          || row.formatter_version !== first.formatter_version
          || !this.sqliteMailboxRowIsClaimable(row, nowMs)
        ) break;
        selected.push(row);
      }
      const itemIds = selected.map((row) => String(row.item_id));
      const eventIds = selected.map((row) => String(row.event_id));
      const claimToken = randomUUID();
      const claimExpiresAtMs = nowMs + Math.max(1, request.leaseTtlMs);
      const attemptedAt = new Date(nowMs).toISOString();
      this.db.prepare(`
        UPDATE mailbox_items
           SET state = 'claimed', attempt_count = attempt_count + 1,
               last_attempt_at = ?, claim_owner = ?, claim_token = ?,
               claim_expires_at = ?, updated_at = ?
         WHERE item_id IN (${placeholders(itemIds.length)})
      `).run(
        attemptedAt,
        request.ownerId,
        claimToken,
        claimExpiresAtMs,
        attemptedAt,
        ...itemIds,
      );
      this.db.prepare(`
        UPDATE event_deliveries
           SET monitor_attempt_count = monitor_attempt_count + 1,
               monitor_last_attempt_at = ?
         WHERE consumer_id = ?
           AND monitor_state = 'pending'
           AND event_id IN (${placeholders(eventIds.length)})
      `).run(attemptedAt, request.consumerId, ...eventIds);
      const claimedRows = this.db.prepare(`
        SELECT * FROM mailbox_items
         WHERE item_id IN (${placeholders(itemIds.length)})
         ORDER BY event_commit_seq ASC, item_id ASC
      `).all(...itemIds);
      return {
        ownerId: request.ownerId,
        claimToken,
        claimExpiresAtMs,
        items: claimedRows.map((row) => mailboxItemFromRow<T>(row)),
      };
    });
  }

  ackMailboxClaim(request: AckMailboxClaimRequest): AckMailboxClaimResult {
    const nowMs = request.nowMs ?? Date.now();
    return this.transaction(() => {
      const rows = this.db.prepare(`
        SELECT * FROM mailbox_items
         WHERE claim_owner = ? AND claim_token = ?
         ORDER BY event_commit_seq ASC, item_id ASC
      `).all(request.ownerId, request.claimToken);
      if (rows.length === 0) return { status: 'claim_lost', itemIds: [], eventIds: [] };
      const itemIds = rows.map((row) => String(row.item_id));
      const eventIds = rows.map((row) => String(row.event_id));
      if (rows.every((row) => row.state === 'accepted')) {
        return { status: 'already_accepted', itemIds, eventIds };
      }
      if (
        rows.some((row) => row.state !== 'claimed')
        || rows.some((row) => numberValue(row.claim_expires_at) <= nowMs)
      ) {
        return { status: 'claim_lost', itemIds: [], eventIds: [] };
      }

      const acceptedAt = request.acceptedAt ?? new Date(nowMs).toISOString();
      const acceptanceLevel = request.acceptanceLevel ?? 'sink_emitted';
      const updatedAt = new Date(nowMs).toISOString();
      this.db.prepare(`
        UPDATE mailbox_items
           SET state = 'accepted', accepted_at = ?, acceptance_level = ?,
               receipt_id = ?, last_error_code = NULL, updated_at = ?
         WHERE item_id IN (${placeholders(itemIds.length)})
      `).run(acceptedAt, acceptanceLevel, request.receiptId ?? null, updatedAt, ...itemIds);

      const consumerId = String(rows[0].consumer_id);
      this.db.prepare(`
        UPDATE event_deliveries
           SET monitor_state = 'notified', monitor_delivered_at = ?
         WHERE consumer_id = ?
           AND monitor_state = 'pending'
           AND event_id IN (${placeholders(eventIds.length)})
      `).run(acceptedAt, consumerId, ...eventIds);
      return { status: 'accepted', itemIds, eventIds };
    });
  }

  nackMailboxClaim(request: NackMailboxClaimRequest): NackMailboxClaimResult {
    const nowMs = request.nowMs ?? Date.now();
    return this.transaction(() => {
      const rows = this.db.prepare(`
        SELECT * FROM mailbox_items
         WHERE claim_owner = ? AND claim_token = ? AND state = 'claimed'
         ORDER BY event_commit_seq ASC, item_id ASC
      `).all(request.ownerId, request.claimToken);
      if (
        rows.length === 0
        || rows.some((row) => numberValue(row.claim_expires_at) <= nowMs)
      ) return { status: 'claim_lost', itemIds: [] };
      const remainsCurrent = rows.every((row) => this.mailboxSessionIsCurrent(
        String(row.session_id),
        String(row.segment_id),
      ));
      const itemIds = rows.map((row) => String(row.item_id));
      const updatedAt = new Date(nowMs).toISOString();
      this.db.prepare(`
        UPDATE mailbox_items
           SET state = ?, available_at = ?, claim_owner = NULL,
               claim_token = NULL, claim_expires_at = NULL,
               last_error_code = ?, updated_at = ?
         WHERE item_id IN (${placeholders(itemIds.length)})
      `).run(
        remainsCurrent ? 'pending' : 'suppressed',
        request.retryAtMs,
        request.errorCode ?? null,
        updatedAt,
        ...itemIds,
      );
      if (!remainsCurrent) {
        for (const row of rows) this.suppressMailboxEventDeliveryRow(row, true);
      }
      return {
        status: remainsCurrent ? 'released' : 'suppressed',
        itemIds,
      };
    });
  }

  suppressMailboxItems(filter: SuppressMailboxItemsFilter): number {
    const nowMs = filter.nowMs ?? Date.now();
    return this.transaction(() => {
      const clauses = ['consumer_id = ?'];
      const params: unknown[] = [filter.consumerId];
      if (filter.sessionId) {
        clauses.push('session_id = ?');
        params.push(filter.sessionId);
      }
      if (filter.segmentId) {
        clauses.push('segment_id = ?');
        params.push(filter.segmentId);
      }
      if (filter.excludeSegmentId) {
        clauses.push('segment_id <> ?');
        params.push(filter.excludeSegmentId);
      }
      this.addInFilter(clauses, params, 'event_id', filter.eventIds);
      this.addInFilter(clauses, params, 'mode', filter.modes);
      if (filter.includeExpiredClaims) {
        clauses.push(`(state = 'pending' OR (state = 'claimed' AND claim_expires_at <= ?))`);
        params.push(nowMs);
      } else {
        clauses.push(`state = 'pending'`);
      }
      const rows = this.db.prepare(`
        SELECT * FROM mailbox_items WHERE ${clauses.join(' AND ')}
      `).all(...params);
      if (rows.length === 0) return 0;
      const itemIds = rows.map((row) => String(row.item_id));
      const updatedAt = new Date(nowMs).toISOString();
      this.db.prepare(`
        UPDATE mailbox_items
           SET state = 'suppressed', claim_owner = NULL, claim_token = NULL,
               claim_expires_at = NULL, updated_at = ?
         WHERE item_id IN (${placeholders(itemIds.length)})
      `).run(updatedAt, ...itemIds);
      for (const row of rows) {
        const deferDetails = !this.deliveryScopeIsCurrent(
          String(row.session_id),
          String(row.segment_id),
        );
        this.suppressMailboxEventDeliveryRow(row, deferDetails);
      }
      return rows.length;
    });
  }

  sourceCheckpoint(sourceId: string): SourceCheckpointRecord | null {
    const row = this.db.prepare('SELECT * FROM source_checkpoints WHERE source_id = ?').get(sourceId);
    return row ? checkpointFromRow(row) : null;
  }

  private configure(): void {
    this.db.exec('PRAGMA foreign_keys = ON');
    this.db.exec('PRAGMA busy_timeout = 5000');
    this.db.exec('PRAGMA journal_mode = WAL');
    this.db.exec('PRAGMA synchronous = FULL');
  }

  private migrate(): void {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS schema_migrations(
        version INTEGER PRIMARY KEY,
        applied_at TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS play_sessions(
        session_id TEXT PRIMARY KEY,
        adapter_id TEXT NOT NULL,
        account_id TEXT NOT NULL,
        resume_key TEXT,
        mode TEXT,
        status TEXT NOT NULL CHECK(status IN ('open', 'sealed')),
        is_current INTEGER NOT NULL DEFAULT 0 CHECK(is_current IN (0, 1)),
        opened_at TEXT NOT NULL,
        sealed_at TEXT
      );
      CREATE UNIQUE INDEX IF NOT EXISTS one_current_session_per_account
        ON play_sessions(adapter_id, account_id) WHERE is_current = 1;
      CREATE INDEX IF NOT EXISTS sessions_by_resume_key
        ON play_sessions(adapter_id, account_id, resume_key);

      CREATE TABLE IF NOT EXISTS event_segments(
        segment_id TEXT PRIMARY KEY,
        session_id TEXT NOT NULL REFERENCES play_sessions(session_id),
        segment_seq INTEGER NOT NULL,
        kind TEXT NOT NULL,
        scope_ref TEXT,
        status TEXT NOT NULL CHECK(status IN ('open', 'closed')),
        is_current INTEGER NOT NULL DEFAULT 0 CHECK(is_current IN (0, 1)),
        opened_at TEXT NOT NULL,
        closed_at TEXT,
        UNIQUE(session_id, segment_seq)
      );
      CREATE UNIQUE INDEX IF NOT EXISTS one_current_segment_per_session
        ON event_segments(session_id) WHERE is_current = 1;

      CREATE TABLE IF NOT EXISTS events(
        commit_seq INTEGER PRIMARY KEY AUTOINCREMENT,
        event_id TEXT NOT NULL UNIQUE,
        epoch_seq INTEGER,
        session_id TEXT NOT NULL REFERENCES play_sessions(session_id),
        segment_id TEXT NOT NULL REFERENCES event_segments(segment_id),
        type TEXT NOT NULL,
        producer TEXT NOT NULL CHECK(producer IN ('backend_push', 'local_synth')),
        producer_plugin_id TEXT,
        producer_output_index INTEGER,
        delivery_class TEXT NOT NULL CHECK(delivery_class IN ('monitor_and_events', 'events_only', 'internal_only')),
        occurred_at TEXT NOT NULL,
        received_at TEXT NOT NULL,
        payload_json TEXT NOT NULL,
        source_id TEXT NOT NULL,
        source_event_id TEXT,
        source_sequence TEXT,
        connection_id TEXT,
        replayed INTEGER NOT NULL DEFAULT 0 CHECK(replayed IN (0, 1))
      );
      CREATE UNIQUE INDEX IF NOT EXISTS events_by_remote_id
        ON events(source_id, source_event_id) WHERE source_event_id IS NOT NULL;
      CREATE INDEX IF NOT EXISTS events_by_scope
        ON events(session_id, segment_id, commit_seq);
      CREATE INDEX IF NOT EXISTS events_by_type
        ON events(session_id, type, commit_seq DESC);

      CREATE TABLE IF NOT EXISTS event_deliveries(
        event_id TEXT NOT NULL REFERENCES events(event_id),
        consumer_id TEXT NOT NULL,
        monitor_state TEXT NOT NULL CHECK(monitor_state IN ('not_applicable', 'pending', 'notified', 'suppressed')),
        monitor_attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(monitor_attempt_count >= 0),
        monitor_last_attempt_at TEXT,
        monitor_delivered_at TEXT,
        detail_state TEXT NOT NULL CHECK(detail_state IN ('not_applicable', 'pending', 'deferred', 'delivered')),
        detail_delivered_at TEXT,
        detail_delivery_via TEXT CHECK(detail_delivery_via IS NULL OR detail_delivery_via IN ('stdout_full', 'mailbox_full', 'ccl_events', 'ccl_events_backlog')),
        PRIMARY KEY(event_id, consumer_id),
        CHECK(
          (detail_state = 'delivered' AND detail_delivered_at IS NOT NULL AND detail_delivery_via IS NOT NULL)
          OR
          (detail_state <> 'delivered' AND detail_delivered_at IS NULL AND detail_delivery_via IS NULL)
        )
      );
      CREATE INDEX IF NOT EXISTS pending_details
        ON event_deliveries(consumer_id, detail_state, event_id);

      CREATE TABLE IF NOT EXISTS source_checkpoints(
        source_id TEXT PRIMARY KEY,
        checkpoint TEXT NOT NULL,
        source_sequence TEXT,
        event_id TEXT NOT NULL REFERENCES events(event_id),
        committed_at TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS consumer_leases(
        consumer_id TEXT NOT NULL,
        channel TEXT NOT NULL CHECK(channel IN ('monitor', 'details')),
        owner_id TEXT NOT NULL,
        expires_at INTEGER NOT NULL,
        PRIMARY KEY(consumer_id, channel)
      );
    `);
    this.db.prepare(`
      INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (1, ?)
    `).run(new Date().toISOString());

    const sequenceScopeMigration = this.db.prepare(`
      SELECT version FROM schema_migrations WHERE version = 2
    `).get();
    if (!sequenceScopeMigration) {
      this.transaction(() => {
        // Version 1 treated source sequence as globally unique for a source.
        // Many transports restart their sequence each connection/game, so the
        // durable identity must include the logical event scope.
        this.db.exec('DROP INDEX IF EXISTS events_by_remote_sequence');
        this.db.exec(`
          CREATE UNIQUE INDEX events_by_remote_sequence
            ON events(source_id, session_id, segment_id, source_sequence)
           WHERE source_sequence IS NOT NULL
        `);
        this.db.prepare(`
          INSERT INTO schema_migrations(version, applied_at) VALUES (2, ?)
        `).run(new Date().toISOString());
      });
    }

    const monitorAttemptMigration = this.db.prepare(`
      SELECT version FROM schema_migrations WHERE version = 4
    `).get();
    if (!monitorAttemptMigration) {
      this.transaction(() => {
        const columns = new Set(
          this.db.prepare('PRAGMA table_info(event_deliveries)')
            .all()
            .map((row) => String(row.name)),
        );
        if (!columns.has('monitor_attempt_count')) {
          this.db.exec(`
            ALTER TABLE event_deliveries
            ADD COLUMN monitor_attempt_count INTEGER NOT NULL DEFAULT 0
            CHECK(monitor_attempt_count >= 0)
          `);
        }
        if (!columns.has('monitor_last_attempt_at')) {
          this.db.exec(`
            ALTER TABLE event_deliveries
            ADD COLUMN monitor_last_attempt_at TEXT
          `);
        }
        this.db.prepare(`
          INSERT INTO schema_migrations(version, applied_at) VALUES (4, ?)
        `).run(new Date().toISOString());
      });
    }

    const mailboxMigration = this.db.prepare(`
      SELECT version FROM schema_migrations WHERE version = 5
    `).get();
    if (!mailboxMigration) {
      this.transaction(() => {
        const deliverySchema = this.db.prepare(`
          SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'event_deliveries'
        `).get();
        if (!String(deliverySchema?.sql ?? '').includes('mailbox_full')) {
          this.db.exec(`
            ALTER TABLE event_deliveries RENAME TO event_deliveries_v4;
            CREATE TABLE event_deliveries(
              event_id TEXT NOT NULL REFERENCES events(event_id),
              consumer_id TEXT NOT NULL,
              monitor_state TEXT NOT NULL CHECK(monitor_state IN ('not_applicable', 'pending', 'notified', 'suppressed')),
              monitor_attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(monitor_attempt_count >= 0),
              monitor_last_attempt_at TEXT,
              monitor_delivered_at TEXT,
              detail_state TEXT NOT NULL CHECK(detail_state IN ('not_applicable', 'pending', 'deferred', 'delivered')),
              detail_delivered_at TEXT,
              detail_delivery_via TEXT CHECK(detail_delivery_via IS NULL OR detail_delivery_via IN ('stdout_full', 'mailbox_full', 'ccl_events', 'ccl_events_backlog')),
              PRIMARY KEY(event_id, consumer_id),
              CHECK(
                (detail_state = 'delivered' AND detail_delivered_at IS NOT NULL AND detail_delivery_via IS NOT NULL)
                OR
                (detail_state <> 'delivered' AND detail_delivered_at IS NULL AND detail_delivery_via IS NULL)
              )
            );
            INSERT INTO event_deliveries(
              event_id, consumer_id, monitor_state, monitor_attempt_count,
              monitor_last_attempt_at, monitor_delivered_at, detail_state,
              detail_delivered_at, detail_delivery_via
            )
            SELECT
              event_id, consumer_id, monitor_state, monitor_attempt_count,
              monitor_last_attempt_at, monitor_delivered_at, detail_state,
              detail_delivered_at, detail_delivery_via
            FROM event_deliveries_v4;
            DROP TABLE event_deliveries_v4;
            CREATE INDEX pending_details
              ON event_deliveries(consumer_id, detail_state, event_id);
          `);
        }
        this.db.exec(`
          CREATE TABLE IF NOT EXISTS mailbox_items(
            item_id TEXT PRIMARY KEY,
            schema_version INTEGER NOT NULL CHECK(schema_version >= 1),
            formatter_version TEXT NOT NULL,
            consumer_id TEXT NOT NULL,
            session_id TEXT NOT NULL REFERENCES play_sessions(session_id),
            segment_id TEXT NOT NULL REFERENCES event_segments(segment_id),
            event_id TEXT NOT NULL REFERENCES events(event_id),
            event_commit_seq INTEGER NOT NULL,
            mode TEXT NOT NULL CHECK(mode IN ('short', 'full')),
            coalesce_key TEXT NOT NULL,
            payload_json TEXT NOT NULL,
            state TEXT NOT NULL CHECK(state IN ('pending', 'claimed', 'accepted', 'suppressed')),
            attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
            available_at INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL,
            last_attempt_at TEXT,
            claim_owner TEXT,
            claim_token TEXT,
            claim_expires_at INTEGER,
            accepted_at TEXT,
            acceptance_level TEXT CHECK(acceptance_level IS NULL OR acceptance_level IN ('sink_emitted', 'host_accepted')),
            receipt_id TEXT,
            last_error_code TEXT,
            UNIQUE(consumer_id, event_id, mode),
            CHECK(
              state <> 'claimed'
              OR (claim_owner IS NOT NULL AND claim_token IS NOT NULL AND claim_expires_at IS NOT NULL)
            ),
            CHECK(
              state <> 'accepted'
              OR (accepted_at IS NOT NULL AND acceptance_level IS NOT NULL)
            )
          );
          CREATE INDEX IF NOT EXISTS mailbox_fifo
            ON mailbox_items(
              consumer_id, session_id, segment_id, mode, event_commit_seq, item_id
            );
          CREATE INDEX IF NOT EXISTS mailbox_by_claim
            ON mailbox_items(claim_token, claim_owner, state);
        `);
        this.db.prepare(`
          INSERT INTO schema_migrations(version, applied_at) VALUES (5, ?)
        `).run(new Date().toISOString());
      });
    }

    const epochMigration = this.db.prepare(`
      SELECT version FROM schema_migrations WHERE version = 6
    `).get();
    if (!epochMigration) {
      this.transaction(() => {
        const columns = new Set(
          this.db.prepare('PRAGMA table_info(events)')
            .all()
            .map((row) => String(row.name)),
        );
        if (!columns.has('epoch_seq')) {
          this.db.exec('ALTER TABLE events ADD COLUMN epoch_seq INTEGER');
        }
        this.db.exec(`
          CREATE INDEX IF NOT EXISTS events_by_epoch
          ON events(session_id, segment_id, epoch_seq, commit_seq)
        `);
        this.db.prepare(`
          INSERT INTO schema_migrations(version, applied_at) VALUES (6, ?)
        `).run(new Date().toISOString());
      });
    }

    const pluginProducerMigration = this.db.prepare(`
      SELECT version FROM schema_migrations WHERE version = 7
    `).get();
    if (!pluginProducerMigration) {
      this.transaction(() => {
        const columns = new Set(
          this.db.prepare('PRAGMA table_info(events)')
            .all()
            .map((row) => String(row.name)),
        );
        if (!columns.has('producer_plugin_id')) {
          this.db.exec('ALTER TABLE events ADD COLUMN producer_plugin_id TEXT');
        }
        if (!columns.has('producer_output_index')) {
          this.db.exec('ALTER TABLE events ADD COLUMN producer_output_index INTEGER');
        }
        this.db.prepare(`
          INSERT INTO schema_migrations(version, applied_at) VALUES (7, ?)
        `).run(new Date().toISOString());
      });
    }
  }

  private insertInitialDelivery(event: CanonicalEvent, consumerId: string): void {
    const delivery = initialDeliveryFor(event, consumerId);
    const scope = this.db.prepare(`
      SELECT s.is_current AS segment_is_current, p.is_current AS session_is_current
        FROM event_segments s
        JOIN play_sessions p ON p.session_id = s.session_id
       WHERE s.segment_id = ? AND s.session_id = ?
    `).get(event.segmentId, event.sessionId);
    if (
      delivery.detailState === 'pending'
      && (
        numberValue(scope?.session_is_current) !== 1
        || numberValue(scope?.segment_is_current) !== 1
      )
    ) {
      delivery.detailState = 'deferred';
    }
    if (
      delivery.monitorState === 'pending'
      && numberValue(scope?.session_is_current) !== 1
    ) {
      delivery.monitorState = 'suppressed';
    }
    this.db.prepare(`
      INSERT OR IGNORE INTO event_deliveries(
        event_id, consumer_id, monitor_state, monitor_delivered_at,
        detail_state, detail_delivered_at, detail_delivery_via
      ) VALUES (?, ?, ?, ?, ?, ?, ?)
    `).run(
      delivery.eventId,
      delivery.consumerId,
      delivery.monitorState,
      delivery.monitorDeliveredAt ?? null,
      delivery.detailState,
      delivery.detailDeliveredAt ?? null,
      delivery.detailDeliveryVia ?? null,
    );
  }

  private upsertCheckpoint(checkpoint: SourceCheckpointRecord): void {
    this.db.prepare(`
      INSERT INTO source_checkpoints(source_id, checkpoint, source_sequence, event_id, committed_at)
      VALUES (?, ?, ?, ?, ?)
      ON CONFLICT(source_id) DO UPDATE SET
        checkpoint = excluded.checkpoint,
        source_sequence = excluded.source_sequence,
        event_id = excluded.event_id,
        committed_at = excluded.committed_at
    `).run(
      checkpoint.sourceId,
      checkpoint.checkpoint,
      checkpoint.sourceSequence ?? null,
      checkpoint.eventId,
      checkpoint.committedAt,
    );
  }

  private findByRemoteId(sourceId: string, remoteEventId: string): CanonicalEvent | null {
    const row = this.db.prepare(`
      SELECT * FROM events WHERE source_id = ? AND source_event_id = ? LIMIT 1
    `).get(sourceId, remoteEventId);
    return row ? eventFromRow(row) : null;
  }

  private findByRemoteSequence(
    sourceId: string,
    sessionId: string,
    segmentId: string,
    sequence: string,
  ): CanonicalEvent | null {
    const row = this.db.prepare(`
      SELECT * FROM events
       WHERE source_id = ?
         AND session_id = ?
         AND segment_id = ?
         AND source_sequence = ?
       LIMIT 1
    `).get(sourceId, sessionId, segmentId, sequence);
    return row ? eventFromRow(row) : null;
  }

  private mailboxSessionIsCurrent(sessionId: string, segmentId: string): boolean {
    return !!this.db.prepare(`
      SELECT 1
        FROM play_sessions p
        JOIN event_segments s ON s.session_id = p.session_id
       WHERE p.session_id = ?
         AND p.is_current = 1
         AND s.segment_id = ?
       LIMIT 1
    `).get(sessionId, segmentId);
  }

  private deliveryScopeIsCurrent(sessionId: string, segmentId: string): boolean {
    return !!this.db.prepare(`
      SELECT 1
        FROM play_sessions p
        JOIN event_segments s ON s.session_id = p.session_id
       WHERE p.session_id = ?
         AND p.is_current = 1
         AND s.segment_id = ?
         AND s.is_current = 1
       LIMIT 1
    `).get(sessionId, segmentId);
  }

  private retireMailboxScope(
    column: 'session_id' | 'segment_id',
    value: string,
    updatedAt: string,
  ): void {
    this.db.prepare(`
      UPDATE mailbox_items
         SET state = 'suppressed', updated_at = ?
       WHERE state = 'pending' AND ${column} = ?
    `).run(updatedAt, value);
    this.db.prepare(`
      UPDATE event_deliveries
         SET detail_state = CASE
               WHEN detail_state = 'pending' THEN 'deferred'
               ELSE detail_state
             END,
             monitor_state = CASE
               WHEN monitor_state = 'pending'
                AND NOT EXISTS (
                  SELECT 1
                    FROM mailbox_items m
                   WHERE m.event_id = event_deliveries.event_id
                     AND m.consumer_id = event_deliveries.consumer_id
                     AND m.state = 'claimed'
                ) THEN 'suppressed'
               ELSE monitor_state
             END
       WHERE event_id IN (
         SELECT event_id FROM events WHERE ${column} = ?
       )
    `).run(value);
  }

  private sqliteMailboxRowIsClaimable(row: Record<string, unknown>, nowMs: number): boolean {
    return row.state === 'pending'
      ? numberValue(row.available_at) <= nowMs
      : row.state === 'claimed' && numberValue(row.claim_expires_at) <= nowMs;
  }

  private suppressMailboxEventDeliveryRow(
    row: Record<string, unknown>,
    deferDetails: boolean,
  ): void {
    this.db.prepare(`
      UPDATE event_deliveries
         SET monitor_state = CASE
               WHEN monitor_state = 'pending' THEN 'suppressed'
               ELSE monitor_state
             END,
             detail_state = CASE
               WHEN ? = 1 AND detail_state = 'pending' THEN 'deferred'
               ELSE detail_state
             END
       WHERE event_id = ? AND consumer_id = ?
    `).run(
      booleanInt(deferDetails),
      String(row.event_id),
      String(row.consumer_id),
    );
  }

  private suppressExpiredMailboxClaimsOutsideCurrentSession(
    consumerId: string,
    nowMs: number,
  ): void {
    const rows = this.db.prepare(`
      SELECT m.*
        FROM mailbox_items m
       WHERE m.consumer_id = ?
         AND m.state = 'claimed'
         AND m.claim_expires_at <= ?
         AND NOT EXISTS (
           SELECT 1
             FROM play_sessions p
             JOIN event_segments s ON s.session_id = p.session_id
            WHERE p.session_id = m.session_id
              AND p.is_current = 1
              AND s.segment_id = m.segment_id
         )
    `).all(consumerId, nowMs);
    if (rows.length === 0) return;
    const itemIds = rows.map((row) => String(row.item_id));
    this.db.prepare(`
      UPDATE mailbox_items
         SET state = 'suppressed', claim_owner = NULL, claim_token = NULL,
             claim_expires_at = NULL, updated_at = ?
       WHERE item_id IN (${placeholders(itemIds.length)})
    `).run(new Date(nowMs).toISOString(), ...itemIds);
    for (const row of rows) this.suppressMailboxEventDeliveryRow(row, true);
  }

  private addInFilter(
    clauses: string[],
    params: unknown[],
    column: string,
    values: readonly unknown[] | undefined,
  ): void {
    if (!values) return;
    // An explicitly empty filter means "match nothing", matching the Memory
    // repository. Treating it as absent is especially dangerous for suppress
    // operations because it would broaden the update to every Mailbox item.
    if (values.length === 0) {
      clauses.push('0 = 1');
      return;
    }
    clauses.push(`${column} IN (${placeholders(values.length)})`);
    params.push(...values);
  }

  private assertLeaseIfPresent(lease: ConsumerLease | undefined, nowMs?: number): void {
    if (lease && !this.leaseIsValid(lease, nowMs)) throw new DeliveryLeaseLostError(lease);
  }

  private transaction<T>(fn: () => T): T {
    this.db.exec('BEGIN IMMEDIATE');
    try {
      const result = fn();
      this.db.exec('COMMIT');
      return result;
    } catch (error) {
      try { this.db.exec('ROLLBACK'); } catch {}
      throw error;
    }
  }
}