diff --git a/cloud/apps/connector/package.json b/cloud/apps/connector/package.json index 09fbfb72cdb..08800579cc4 100644 --- a/cloud/apps/connector/package.json +++ b/cloud/apps/connector/package.json @@ -27,6 +27,7 @@ "zod": "^4.6.7" }, "devDependencies": { + "@orca-cloud/postgres-schema": "workspace:*", "@types/node": "^24.19.2", "@types/pg": "^8.23.1", "@types/ws": "8.18.2", diff --git a/cloud/apps/connector/src/connector-database.ts b/cloud/apps/connector/src/connector-database.ts index b4b1d642621..1ee811ece8c 100644 --- a/cloud/apps/connector/src/connector-database.ts +++ b/cloud/apps/connector/src/connector-database.ts @@ -8,7 +8,10 @@ import { connectorSchemaStatements } from './connector-schema.js' import { discordSchemaStatements } from './discord-schema.js' import { slackSchemaStatements } from './slack-schema.js' -const schemaStatements = () => [ +// Production has run these CREATE TABLEs, so editing one never reaches it. Change the schema by +// appending idempotent statements (ADD COLUMN IF NOT EXISTS, nullable or defaulted) after them; +// connector-schema-upgrade.test.ts boots the shipped baseline, then this list. +export const connectorBootStatements = () => [ ...connectorSchemaStatements(), ...slackSchemaStatements(), ...discordSchemaStatements() @@ -20,10 +23,10 @@ export async function openConnectorDatabase( return await openCloudDatabase({ ...input, service: 'connector', - schemaStatements: schemaStatements() + schemaStatements: connectorBootStatements() }) } export async function openInMemoryConnectorDatabase(): Promise { - return await openInMemoryCloudDatabase(schemaStatements()) + return await openInMemoryCloudDatabase(connectorBootStatements()) } diff --git a/cloud/apps/connector/src/connector-schema-baseline.test-fixture.ts b/cloud/apps/connector/src/connector-schema-baseline.test-fixture.ts new file mode 100644 index 00000000000..2b46bd57eb7 --- /dev/null +++ b/cloud/apps/connector/src/connector-schema-baseline.test-fixture.ts @@ -0,0 +1,162 @@ +// The connector's boot statements at #34100, the schema production starts from. Never edit: the +// upgrade test boots this first, then today's statements, as an existing production database would. +const BASELINE = ` +CREATE TABLE IF NOT EXISTS connector_hosts ( + host_id TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + host_name TEXT NOT NULL, + platform TEXT NOT NULL, + runtime TEXT NOT NULL, + orca_version TEXT NOT NULL, + hello_rev BIGINT NOT NULL, + hello TEXT NOT NULL, + unsettled_due INTEGER NOT NULL, + events_seq BIGINT, + first_seen_at BIGINT NOT NULL, + last_seen_at BIGINT NOT NULL +); + +CREATE INDEX IF NOT EXISTS connector_hosts_account ON connector_hosts(account_id); + +CREATE TABLE IF NOT EXISTS connector_threads ( + source TEXT NOT NULL, + external_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + container_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + owner_user_id TEXT NOT NULL, + account_id TEXT NOT NULL, + host_id TEXT NOT NULL, + place_id TEXT NOT NULL, + pane_key TEXT, + agent TEXT, + open_turn_id BIGINT, + turn_state TEXT, + turn_stop_reason TEXT, + pending_prompt TEXT, + created_at BIGINT NOT NULL, + PRIMARY KEY (source, external_id) +); + +CREATE UNIQUE INDEX IF NOT EXISTS connector_threads_place + ON connector_threads(source, workspace_id, container_id, thread_id, owner_user_id); + +CREATE INDEX IF NOT EXISTS connector_threads_host ON connector_threads(host_id); + +CREATE TABLE IF NOT EXISTS connector_inbox ( + host_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + source TEXT NOT NULL, + external_id TEXT NOT NULL, + message_id TEXT, + seq BIGINT NOT NULL, + kind TEXT NOT NULL, + payload TEXT, + state TEXT NOT NULL, + lease_until BIGINT, + expires_at BIGINT NOT NULL, + result TEXT, + created_at BIGINT NOT NULL, + settled_at BIGINT, + PRIMARY KEY (host_id, delivery_id) +); + +CREATE INDEX IF NOT EXISTS connector_inbox_host_state ON connector_inbox(host_id, state, seq); + +CREATE INDEX IF NOT EXISTS connector_inbox_conversation + ON connector_inbox(host_id, source, external_id, seq); + +CREATE INDEX IF NOT EXISTS connector_inbox_expiry ON connector_inbox(state, expires_at); + +CREATE TABLE IF NOT EXISTS connector_seen ( + source TEXT NOT NULL, + message_key TEXT NOT NULL, + seen_at BIGINT NOT NULL, + PRIMARY KEY (source, message_key) +); + +CREATE INDEX IF NOT EXISTS connector_seen_age ON connector_seen(seen_at); + +CREATE TABLE IF NOT EXISTS connector_identities ( + source TEXT NOT NULL, + workspace_id TEXT NOT NULL, + user_id TEXT NOT NULL, + account_id TEXT NOT NULL, + default_host_id TEXT, + default_project TEXT, + container_projects TEXT NOT NULL, + linked_at BIGINT NOT NULL, + PRIMARY KEY (source, workspace_id, user_id) +); + +CREATE TABLE IF NOT EXISTS connector_thread_context ( + source TEXT NOT NULL, + external_id TEXT NOT NULL, + message_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + author_id TEXT NOT NULL, + author_name TEXT, + text TEXT NOT NULL, + created_at BIGINT NOT NULL, + PRIMARY KEY (source, external_id, message_id) +); + +CREATE INDEX IF NOT EXISTS connector_thread_context_author + ON connector_thread_context(source, workspace_id, author_id); + +CREATE TABLE IF NOT EXISTS connector_posts ( + source TEXT NOT NULL, + post_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + container_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + kind TEXT NOT NULL, + body TEXT NOT NULL, + turn_key TEXT, + rev INTEGER NOT NULL, + state TEXT NOT NULL, + attempts INTEGER NOT NULL, + queued_at BIGINT NOT NULL, + next_at BIGINT NOT NULL, + message_ref TEXT, + updated_at BIGINT NOT NULL, + PRIMARY KEY (source, post_id) +); + +CREATE INDEX IF NOT EXISTS connector_posts_due ON connector_posts(state, next_at); + +CREATE INDEX IF NOT EXISTS connector_posts_container + ON connector_posts(source, container_id, state, queued_at); + +CREATE INDEX IF NOT EXISTS connector_posts_turn ON connector_posts(source, turn_key); + +CREATE TABLE IF NOT EXISTS connector_slack_installs ( + team_id TEXT PRIMARY KEY, + bot_user_id TEXT NOT NULL, + bot_token TEXT NOT NULL, + paused INTEGER NOT NULL, + installed_at BIGINT NOT NULL +); + +CREATE TABLE IF NOT EXISTS connector_slack_users ( + team_id TEXT NOT NULL, + slack_user_id TEXT NOT NULL, + membership TEXT NOT NULL, + display_name TEXT NOT NULL, + checked_at BIGINT NOT NULL, + PRIMARY KEY (team_id, slack_user_id) +); + +CREATE TABLE IF NOT EXISTS connector_discord_gateway ( + slot INTEGER PRIMARY KEY, + session_id TEXT NOT NULL, + resume_url TEXT NOT NULL, + sequence BIGINT NOT NULL +); +` + +export function connectorSchemaBaseline(): string[] { + return BASELINE.split(';') + .map((statement) => statement.trim()) + .filter((statement) => statement.length > 0) +} diff --git a/cloud/apps/connector/src/connector-schema-upgrade.test.ts b/cloud/apps/connector/src/connector-schema-upgrade.test.ts new file mode 100644 index 00000000000..309ecd8661d --- /dev/null +++ b/cloud/apps/connector/src/connector-schema-upgrade.test.ts @@ -0,0 +1,111 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { randomUUID } from 'node:crypto' +import { openCloudDatabase, type CloudDatabase } from '@orca-cloud/cloud-database' +import { requireSchemaLockTarget } from '@orca-cloud/postgres-schema' +import pg from 'pg' +import { afterEach, describe, expect, it } from 'vitest' +import { connectorBootStatements } from './connector-database.js' +import { connectorSchemaBaseline } from './connector-schema-baseline.test-fixture.js' + +const testDatabaseUrl = process.env.ORCA_CONNECTOR_TEST_DATABASE_URL + +type Engine = 'sqlite' | 'postgres' +type Store = { databaseUrl?: string; dataDir: string } + +const cleanups: (() => Promise)[] = [] +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) +}) + +async function freshStore(engine: Engine): Promise { + if (engine === 'sqlite') { + const dataDir = mkdtempSync(join(tmpdir(), 'connector-schema-')) + cleanups.push(async () => rmSync(dataDir, { recursive: true, force: true })) + return { dataDir } + } + const admin = new pg.Client({ connectionString: testDatabaseUrl }) + await admin.connect() + const schema = `connector_${randomUUID().replaceAll('-', '')}` + await admin.query(`CREATE SCHEMA ${schema}`) + cleanups.push(async () => { + await admin.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`) + await admin.end() + }) + const url = new URL(testDatabaseUrl!) + url.searchParams.set('options', `-c search_path=${schema}`) + return { databaseUrl: url.toString(), dataDir: '' } +} + +// One boot per list, through the same path a deploy takes; returns the catalog it left. +async function boot(store: Store, ...lists: string[][]): Promise { + let database: CloudDatabase | undefined + for (const schemaStatements of lists) { + await database?.close() + database = await openCloudDatabase({ ...store, service: 'connector', schemaStatements }) + } + try { + return await catalog(database!) + } finally { + await database!.close() + } +} + +async function catalog(database: CloudDatabase): Promise { + if (database.dialect === 'postgres') { + const rows = await database.query( + `SELECT table_name || '.' || column_name || ' ' || data_type || ' ' || is_nullable || ' ' || + COALESCE(column_default, '') AS entry + FROM information_schema.columns WHERE table_schema = current_schema() + UNION ALL + SELECT regexp_replace(indexdef, ' ON \\S+\\.', ' ON ') FROM pg_indexes + WHERE schemaname = current_schema() + UNION ALL + SELECT conrelid::regclass::text || ' ' || conname || ' ' || pg_get_constraintdef(oid) + FROM pg_constraint WHERE connamespace = current_schema()::regnamespace + ORDER BY 1` + ) + return rows.map((row) => String(row.entry)) + } + // SQLite rewrites a table's stored CREATE text on ADD COLUMN, so it compares like a fresh one. + const objects = await database.query( + 'SELECT sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY type, name' + ) + return objects.map((object) => String(object.sql)) +} + +describe.each(testDatabaseUrl ? ['sqlite', 'postgres'] : ['sqlite'])( + 'connector schema upgrade on %s', + (engine) => { + it('upgrades the shipped baseline to the schema a fresh database gets', async () => { + const fresh = await boot(await freshStore(engine), connectorBootStatements()) + const upgraded = await boot( + await freshStore(engine), + connectorSchemaBaseline(), + connectorBootStatements() + ) + expect(upgraded).toEqual(fresh) + expect(fresh.length).toBeGreaterThan(0) + }) + + it.each([ + ['a column', ',\n probe TEXT,'], + ['a constraint', ' UNIQUE,'] + ])('catches %s added inside a shipped CREATE TABLE', async (_, edit) => { + const edited = connectorBootStatements().map((statement) => + statement.replace('host_name TEXT NOT NULL,', `host_name TEXT NOT NULL${edit}`) + ) + const fresh = await boot(await freshStore(engine), edited) + const upgraded = await boot(await freshStore(engine), connectorSchemaBaseline(), edited) + expect(upgraded).not.toEqual(fresh) + }) + } +) + +it('gives every boot statement a shape the Postgres applier can read', () => { + // The applier throws at boot on an ALTER or index it cannot parse; catch that without Postgres. + for (const statement of connectorBootStatements()) { + expect(() => requireSchemaLockTarget(statement)).not.toThrow() + } +}) diff --git a/cloud/apps/connector/src/connector-schema.ts b/cloud/apps/connector/src/connector-schema.ts index 93b5207bd3f..ac81fd3fdd6 100644 --- a/cloud/apps/connector/src/connector-schema.ts +++ b/cloud/apps/connector/src/connector-schema.ts @@ -1,4 +1,5 @@ // Applied at startup for both dialects, so every column type reads the same in SQLite and PostgreSQL. +// Shipped statements are frozen; see connectorBootStatements for how to change the schema. const CONNECTOR_SCHEMA = ` CREATE TABLE IF NOT EXISTS connector_hosts ( host_id TEXT PRIMARY KEY, diff --git a/cloud/pnpm-lock.yaml b/cloud/pnpm-lock.yaml index 467497ce945..c8b32ab5f15 100644 --- a/cloud/pnpm-lock.yaml +++ b/cloud/pnpm-lock.yaml @@ -212,6 +212,9 @@ importers: specifier: ^4.6.7 version: 4.6.7 devDependencies: + '@orca-cloud/postgres-schema': + specifier: workspace:* + version: link:../../packages/postgres-schema '@types/node': specifier: ^24.19.2 version: 24.19.2