import type { Database } from 'bun:sqlite';
import { BUSY_TIMEOUT_MS, ensureWalMode } from './wal';

const SCHEMA_VERSION = 1;

const DDL = `
CREATE TABLE IF NOT EXISTS schema_version (
  version INTEGER PRIMARY KEY
);

CREATE TABLE IF NOT EXISTS events (
  id            INTEGER PRIMARY KEY AUTOINCREMENT,
  type          TEXT NOT NULL,
  payload       TEXT,
  emitted_at    INTEGER NOT NULL,
  emitted_by    TEXT,
  dedup_key     TEXT,
  reply_to      TEXT,
  reply_for     INTEGER REFERENCES events(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS events_type_idx ON events(type);
CREATE UNIQUE INDEX IF NOT EXISTS events_dedup_idx
  ON events(dedup_key) WHERE dedup_key IS NOT NULL;
CREATE INDEX IF NOT EXISTS events_reply_for_idx
  ON events(reply_for) WHERE reply_for IS NOT NULL;

CREATE TABLE IF NOT EXISTS subscribers (
  id              INTEGER PRIMARY KEY AUTOINCREMENT,
  name            TEXT NOT NULL UNIQUE,
  pattern         TEXT NOT NULL,
  like_pattern    TEXT NOT NULL,
  segment_count   INTEGER NOT NULL,
  handler         TEXT NOT NULL,
  max_attempts    INTEGER NOT NULL DEFAULT 3,
  timeout_ms      INTEGER NOT NULL DEFAULT 60000,
  registered_at   INTEGER NOT NULL,
  registered_by   TEXT
);

CREATE TABLE IF NOT EXISTS deliveries (
  event_id         INTEGER NOT NULL REFERENCES events(id) ON DELETE CASCADE,
  subscriber_id    INTEGER NOT NULL REFERENCES subscribers(id) ON DELETE CASCADE,
  status           TEXT NOT NULL,
  attempts         INTEGER NOT NULL DEFAULT 0,
  next_attempt_at  INTEGER,
  started_at       INTEGER,
  finished_at      INTEGER,
  last_error       TEXT,
  PRIMARY KEY (event_id, subscriber_id)
);
CREATE INDEX IF NOT EXISTS deliveries_pending_idx
  ON deliveries(next_attempt_at) WHERE status = 'pending';
CREATE INDEX IF NOT EXISTS deliveries_running_idx
  ON deliveries(started_at) WHERE status = 'running';

CREATE TABLE IF NOT EXISTS dispatcher_heartbeat (
  dispatcher_id    TEXT PRIMARY KEY,
  last_heartbeat   INTEGER NOT NULL,
  started_at       INTEGER NOT NULL,
  pid              INTEGER NOT NULL,
  version          TEXT
);

-- Last emission per timer interval (timer.tick.* events; see timer.ts).
-- Deliberately NOT derived from the events table: retention sweeps
-- delete old events, which would make a quiet bus re-tick early.
-- Additive table — IF NOT EXISTS upgrades existing buses in place, so
-- SCHEMA_VERSION stays put (a bump would brick every live bus DB).
CREATE TABLE IF NOT EXISTS timer_state (
  interval         TEXT PRIMARY KEY,
  last_emitted_at  INTEGER NOT NULL
);
`;

export function applySchema(db: Database): void {
  // Set FIRST so it covers everything below. It does not cover the WAL switch
  // (see ensureWalMode), but it does cover the concurrent DDL that follows,
  // which two processes creating the same fresh db will otherwise contend on.
  db.exec(`PRAGMA busy_timeout = ${BUSY_TIMEOUT_MS}`);
  ensureWalMode(db);
  db.exec('PRAGMA foreign_keys = ON');
  db.exec(DDL);

  // Seed-then-verify, NOT check-then-insert (#552). The previous form read the
  // row, and inserted when it found none — two processes opening the same fresh
  // db both saw no row and both inserted, so the loser died with
  // `UNIQUE constraint failed: schema_version.version`. `OR IGNORE` makes the
  // seed idempotent, and reading afterwards means the version check covers the
  // seeded and pre-existing cases identically instead of branching on which
  // happened.
  db.run('INSERT OR IGNORE INTO schema_version (version) VALUES (?)', [SCHEMA_VERSION]);

  const row = db.query<{ version: number }, []>('SELECT version FROM schema_version LIMIT 1').get();

  if (row && row.version !== SCHEMA_VERSION) {
    throw new Error(
      `event-bus: database schema version ${row.version} does not match library version ${SCHEMA_VERSION}. Migration required.`,
    );
  }
}

export { SCHEMA_VERSION };