refactor(cloud): share push's database layer as a package (#33714)

Moves the push gateway's database layer into a shared cloud/packages/cloud-database package (service name and schema passed in). Push's behaviour, log names, SQLite file and tests are unchanged.
2026-10-30 08:37:49 +00:00 · 2026-10-26 01:10:55 -04:00
parent d7cec1c3f8
commit 44ac5e284e
11 changed files with 484 additions and 320 deletions
@@ -6,12 +6,16 @@ RUN npm install --prefix /opt/pnpm --ignore-scripts --no-audit --no-fund "$(node
  && npm audit signatures --prefix /opt/pnpm
 COPY packages/push-contract/package.json packages/push-contract/package.json
 COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
+COPY packages/cloud-database/package.json packages/cloud-database/package.json
 COPY apps/push/package.json apps/push/package.json
 RUN pnpm install --frozen-lockfile
 COPY packages/push-contract packages/push-contract
 COPY apps/push apps/push
 COPY packages/postgres-schema packages/postgres-schema
-RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build && pnpm --filter @orca-cloud/push build
+COPY packages/cloud-database packages/cloud-database
+RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build \
+  && pnpm --filter @orca-cloud/cloud-database build \
+  && pnpm --filter @orca-cloud/push build

 FROM node:24.22.1-alpine@sha256:feb76650b7195319268c59cb2d419c732d6c499b13bf5296f16ab59eb1f0a4f3 AS runtime
 ENV NODE_ENV=production
@@ -22,9 +26,11 @@ ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
 COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./
 COPY packages/push-contract/package.json packages/push-contract/package.json
 COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
+COPY packages/cloud-database/package.json packages/cloud-database/package.json
 COPY apps/push/package.json apps/push/package.json
 COPY --from=build /app/packages/push-contract/dist packages/push-contract/dist
 COPY --from=build /app/packages/postgres-schema/dist packages/postgres-schema/dist
+COPY --from=build /app/packages/cloud-database/dist packages/cloud-database/dist
 COPY --from=build /app/apps/push/dist apps/push/dist
 RUN pnpm install --prod --frozen-lockfile --filter @orca-cloud/push... \
  && rm -rf /opt/pnpm /usr/local/lib/node_modules /opt/yarn-* /root/.cache /root/.local \
@@ -9,25 +9,24 @@
    "clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"",
    "dev": "tsx watch src/index.ts",
    "lint": "tsc -p tsconfig.json --noEmit",
