diff --git a/cloud/apps/push/Dockerfile b/cloud/apps/push/Dockerfile index 440e89b7964..5a2c57ca61a 100644 --- a/cloud/apps/push/Dockerfile +++ b/cloud/apps/push/Dockerfile @@ -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 \ diff --git a/cloud/apps/push/package.json b/cloud/apps/push/package.json index 51c0c539d2c..d932fed48fe 100644 --- a/cloud/apps/push/package.json +++ b/cloud/apps/push/package.json @@ -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" diff --git a/cloud/apps/push/src/push-database.ts b/cloud/apps/push/src/push-database.ts index 41b6cf9696b..0684d1a3f66 100644 --- a/cloud/apps/push/src/push-database.ts +++ b/cloud/apps/push/src/push-database.ts @@ -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 - -export interface PushDatabase { - readonly dialect: 'sqlite' | 'postgres' - query(sql: string, params?: unknown[]): Promise - transaction(operation: (transaction: PushDatabase) => Promise): Promise - // 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 - // Non-blocking variant: false means another transaction holds the scope. - tryLockScope(key: string): Promise - // Shared try-lock: holders of one key coexist, and an exclusive holder excludes them all. - tryLockSharedScope(key: string): Promise - close(): Promise +export async function openPushDatabase( + input: Omit +): Promise { + 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 { - 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(operation: (transaction: PushDatabase) => Promise): Promise { - return await operation(this) - } - - // BEGIN IMMEDIATE already holds SQLite’s single writer lock. - async lockQuotaScope(): Promise {} - - async tryLockScope(): Promise { - return true - } - - async tryLockSharedScope(): Promise { - return true - } - - async close(): Promise {} -} - -class SqliteDatabase extends SqliteTransaction { - // Serialize callers: synchronous node:sqlite has no nested transactions, so overlapping BEGIN would race. - private tail: Promise = Promise.resolve() - - override async query(sql: string, params: unknown[] = []): Promise { - await this.tail - return await super.query(sql, params) - } - - override async transaction(operation: (transaction: PushDatabase) => Promise): Promise { - 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 { - 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 { - const result = await this.client.query(postgresSql(sql), params) - return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }] - } - - async transaction(operation: (transaction: PushDatabase) => Promise): Promise { - 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 { - await this.query('SELECT pg_advisory_xact_lock(hashtext(?::text))', [key]) - } - - async tryLockScope(key: string): Promise { - 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 { - const [row] = await this.query( - 'SELECT pg_try_advisory_xact_lock_shared(hashtext(?::text)) AS locked', - [key] - ) - return row?.locked === true - } - - async close(): Promise {} -} - -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 { - 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 { - 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(operation: (transaction: PushDatabase) => Promise): Promise { - 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 { - throw new Error('lock_quota_scope_requires_transaction') - } - - async tryLockScope(): Promise { - throw new Error('lock_quota_scope_requires_transaction') - } - - async tryLockSharedScope(): Promise { - throw new Error('lock_quota_scope_requires_transaction') - } - - async close(): Promise { - await this.pool.end() - } -} - -async function applySchema(database: PushDatabase): Promise { - 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 { - 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): 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 { - 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 { - 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 { + return await openInMemoryCloudDatabase(pushSchemaStatements()) } diff --git a/cloud/packages/cloud-database/package.json b/cloud/packages/cloud-database/package.json new file mode 100644 index 00000000000..206fd4f6f70 --- /dev/null +++ b/cloud/packages/cloud-database/package.json @@ -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" + } +} diff --git a/cloud/packages/cloud-database/src/cloud-database.test.ts b/cloud/packages/cloud-database/src/cloud-database.test.ts new file mode 100644 index 00000000000..72047e09865 --- /dev/null +++ b/cloud/packages/cloud-database/src/cloud-database.test.ts @@ -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() + }) +}) diff --git a/cloud/packages/cloud-database/src/cloud-database.ts b/cloud/packages/cloud-database/src/cloud-database.ts new file mode 100644 index 00000000000..7a9ef3cf205 --- /dev/null +++ b/cloud/packages/cloud-database/src/cloud-database.ts @@ -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 + +export type CloudDatabase = { + readonly dialect: 'sqlite' | 'postgres' + query(sql: string, params?: unknown[]): Promise + transaction(operation: (transaction: CloudDatabase) => Promise): Promise + // 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 + // Non-blocking variant: false means another transaction holds the scope. + tryLockScope(key: string): Promise + // Shared try-lock: holders of one key coexist, and an exclusive holder excludes them all. + tryLockSharedScope(key: string): Promise + close(): Promise +} + +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 { + 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(operation: (transaction: CloudDatabase) => Promise): Promise { + return await operation(this) + } + + // BEGIN IMMEDIATE already holds SQLite’s single writer lock. + async lockQuotaScope(): Promise {} + + async tryLockScope(): Promise { + return true + } + + async tryLockSharedScope(): Promise { + return true + } + + async close(): Promise {} +} + +class SqliteDatabase extends SqliteTransaction { + // Serialize callers: synchronous node:sqlite has no nested transactions, so overlapping BEGIN would race. + private tail: Promise = Promise.resolve() + + override async query(sql: string, params: unknown[] = []): Promise { + await this.tail + return await super.query(sql, params) + } + + override async transaction(operation: (transaction: CloudDatabase) => Promise): Promise { + 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 { + 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 { + const result = await this.client.query(postgresSql(sql), params) + return returnsRows(sql) ? (result.rows as SqlRow[]) : [{ changes: result.rowCount ?? 0 }] + } + + async transaction(operation: (transaction: CloudDatabase) => Promise): Promise { + 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 { + await this.query('SELECT pg_advisory_xact_lock(hashtext(?::text))', [key]) + } + + async tryLockScope(key: string): Promise { + 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 { + const [row] = await this.query( + 'SELECT pg_try_advisory_xact_lock_shared(hashtext(?::text)) AS locked', + [key] + ) + return row?.locked === true + } + + async close(): Promise {} +} + +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 { + 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 { + 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(operation: (transaction: CloudDatabase) => Promise): Promise { + 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 { + throw new Error('lock_quota_scope_requires_transaction') + } + + async tryLockScope(): Promise { + throw new Error('lock_quota_scope_requires_transaction') + } + + async tryLockSharedScope(): Promise { + throw new Error('lock_quota_scope_requires_transaction') + } + + async close(): Promise { + await this.pool.end() + } +} + +async function applySchema(database: CloudDatabase, statements: string[]): Promise { + 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 { + 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, 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 { + 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 { + const sqlite = new DatabaseSync(':memory:') + sqlite.exec('PRAGMA foreign_keys = ON;') + const database = new SqliteDatabase(sqlite) + await applySchema(database, schemaStatements) + return database +} diff --git a/cloud/packages/cloud-database/src/index.ts b/cloud/packages/cloud-database/src/index.ts new file mode 100644 index 00000000000..3b7f278cbc2 --- /dev/null +++ b/cloud/packages/cloud-database/src/index.ts @@ -0,0 +1,7 @@ +export { + openCloudDatabase, + openInMemoryCloudDatabase, + type CloudDatabase, + type OpenCloudDatabaseInput, + type SqlRow +} from './cloud-database.js' diff --git a/cloud/packages/cloud-database/tsconfig.build.json b/cloud/packages/cloud-database/tsconfig.build.json new file mode 100644 index 00000000000..94c84b60803 --- /dev/null +++ b/cloud/packages/cloud-database/tsconfig.build.json @@ -0,0 +1,11 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "declaration": true, + "emitDeclarationOnly": false, + "noEmit": false, + "outDir": "dist", + "rootDir": "src" + }, + "exclude": ["src/**/*.test.ts"] +} diff --git a/cloud/packages/cloud-database/tsconfig.json b/cloud/packages/cloud-database/tsconfig.json new file mode 100644 index 00000000000..a552e34dbe9 --- /dev/null +++ b/cloud/packages/cloud-database/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "noEmit": true }, + "include": ["src/**/*.ts"] +} diff --git a/cloud/pnpm-lock.yaml b/cloud/pnpm-lock.yaml index 2f7c0345468..8eeafd28169 100644 --- a/cloud/pnpm-lock.yaml +++ b/cloud/pnpm-lock.yaml @@ -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': diff --git a/config/scripts/check-test-file-runners.mjs b/config/scripts/check-test-file-runners.mjs index 4defe6df355..f4689ef484f 100644 --- a/config/scripts/check-test-file-runners.mjs +++ b/config/scripts/check-test-file-runners.mjs @@ -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',