diff --git a/src/runtime/database/sql-database-adapter.ts b/src/runtime/database/sql-database-adapter.ts index 1701910..f2eaf1c 100644 --- a/src/runtime/database/sql-database-adapter.ts +++ b/src/runtime/database/sql-database-adapter.ts @@ -46,6 +46,30 @@ export const SQLITE_JOURNAL_MODE = 'WAL'; */ export const SQLITE_BUSY_TIMEOUT_MS = 5000; +/** + * Indexes supporting watcher polling and retention cleanup. + * + * Watcher polling filters `watcher_tasks` by `status` and orders by + * `created_at`, while retention deletes rows by `created_at` combined with a + * `status` filter. The same shape applies to `webhook_events`. Auth challenges + * are pruned by `expires_at`, and idempotency keys by `created_at`. Each index + * is created with `IF NOT EXISTS` so migrations stay repeatable on both fresh + * and already-migrated databases. + */ +export const SQLITE_INDEX_STATEMENTS: readonly string[] = [ + 'CREATE INDEX IF NOT EXISTS idx_watcher_tasks_status_created_at ON watcher_tasks (status, created_at)', + 'CREATE INDEX IF NOT EXISTS idx_webhook_events_status_created_at ON webhook_events (status, created_at)', + 'CREATE INDEX IF NOT EXISTS idx_auth_challenges_expires_at ON auth_challenges (expires_at)', + 'CREATE INDEX IF NOT EXISTS idx_idempotency_keys_created_at ON idempotency_keys (created_at)', +]; + +export const POSTGRES_INDEX_STATEMENTS: readonly string[] = [ + 'CREATE INDEX IF NOT EXISTS idx_watcher_tasks_status_created_at ON watcher_tasks (status, created_at)', + 'CREATE INDEX IF NOT EXISTS idx_webhook_events_status_created_at ON webhook_events (status, created_at)', + 'CREATE INDEX IF NOT EXISTS idx_auth_challenges_expires_at ON auth_challenges (expires_at)', + 'CREATE INDEX IF NOT EXISTS idx_idempotency_keys_created_at ON idempotency_keys (created_at)', +]; + function nowIso(): string { return new Date().toISOString(); } @@ -296,6 +320,9 @@ export class SqlDatabaseAdapter implements DatabaseAdapter { "ALTER TABLE idempotency_keys ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'", ); } + for (const statement of SQLITE_INDEX_STATEMENTS) { + this.sqlite.exec(statement); + } return; } @@ -361,6 +388,9 @@ export class SqlDatabaseAdapter implements DatabaseAdapter { created_at TIMESTAMPTZ NOT NULL ); `); + for (const statement of POSTGRES_INDEX_STATEMENTS) { + await this.postgres.query(statement); + } return; } diff --git a/tests/runtime/database/migrate-single-flight.test.ts b/tests/runtime/database/migrate-single-flight.test.ts index 4778c6d..bba205c 100644 --- a/tests/runtime/database/migrate-single-flight.test.ts +++ b/tests/runtime/database/migrate-single-flight.test.ts @@ -81,9 +81,9 @@ describe('Concurrent database migrations (#600)', () => { await Promise.all(Array.from({ length: 10 }, () => adapter.migrate())); - // Six statements define one migration sequence; concurrent callers + // Ten statements define one migration sequence; concurrent callers // must share it rather than each replaying the whole sequence. - expect(client.queries).toHaveLength(6); + expect(client.queries).toHaveLength(10); expect(client.maxInFlight).toBe(1); }); @@ -98,7 +98,7 @@ describe('Concurrent database migrations (#600)', () => { const results = await Promise.all(Array.from({ length: 5 }, () => adapter.migrate())); expect(results).toEqual([undefined, undefined, undefined, undefined, undefined]); - expect(client.queries).toHaveLength(6); + expect(client.queries).toHaveLength(10); }); it('serializes SQLite migration calls into a single sequence', async () => { @@ -119,7 +119,7 @@ describe('Concurrent database migrations (#600)', () => { await Promise.all(Array.from({ length: 5 }, () => adapter.migrate())); - expect(execCount).toBe(1); + expect(execCount).toBe(5); // The schema is usable after the shared migration. await adapter.insertAuthChallenge({ @@ -174,7 +174,7 @@ describe('Concurrent database migrations (#600)', () => { expect(client.queries).toHaveLength(1); await expect(adapter.migrate()).resolves.toBeUndefined(); - expect(client.queries).toHaveLength(7); + expect(client.queries).toHaveLength(11); }); it('lets a fresh caller retry once the in-flight migration has settled', async () => { @@ -192,6 +192,6 @@ describe('Concurrent database migrations (#600)', () => { // A later, non-overlapping call runs its own sequence because the guard is // cleared after the shared sequence settles. await adapter.migrate(); - expect(client.queries).toHaveLength(12); + expect(client.queries).toHaveLength(20); }); }); diff --git a/tests/runtime/database/sql-database-adapter.test.ts b/tests/runtime/database/sql-database-adapter.test.ts index 725148f..031c2be 100644 --- a/tests/runtime/database/sql-database-adapter.test.ts +++ b/tests/runtime/database/sql-database-adapter.test.ts @@ -139,6 +139,18 @@ describe('SqlDatabaseAdapter (sqlite)', () => { }); }); + it('creates indexes for watcher and retention queries', async () => { + const sqlite = (adapter as unknown as { sqlite: Database }).sqlite; + const indexes = sqlite + .prepare("SELECT name FROM sqlite_master WHERE type = 'index'") + .all() as Array<{ name: string }>; + const indexNames = indexes.map((row) => row.name); + + expect(indexNames).toContain('idx_watcher_tasks_status_created_at'); + expect(indexNames).toContain('idx_webhook_events_status_created_at'); + expect(indexNames).toContain('idx_auth_challenges_expires_at'); + }); + it('preserves records when migrations run twice', async () => { await adapter.insertAuthChallenge({ id: 'migration-challenge', diff --git a/tests/runtime/sql-adapter-cleanup.test.ts b/tests/runtime/sql-adapter-cleanup.test.ts index 1803231..b862172 100644 --- a/tests/runtime/sql-adapter-cleanup.test.ts +++ b/tests/runtime/sql-adapter-cleanup.test.ts @@ -33,6 +33,20 @@ describe('SqlDatabaseAdapter – cleanupOldRecords (sqlite)', () => { } }); + it('creates supporting indexes for watcher and retention queries', () => { + const indexes = raw + .prepare( + `SELECT name, tbl_name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%' ORDER BY name`, + ) + .all() as Array<{ name: string; tbl_name: string }>; + const byName = new Map(indexes.map((i) => [i.name, i.tbl_name])); + + expect(byName.get('idx_watcher_tasks_status_created_at')).toBe('watcher_tasks'); + expect(byName.get('idx_webhook_events_status_created_at')).toBe('webhook_events'); + expect(byName.get('idx_auth_challenges_expires_at')).toBe('auth_challenges'); + expect(byName.get('idx_idempotency_keys_created_at')).toBe('idempotency_keys'); + }); + it('retains rows exactly on the cleanup cutoff while removing strictly older rows', async () => { const challengeExpired = `challenge-expired-${randomUUID()}`; const challengeKept = `challenge-kept-${randomUUID()}`; diff --git a/tests/runtime/sql-adapter-watcher-tasks.test.ts b/tests/runtime/sql-adapter-watcher-tasks.test.ts index aacdf67..6deceb3 100644 --- a/tests/runtime/sql-adapter-watcher-tasks.test.ts +++ b/tests/runtime/sql-adapter-watcher-tasks.test.ts @@ -7,7 +7,10 @@ import type { DatabaseAdapter } from '@/runtime/interfaces.ts'; interface SqliteHandle { exec(sql: string): void; - prepare(sql: string): { run(...params: unknown[]): unknown }; + prepare(sql: string): { + run(...params: unknown[]): unknown; + all(...params: unknown[]): unknown[]; + }; } function getSqlite(db: DatabaseAdapter): SqliteHandle { @@ -41,6 +44,13 @@ function insertWatcherRow( ); } +function listIndexes(db: DatabaseAdapter, table: string): string[] { + const rows = getSqlite(db) + .prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ?") + .all(table) as Array<{ name: string }>; + return rows.map((row) => row.name); +} + describe('SqlDatabaseAdapter – watcher task persistence and processed counts', () => { const dbUrl = makeSqliteDbUrlForTests(); const dbPath = dbUrl.startsWith('file:') ? dbUrl.slice('file:'.length) : dbUrl; @@ -253,4 +263,20 @@ describe('SqlDatabaseAdapter – watcher task persistence and processed counts', const pending = await db.listPendingWatcherTasks(1); expect(pending.length).toBeLessThanOrEqual(1); }); + + it('creates supporting indexes for watcher task queries', () => { + const indexes = listIndexes(db, 'watcher_tasks'); + expect(indexes).toContain('idx_watcher_tasks_status_created_at'); + }); + + it('uses the pending status index for watcher task lookups', () => { + const plan = getSqlite(db) + .prepare( + "EXPLAIN QUERY PLAN SELECT id, created_at FROM watcher_tasks WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1", + ) + .all() as Array<{ detail: string }>; + expect(plan.some((row) => row.detail.includes('idx_watcher_tasks_status_created_at'))).toBe( + true, + ); + }); });