-    "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build",
+    "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/push-contract build && pnpm --filter @orca-cloud/cloud-database build",
    "start": "node dist/index.js",
    "test": "vitest run",
    "typecheck": "tsc -p tsconfig.json --noEmit"
  },
  "dependencies": {
    "@hono/node-server": "^2.1.5",
-    "@orca-cloud/postgres-schema": "workspace:*",
+    "@orca-cloud/cloud-database": "workspace:*",
    "@orca-cloud/push-contract": "workspace:*",
    "google-auth-library": "^11.3.1",
    "hono": "^4.13.13",
-    "pg": "^8.23.1",
-    "pg-connection-string": "2.14.1",
    "tweetnacl": "^1.0.3",
    "zod": "^4.6.7"
  },
  "devDependencies": {
    "@types/node": "^24.19.2",
    "@types/pg": "^8.23.1",
+    "pg": "^8.23.1",
    "tsx": "^4.23.15",
    "typescript": "^7.0.2",
    "vitest": "^5.0.5"
@@ -1,313 +1,19 @@
-import { mkdirSync } from 'node:fs'
-import { join } from 'node:path'
-import { DatabaseSync } from 'node:sqlite'
-import pg from 'pg'
-import { parseIntoClientConfig } from 'pg-connection-string'
-import { applyPostgresSchema } from '@orca-cloud/postgres-schema'
+import {
+  openCloudDatabase,
+  openInMemoryCloudDatabase,
+  type CloudDatabase,
+  type OpenCloudDatabaseInput
+} from '@orca-cloud/cloud-database'
 import { pushSchemaStatements } from './push-schema.js'

-const POSTGRES_LOCK_TIMEOUT_MS = 1_000
-const POSTGRES_CONNECTION_TIMEOUT_MS = 2_000
-const POSTGRES_STATEMENT_TIMEOUT_MS = 5_000
-const POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 5_000
-const POSTGRES_TRANSACTION_ATTEMPTS = 3
-const POSTGRES_RETRY_MAX_DELAY_MS = 25
+export type { CloudDatabase as PushDatabase, SqlRow } from '@orca-cloud/cloud-database'

-export type SqlRow = Record<string, unknown>
-
-export interface PushDatabase {
-  readonly dialect: 'sqlite' | 'postgres'
-  query(sql: string, params?: unknown[]): Promise<SqlRow[]>
-  transaction<T>(operation: (transaction: PushDatabase) => Promise<T>): Promise<T>
-  // Serializes every transaction that reads then writes the same identity's
-  // quota rows. Must be called inside a transaction; it releases at commit.
-  lockQuotaScope(key: string): Promise<void>
-  // Non-blocking variant: false means another transaction holds the scope.
-  tryLockScope(key: string): Promise<boolean>
-  // Shared try-lock: holders of one key coexist, and an exclusive holder excludes them all.
-  tryLockSharedScope(key: string): Promise<boolean>
-  close(): Promise<void>
+export async function openPushDatabase(
+  input: Omit<OpenCloudDatabaseInput, 'service' | 'schemaStatements'>
+): Promise<CloudDatabase> {
+  return await openCloudDatabase({ ...input, service: 'push', schemaStatements: pushSchemaStatements() })
 }

-function postgresSql(sql: string): string {
-  let index = 0
-  return sql.replace(/\?/g, () => `$${++index}`)
-}
-
-function returnsRows(sql: string): boolean {
-  return /^\s*(select|with)/i.test(sql) || /returning/i.test(sql)
-}
-
-class SqliteTransaction implements PushDatabase {
-  readonly dialect = 'sqlite' as const
-
-  constructor(protected readonly database: DatabaseSync) {}
-
-  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
-    const statement = this.database.prepare(sql)
-    const bound = params.map((value) => (value === undefined ? null : value)) as never[]
-    if (returnsRows(sql)) return statement.all(...bound) as SqlRow[]
-    const result = statement.run(...bound)
-    return [{ changes: Number(result.changes) }]
-  }
-
-  async transaction<T>(operation: (transaction: PushDatabase) => Promise<T>): Promise<T> {
-    return await operation(this)
-  }
-
-  // BEGIN IMMEDIATE already holds SQLite’s single writer lock.
-  async lockQuotaScope(): Promise<void> {}
-
-  async tryLockScope(): Promise<boolean> {
-    return true
-  }
-
-  async tryLockSharedScope(): Promise<boolean> {
-    return true
-  }
-
-  async close(): Promise<void> {}
-}
-
-class SqliteDatabase extends SqliteTransaction {
-  // Serialize callers: synchronous node:sqlite has no nested transactions, so overlapping BEGIN would race.
-  private tail: Promise<void> = Promise.resolve()
-
-  override async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
-    await this.tail
-    return await super.query(sql, params)
-  }
-
-  override async transaction<T>(operation: (transaction: PushDatabase) => Promise<T>): Promise<T> {
-    const previous = this.tail
-    let release!: () => void
-    this.tail = new Promise((resolve) => (release = resolve))
-    await previous
-    try {
-      this.database.exec('BEGIN IMMEDIATE')
-      const transaction = new SqliteTransaction(this.database)
-      try {
-        const result = await operation(transaction)
-        this.database.exec('COMMIT')
-        return result
-      } catch (error) {
-        this.database.exec('ROLLBACK')
-        throw error
-      }
-    } finally {
-      release()
-    }
-  }
-
-  override async close(): Promise<void> {
-    await this.tail
-    this.database.close()
-  }
-}
-
-class PostgresTransaction implements PushDatabase {
-  readonly dialect = 'postgres' as const
-
-  constructor(private readonly client: pg.PoolClient) {}
-
-  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
-    const result = await this.client.query(postgresSql(sql), params)
-    return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
-  }
-
-  async transaction<T>(operation: (transaction: PushDatabase) => Promise<T>): Promise<T> {
-    return await operation(this)
-  }
-
-  // READ COMMITTED lets a concurrent count-then-insert read the same
-  // under-quota total, so the identity is serialized for the whole transaction.
-  async lockQuotaScope(key: string): Promise<void> {
-    await this.query('SELECT pg_advisory_xact_lock(hashtext(?::text))', [key])
-  }
-
-  async tryLockScope(key: string): Promise<boolean> {
-    const [row] = await this.query('SELECT pg_try_advisory_xact_lock(hashtext(?::text)) AS locked', [
-      key
-    ])
-    return row?.locked === true
-  }
-
-  async tryLockSharedScope(key: string): Promise<boolean> {
-    const [row] = await this.query(
-      'SELECT pg_try_advisory_xact_lock_shared(hashtext(?::text)) AS locked',
-      [key]
-    )
-    return row?.locked === true
-  }
-
-  async close(): Promise<void> {}
-}
-
-function retryablePostgresTransactionError(error: unknown): boolean {
-  const code = String((error as { code?: unknown }).code)
-  // 57014 statement timeout aborts transactions like lock timeout; use the same bounded retry.
-  return code === '40P01' || code === '40001' || code === '55P03' || code === '57014'
-}
-
-async function waitForPostgresRetry(): Promise<void> {
-  const delayMs = Math.floor(Math.random() * (POSTGRES_RETRY_MAX_DELAY_MS + 1))
-  await new Promise((resolve) => setTimeout(resolve, delayMs))
-}
-
-class PostgresDatabase implements PushDatabase {
-  readonly dialect = 'postgres' as const
-
-  constructor(private readonly pool: pg.Pool) {}
-
-  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
-    const client = await this.pool.connect()
-    try {
-      const result = await client.query(postgresSql(sql), params)
-      return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
-    } finally {
-      client.release()
-    }
-  }
-
-  async transaction<T>(operation: (transaction: PushDatabase) => Promise<T>): Promise<T> {
-    for (let attempt = 1; attempt <= POSTGRES_TRANSACTION_ATTEMPTS; attempt++) {
-      const client = await this.pool.connect()
-      try {
-        await client.query('BEGIN')
-        const result = await operation(new PostgresTransaction(client))
-        await client.query('COMMIT')
-        return result
-      } catch (error) {
-        await client.query('ROLLBACK').catch(() => undefined)
-        if (
-          !retryablePostgresTransactionError(error) ||
-          attempt === POSTGRES_TRANSACTION_ATTEMPTS
-        ) {
-          throw error
-        }
-        console.warn(
-          JSON.stringify({
-            event: 'orca_push_postgres_transaction_retry',
-            code: String((error as { code?: unknown }).code),
-            attempt
-          })
-        )
-      } finally {
-        client.release()
-      }
-      // A PostgreSQL transaction is unusable after an abort, so retry all work
-      // on a fresh pooled client with a small full-jitter delay.
-      await waitForPostgresRetry()
-    }
-    throw new Error('postgres_transaction_retry_exhausted')
-  }
-
-  // Outside an explicit transaction, implicit commit releases the advisory lock before any protected read.
-  async lockQuotaScope(): Promise<void> {
-    throw new Error('lock_quota_scope_requires_transaction')
-  }
-
-  async tryLockScope(): Promise<boolean> {
-    throw new Error('lock_quota_scope_requires_transaction')
-  }
-
-  async tryLockSharedScope(): Promise<boolean> {
-    throw new Error('lock_quota_scope_requires_transaction')
-  }
-
-  async close(): Promise<void> {
-    await this.pool.end()
-  }
-}
-
-async function applySchema(database: PushDatabase): Promise<void> {
-  for (const statement of pushSchemaStatements()) await database.query(statement)
-}
-
-// DDL may outlast request statement_timeout; apply once on a separate untimed connection, closed before serving, or every startup repeats the failure.
-async function applySchemaOnUntimedPool(
-  databaseUrl: string,
-  applicationName: string | undefined
-): Promise<void> {
-  const pool = new pg.Pool({
-    connectionString: databaseUrl,
-    max: 1,
-    application_name: applicationName ? `${applicationName}/schema` : undefined,
-    connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
-    statement_timeout: 0,
-    lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
-    idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
-  })
-  absorbPostgresIdleClientErrors(pool)
-  const database = new PostgresDatabase(pool)
-  try {
-    await applyPostgresSchema(pushSchemaStatements(), (statement) => database.query(statement), {
-      eventPrefix: 'orca_push_postgres_schema',
-      // No catalog pre-check: lock timeout cannot prove DDL already exists, so retain bounded retry.
-      retryLockTimeout: true
-    })
-  } finally {
-    await database.close().catch(() => undefined)
-  }
-}
-
-export function absorbPostgresIdleClientErrors(pool: Pick<pg.Pool, 'on'>): void {
-  pool.on('error', () => {
-    // Handle idle pg errors after the pool removes failed clients; otherwise transient SQL faults crash-loop the service.
-    console.warn('[orca-push] idle PostgreSQL client failed')
-  })
-}
-
-export async function openPushDatabase(input: {
-  databaseUrl?: string
-  dataDir: string
-  poolMax?: number
-  applicationName?: string
-  readOnly?: boolean
-}): Promise<PushDatabase> {
-  let database: PushDatabase
-  if (input.databaseUrl) {
-    if (!input.readOnly) await applySchemaOnUntimedPool(input.databaseUrl, input.applicationName)
-    let connection: pg.ClientConfig = { connectionString: input.databaseUrl }
-    if (input.readOnly) {
-      connection = parseIntoClientConfig(input.databaseUrl)
-      // A URL parameter must not trigger a second parse that overrides read-only options.
-      delete connection.connectionString
-      connection.options = `${connection.options ?? ''} -c default_transaction_read_only=on`.trim()
-    }
-    const pool = new pg.Pool({
-      ...connection,
-      max: input.poolMax ?? 10,
-      application_name: input.applicationName,
-      connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
-      statement_timeout: POSTGRES_STATEMENT_TIMEOUT_MS,
-      lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
-      idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
-    })
-    absorbPostgresIdleClientErrors(pool)
-    database = new PostgresDatabase(pool)
-  } else {
-    mkdirSync(input.dataDir, { recursive: true })
-    const sqlite = new DatabaseSync(join(input.dataDir, 'orca-push.sqlite'), {
-      readOnly: input.readOnly ?? false
-    })
-    if (!input.readOnly) sqlite.exec('PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;')
-    database = new SqliteDatabase(sqlite)
-  }
-  if (database.dialect === 'postgres' || input.readOnly) return database
-  try {
-    await applySchema(database)
-    return database
-  } catch (error) {
-    await database.close().catch(() => undefined)
-    throw error
-  }
-}
-
-export async function openInMemoryPushDatabase(): Promise<PushDatabase> {
-  const sqlite = new DatabaseSync(':memory:')
-  sqlite.exec('PRAGMA foreign_keys = ON;')
-  const database = new SqliteDatabase(sqlite)
-  await applySchema(database)
-  return database
+export async function openInMemoryPushDatabase(): Promise<CloudDatabase> {
+  return await openInMemoryCloudDatabase(pushSchemaStatements())
 }
@@ -0,0 +1,27 @@
+{
+  "name": "@orca-cloud/cloud-database",
+  "private": true,
+  "version": "0.0.0",
+  "type": "module",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "scripts": {
+    "build": "tsc -p tsconfig.build.json",
+    "clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"",
+    "lint": "tsc -p tsconfig.json --noEmit",
+    "pretest": "pnpm --filter @orca-cloud/postgres-schema build",
+    "test": "vitest run",
+    "typecheck": "tsc -p tsconfig.json --noEmit"
+  },
+  "dependencies": {
+    "@orca-cloud/postgres-schema": "workspace:*",
+    "pg": "^8.23.1",
+    "pg-connection-string": "2.14.1"
+  },
+  "devDependencies": {
+    "@types/node": "^24.19.2",
+    "@types/pg": "^8.23.1",
+    "typescript": "^7.0.2",
+    "vitest": "^5.0.5"
+  }
+}
@@ -0,0 +1,48 @@
+import { existsSync, mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import { openCloudDatabase, openInMemoryCloudDatabase } from './cloud-database.js'
+
+const SCHEMA = ['CREATE TABLE IF NOT EXISTS items (id TEXT PRIMARY KEY, value BIGINT NOT NULL)']
+
+describe('cloud database', () => {
+  const dirs: string[] = []
+  afterEach(() => {
+    for (const dir of dirs.splice(0)) {
+      rmSync(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('names the SQLite file after the service and applies the given schema', async () => {
+    const dataDir = mkdtempSync(join(tmpdir(), 'orca-cloud-database-'))
+    dirs.push(dataDir)
+    const database = await openCloudDatabase({ service: 'sample', schemaStatements: SCHEMA, dataDir })
+    await database.transaction(async (tx) => {
+      await tx.query('INSERT INTO items (id, value) VALUES (?, ?)', ['a', 1])
+    })
+    await database.close()
+    expect(existsSync(join(dataDir, 'orca-sample.sqlite'))).toBe(true)
+
+    const reader = await openCloudDatabase({
+      service: 'sample',
+      schemaStatements: SCHEMA,
+      dataDir,
+      readOnly: true
+    })
+    await expect(reader.query('SELECT id, value FROM items')).resolves.toEqual([{ id: 'a', value: 1 }])
+    await reader.close()
+  })
+
+  it('rolls back a failed transaction in memory', async () => {
+    const database = await openInMemoryCloudDatabase(SCHEMA)
+    await expect(
+      database.transaction(async (tx) => {
+        await tx.query('INSERT INTO items (id, value) VALUES (?, ?)', ['a', 1])
+        throw new Error('abort')
+      })
+    ).rejects.toThrow('abort')
+    await expect(database.query('SELECT id FROM items')).resolves.toEqual([])
+    await database.close()
+  })
+})
@@ -0,0 +1,332 @@
+import { mkdirSync } from 'node:fs'
+import { join } from 'node:path'
+import { DatabaseSync } from 'node:sqlite'
+import pg from 'pg'
+import { parseIntoClientConfig } from 'pg-connection-string'
+import { applyPostgresSchema } from '@orca-cloud/postgres-schema'
+
+const POSTGRES_LOCK_TIMEOUT_MS = 1_000
+const POSTGRES_CONNECTION_TIMEOUT_MS = 2_000
+const POSTGRES_STATEMENT_TIMEOUT_MS = 5_000
+const POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS = 5_000
+const POSTGRES_TRANSACTION_ATTEMPTS = 3
+const POSTGRES_RETRY_MAX_DELAY_MS = 25
+
+export type SqlRow = Record<string, unknown>
+
+export type CloudDatabase = {
+  readonly dialect: 'sqlite' | 'postgres'
+  query(sql: string, params?: unknown[]): Promise<SqlRow[]>
+  transaction<T>(operation: (transaction: CloudDatabase) => Promise<T>): Promise<T>
+  // Serializes every transaction that reads then writes the same identity's
+  // quota rows. Must be called inside a transaction; it releases at commit.
+  lockQuotaScope(key: string): Promise<void>
+  // Non-blocking variant: false means another transaction holds the scope.
+  tryLockScope(key: string): Promise<boolean>
+  // Shared try-lock: holders of one key coexist, and an exclusive holder excludes them all.
+  tryLockSharedScope(key: string): Promise<boolean>
+  close(): Promise<void>
+}
+
+function postgresSql(sql: string): string {
+  let index = 0
+  return sql.replace(/\?/g, () => `$${++index}`)
+}
+
+function returnsRows(sql: string): boolean {
+  return /^\s*(select|with)/i.test(sql) || /returning/i.test(sql)
+}
+
+class SqliteTransaction implements CloudDatabase {
+  readonly dialect = 'sqlite' as const
+
+  constructor(protected readonly database: DatabaseSync) {}
+
+  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
+    const statement = this.database.prepare(sql)
+    const bound = params.map((value) => (value === undefined ? null : value)) as never[]
+    if (returnsRows(sql)) {
+      return statement.all(...bound) as SqlRow[]
+    }
+    const result = statement.run(...bound)
+    return [{ changes: Number(result.changes) }]
+  }
+
+  async transaction<T>(operation: (transaction: CloudDatabase) => Promise<T>): Promise<T> {
+    return await operation(this)
+  }
+
+  // BEGIN IMMEDIATE already holds SQLite’s single writer lock.
+  async lockQuotaScope(): Promise<void> {}
+
+  async tryLockScope(): Promise<boolean> {
+    return true
+  }
+
+  async tryLockSharedScope(): Promise<boolean> {
+    return true
+  }
+
+  async close(): Promise<void> {}
+}
+
+class SqliteDatabase extends SqliteTransaction {
+  // Serialize callers: synchronous node:sqlite has no nested transactions, so overlapping BEGIN would race.
+  private tail: Promise<void> = Promise.resolve()
+
+  override async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
+    await this.tail
+    return await super.query(sql, params)
+  }
+
+  override async transaction<T>(operation: (transaction: CloudDatabase) => Promise<T>): Promise<T> {
+    const previous = this.tail
+    let release!: () => void
+    this.tail = new Promise((resolve) => (release = resolve))
+    await previous
+    try {
+      this.database.exec('BEGIN IMMEDIATE')
+      const transaction = new SqliteTransaction(this.database)
+      try {
+        const result = await operation(transaction)
+        this.database.exec('COMMIT')
+        return result
+      } catch (error) {
+        this.database.exec('ROLLBACK')
+        throw error
+      }
+    } finally {
+      release()
+    }
+  }
+
+  override async close(): Promise<void> {
+    await this.tail
+    this.database.close()
+  }
+}
+
+class PostgresTransaction implements CloudDatabase {
+  readonly dialect = 'postgres' as const
+
+  constructor(private readonly client: pg.PoolClient) {}
+
+  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
+    const result = await this.client.query(postgresSql(sql), params)
+    return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
+  }
+
+  async transaction<T>(operation: (transaction: CloudDatabase) => Promise<T>): Promise<T> {
+    return await operation(this)
+  }
+
+  // READ COMMITTED lets a concurrent count-then-insert read the same
+  // under-quota total, so the identity is serialized for the whole transaction.
+  async lockQuotaScope(key: string): Promise<void> {
+    await this.query('SELECT pg_advisory_xact_lock(hashtext(?::text))', [key])
+  }
+
+  async tryLockScope(key: string): Promise<boolean> {
+    const [row] = await this.query('SELECT pg_try_advisory_xact_lock(hashtext(?::text)) AS locked', [
+      key
+    ])
+    return row?.locked === true
+  }
+
+  async tryLockSharedScope(key: string): Promise<boolean> {
+    const [row] = await this.query(
+      'SELECT pg_try_advisory_xact_lock_shared(hashtext(?::text)) AS locked',
+      [key]
+    )
+    return row?.locked === true
+  }
+
+  async close(): Promise<void> {}
+}
+
+function retryablePostgresTransactionError(error: unknown): boolean {
+  const code = String((error as { code?: unknown }).code)
+  // 57014 statement timeout aborts transactions like lock timeout; use the same bounded retry.
+  return code === '40P01' || code === '40001' || code === '55P03' || code === '57014'
+}
+
+async function waitForPostgresRetry(): Promise<void> {
+  const delayMs = Math.floor(Math.random() * (POSTGRES_RETRY_MAX_DELAY_MS + 1))
+  await new Promise((resolve) => setTimeout(resolve, delayMs))
+}
+
+class PostgresDatabase implements CloudDatabase {
+  readonly dialect = 'postgres' as const
+
+  constructor(
+    private readonly pool: pg.Pool,
+    private readonly service: string
+  ) {}
+
+  async query(sql: string, params: unknown[] = []): Promise<SqlRow[]> {
+    const client = await this.pool.connect()
+    try {
+      const result = await client.query(postgresSql(sql), params)
+      return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }]
+    } finally {
+      client.release()
+    }
+  }
+
+  async transaction<T>(operation: (transaction: CloudDatabase) => Promise<T>): Promise<T> {
+    for (let attempt = 1; attempt <= POSTGRES_TRANSACTION_ATTEMPTS; attempt++) {
+      const client = await this.pool.connect()
+      try {
+        await client.query('BEGIN')
+        const result = await operation(new PostgresTransaction(client))
+        await client.query('COMMIT')
+        return result
+      } catch (error) {
+        await client.query('ROLLBACK').catch(() => undefined)
+        if (
+          !retryablePostgresTransactionError(error) ||
+          attempt === POSTGRES_TRANSACTION_ATTEMPTS
+        ) {
+          throw error
+        }
+        console.warn(
+          JSON.stringify({
+            event: `orca_${this.service}_postgres_transaction_retry`,
+            code: String((error as { code?: unknown }).code),
+            attempt
+          })
+        )
+      } finally {
+        client.release()
+      }
+      // A PostgreSQL transaction is unusable after an abort, so retry all work
+      // on a fresh pooled client with a small full-jitter delay.
+      await waitForPostgresRetry()
+    }
+    throw new Error('postgres_transaction_retry_exhausted')
+  }
+
+  // Outside an explicit transaction, implicit commit releases the advisory lock before any protected read.
+  async lockQuotaScope(): Promise<void> {
+    throw new Error('lock_quota_scope_requires_transaction')
+  }
+
+  async tryLockScope(): Promise<boolean> {
+    throw new Error('lock_quota_scope_requires_transaction')
+  }
+
+  async tryLockSharedScope(): Promise<boolean> {
+    throw new Error('lock_quota_scope_requires_transaction')
+  }
+
+  async close(): Promise<void> {
+    await this.pool.end()
+  }
+}
+
+async function applySchema(database: CloudDatabase, statements: string[]): Promise<void> {
+  for (const statement of statements) {
+    await database.query(statement)
+  }
+}
+
+// DDL may outlast request statement_timeout; apply once on a separate untimed connection, closed before serving, or every startup repeats the failure.
+async function applySchemaOnUntimedPool(
+  databaseUrl: string,
+  { applicationName, service, schemaStatements }: OpenCloudDatabaseInput
+): Promise<void> {
+  const pool = new pg.Pool({
+    connectionString: databaseUrl,
+    max: 1,
+    application_name: applicationName ? `${applicationName}/schema` : undefined,
+    connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
+    statement_timeout: 0,
+    lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
+    idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
+  })
+  absorbPostgresIdleClientErrors(pool, service)
+  const database = new PostgresDatabase(pool, service)
+  try {
+    await applyPostgresSchema(schemaStatements, (statement) => database.query(statement), {
+      eventPrefix: `orca_${service}_postgres_schema`,
+      // No catalog pre-check: lock timeout cannot prove DDL already exists, so retain bounded retry.
+      retryLockTimeout: true
+    })
+  } finally {
+    await database.close().catch(() => undefined)
+  }
+}
+
+function absorbPostgresIdleClientErrors(pool: Pick<pg.Pool, 'on'>, service: string): void {
+  pool.on('error', () => {
+    // Handle idle pg errors after the pool removes failed clients; otherwise transient SQL faults crash-loop the service.
+    console.warn(`[orca-${service}] idle PostgreSQL client failed`)
+  })
+}
+
+export type OpenCloudDatabaseInput = {
+  // Names the SQLite file and every log event, e.g. 'push' -> orca-push.sqlite, orca_push_postgres_*.
+  service: string
+  schemaStatements: string[]
+  databaseUrl?: string
+  dataDir: string
+  poolMax?: number
+  applicationName?: string
+  readOnly?: boolean
+}
+
+export async function openCloudDatabase(input: OpenCloudDatabaseInput): Promise<CloudDatabase> {
+  let database: CloudDatabase
+  if (input.databaseUrl) {
+    if (!input.readOnly) {
+      await applySchemaOnUntimedPool(input.databaseUrl, input)
+    }
+    let connection: pg.ClientConfig = { connectionString: input.databaseUrl }
+    if (input.readOnly) {
+      connection = parseIntoClientConfig(input.databaseUrl)
+      // A URL parameter must not trigger a second parse that overrides read-only options.
+      delete connection.connectionString
+      connection.options = `${connection.options ?? ''} -c default_transaction_read_only=on`.trim()
+    }
+    const pool = new pg.Pool({
+      ...connection,
+      max: input.poolMax ?? 10,
+      application_name: input.applicationName,
+      connectionTimeoutMillis: POSTGRES_CONNECTION_TIMEOUT_MS,
+      statement_timeout: POSTGRES_STATEMENT_TIMEOUT_MS,
+      lock_timeout: POSTGRES_LOCK_TIMEOUT_MS,
+      idle_in_transaction_session_timeout: POSTGRES_IDLE_TRANSACTION_TIMEOUT_MS
+    })
+    absorbPostgresIdleClientErrors(pool, input.service)
+    database = new PostgresDatabase(pool, input.service)
+  } else {
+    mkdirSync(input.dataDir, { recursive: true })
+    const sqlite = new DatabaseSync(join(input.dataDir, `orca-${input.service}.sqlite`), {
+      readOnly: input.readOnly ?? false
+    })
+    if (!input.readOnly) {
+      sqlite.exec('PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;')
+    }
+    database = new SqliteDatabase(sqlite)
+  }
+  if (database.dialect === 'postgres' || input.readOnly) {
+    return database
+  }
+  try {
+    await applySchema(database, input.schemaStatements)
+    return database
+  } catch (error) {
+    await database.close().catch(() => undefined)
+    throw error
+  }
+}
+
+export async function openInMemoryCloudDatabase(
+  schemaStatements: string[]
+): Promise<CloudDatabase> {
+  const sqlite = new DatabaseSync(':memory:')
+  sqlite.exec('PRAGMA foreign_keys = ON;')
+  const database = new SqliteDatabase(sqlite)
+  await applySchema(database, schemaStatements)
+  return database
+}
@@ -0,0 +1,7 @@
+export {
+  openCloudDatabase,
+  openInMemoryCloudDatabase,
+  type CloudDatabase,
+  type OpenCloudDatabaseInput,
+  type SqlRow
+} from './cloud-database.js'
@@ -0,0 +1,11 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "declaration": true,
+    "emitDeclarationOnly": false,
+    "noEmit": false,
+    "outDir": "dist",
+    "rootDir": "src"
+  },
+  "exclude": ["src/**/*.test.ts"]
+}
@@ -0,0 +1,5 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": { "noEmit": true },
+  "include": ["src/**/*.ts"]
+}
@@ -190,9 +190,9 @@ importers:
      '@hono/node-server':
        specifier: ^2.1.5
        version: 2.1.5(hono@4.13.13)
