test(connector): freeze the shipped connector schema behind an upgrade test (#34115)

Freezes the connector's shipped table definitions: later schema changes must be additive statements. Adds a baseline fixture and an upgrade test (baseline, then current schema, compared with a fresh database on SQLite and Postgres), plus the statement-shape check relay already uses. No runtime change.

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",

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<CloudDatabase> {
-  return await openInMemoryCloudDatabase(schemaStatements())
+  return await openInMemoryCloudDatabase(connectorBootStatements())
 }

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)
+}

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<void>)[] = []
+afterEach(async () => {
+  await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()))
+})
+
+async function freshStore(engine: Engine): Promise<Store> {
+  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<string[]> {
+  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<string[]> {
+  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<Engine>(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()
+  }
+})

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,

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