-      '@orca-cloud/postgres-schema':
+      '@orca-cloud/cloud-database':
        specifier: workspace:*
-        version: link:../../packages/postgres-schema
+        version: link:../../packages/cloud-database
      '@orca-cloud/push-contract':
        specifier: workspace:*
        version: link:../../packages/push-contract
@@ -202,12 +202,6 @@ importers:
      hono:
        specifier: ^4.13.13
        version: 4.13.13
-      pg:
-        specifier: ^8.23.1
-        version: 8.23.1
-      pg-connection-string:
-        specifier: 2.14.1
-        version: 2.14.1
      tweetnacl:
        specifier: ^1.0.3
        version: 1.0.3
@@ -221,6 +215,9 @@ importers:
      '@types/pg':
        specifier: ^8.23.1
        version: 8.23.1
+      pg:
+        specifier: ^8.23.1
+        version: 8.23.1
      tsx:
        specifier: ^4.23.15
        version: 4.23.15
@@ -308,6 +305,31 @@ importers:
        specifier: ^5.0.5
        version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))

+  packages/cloud-database:
+    dependencies:
+      '@orca-cloud/postgres-schema':
+        specifier: workspace:*
+        version: link:../postgres-schema
+      pg:
+        specifier: ^8.23.1
+        version: 8.23.1
+      pg-connection-string:
+        specifier: 2.14.1
+        version: 2.14.1
+    devDependencies:
+      '@types/node':
+        specifier: ^24.19.2
+        version: 24.19.2
+      '@types/pg':
+        specifier: ^8.23.1
+        version: 8.23.1
+      typescript:
+        specifier: ^7.0.2
+        version: 7.0.2
+      vitest:
+        specifier: ^5.0.5
+        version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))
+
  packages/connector-contract:
    dependencies:
      '@orca-cloud/push-contract':
@@ -66,6 +66,7 @@ const NON_UNIT_TEST_RUNNERS = [
  })),
  ...[
    'cloud/apps/relay-fence-broker',
+    'cloud/packages/cloud-database',
    'cloud/packages/connector-contract',
    'cloud/packages/postgres-schema',
    'cloud/packages/push-contract',