fix: add transport-neutral database contracts

src/http/plugins/db.ts

@@ -3,7 +3,7 @@ import {
   getPostgresConnection,
   getServiceKeyUser,
   getTenantConfig,
-  PgTenantConnection,
+  type TenantConnection,
 } from '@internal/database'
 import {
   areMigrationsUpToDate,
@@ -19,7 +19,7 @@ import { getConfig, MultitenantMigrationStrategy } from '../../config'
 
 declare module 'fastify' {
   interface FastifyRequest {
-    db: PgTenantConnection
+    db: TenantConnection
     latestMigration?: keyof typeof DBMigration
   }
 }

src/http/routes/tus/lifecycle.ts

@@ -1,5 +1,5 @@
 import { SIGNED_URL_SCOPE_UPLOAD } from '@internal/auth'
-import { PgTenantConnection } from '@internal/database'
+import type { TenantConnection } from '@internal/database'
 import { ERRORS, isRenderableError } from '@internal/errors'
 import { logSchema, RequestLogContext } from '@internal/monitoring'
 import { UploadId } from '@storage/protocols/tus'
@@ -38,7 +38,7 @@ export type MultiPartRequest = http.IncomingMessage & {
   upload: RequestLogContext & {
     tenantId: string
     storage: Storage
-    db: PgTenantConnection
+    db: TenantConnection
     owner?: string
     isUpsert: boolean
     resources?: string[]

src/internal/auth/jwks/store-pg.ts

@@ -1,14 +1,14 @@
 import { getConfig } from '../../../config'
-import { PgTransaction, PgTransactionalExecutor } from '../../database/pg-connection'
+import type { DatabaseTransaction, DatabaseTransactionalExecutor } from '../../database/connection'
 import { logger, logSchema } from '../../monitoring'
 import { JWKSManagerStore, JWKStoreItem, PaginatedTenantItem } from './store'
 
 const { multitenantDatabaseQueryTimeout } = getConfig()
 
-export class JWKSManagerStorePg implements JWKSManagerStore<PgTransaction> {
-  constructor(private db: PgTransactionalExecutor) {}
+export class JWKSManagerStorePg implements JWKSManagerStore<DatabaseTransaction> {
+  constructor(private db: DatabaseTransactionalExecutor) {}
 
-  async transaction<T>(callback: (trx: PgTransaction) => Promise<T>): Promise<T> {
+  async transaction<T>(callback: (trx: DatabaseTransaction) => Promise<T>): Promise<T> {
     const trx = await this.db.beginTransaction()
 
     try {
@@ -34,7 +34,7 @@ export class JWKSManagerStorePg implements JWKSManagerStore<PgTransaction> {
     encryptedJwk: string,
     kind: string,
     idempotent = false,
-    trx?: PgTransaction
+    trx?: DatabaseTransaction
   ): Promise<string> {
     const db = trx || this.db
     const insertResult = await db.query<{ id: string }>(
@@ -89,7 +89,7 @@ export class JWKSManagerStorePg implements JWKSManagerStore<PgTransaction> {
     tenantId: string,
     id: string,
     newState: boolean,
-    trx?: PgTransaction
+    trx?: DatabaseTransaction
   ): Promise<boolean> {
     const db = trx || this.db
     const result = await db.query(
@@ -109,7 +109,11 @@ export class JWKSManagerStorePg implements JWKSManagerStore<PgTransaction> {
     return Boolean(result.rowCount && result.rowCount > 0)
   }
 
-  async listActive(tenantId: string, kind?: string, trx?: PgTransaction): Promise<JWKStoreItem[]> {
+  async listActive(
+    tenantId: string,
+    kind?: string,
+    trx?: DatabaseTransaction
+  ): Promise<JWKStoreItem[]> {
     const db = trx || this.db
     const result = await db.query<JWKStoreItem>(
       {

src/internal/database/connection.ts

@@ -0,0 +1,63 @@
+import type { QueryResult, QueryResultRow } from 'pg'
+
+/**
+ * PostgreSQL-compatible connection contracts shared by the direct pg adapter
+ * and the Database Watt transport adapter.
+ *
+ * Concrete pools, clients, and transport details do not belong in this module.
+ */
+export interface DatabaseStatement {
+  text: string
+  values?: unknown[]
+}
+
+export interface DatabaseQueryOptions {
+  signal?: AbortSignal
+}
+
+export type DatabaseQueryArgument = DatabaseQueryOptions | unknown[]
+
+export interface TransactionOptions {
+  isolation?: string
+  retry?: number
+  readOnly?: boolean
+  timeout?: number
+}
+
+export interface DatabaseExecutor {
+  query<T extends QueryResultRow = QueryResultRow>(
+    statement: string | DatabaseStatement,
+    options?: DatabaseQueryArgument
+  ): Promise<QueryResult<T>>
+}
+
+export interface DatabaseTransaction extends DatabaseExecutor {
+  isCompleted(): boolean
+  commit(): Promise<void>
+  rollback(): Promise<void>
+}
+
+export interface DatabaseTransactionalExecutor extends DatabaseExecutor {
+  beginTransaction(options?: TransactionOptions): Promise<DatabaseTransaction>
+}
+
+export interface TenantConnection extends DatabaseTransactionalExecutor {
+  readonly role: string
+  dispose(): void
+  setAbortSignal(signal: AbortSignal): void
+  getAbortSignal(): AbortSignal | undefined
+  asSuperUser(): TenantConnection
+  transaction(options?: TransactionOptions): Promise<DatabaseTransaction>
+  setScope(transaction: DatabaseExecutor): Promise<void>
+}
+
+export function isDatabaseTransaction(executor: DatabaseExecutor): executor is DatabaseTransaction {
+  return (
+    'commit' in executor &&
+    typeof executor.commit === 'function' &&
+    'rollback' in executor &&
+    typeof executor.rollback === 'function' &&
+    'isCompleted' in executor &&
+    typeof executor.isCompleted === 'function'
+  )
+}

src/internal/database/index.ts

@@ -1,4 +1,5 @@
 export * from './client'
+export * from './connection'
 export * from './migration-admin-store-pg'
 export * from './multitenant-pg'
 export * from './pg-connection'

src/internal/database/migration-admin-store-pg.test.ts

@@ -1,21 +1,22 @@
+import type { DatabaseExecutor, DatabaseStatement } from './connection'
 import { MigrationAdminStorePg } from './migration-admin-store-pg'
-import type { PgExecutor, PgStatement } from './pg-connection'
 
 function createMigrationAdminStore() {
   const query = vi.fn().mockResolvedValue({
     rows: [],
     rowCount: 1,
   })
-  const store = new MigrationAdminStorePg({ query } as unknown as PgExecutor, 'pgboss')
+  const db = { query } as unknown as DatabaseExecutor
+  const store = new MigrationAdminStorePg(db, 'pgboss')
 
   return { query, store }
 }
 
-function getLastStatement(query: ReturnType<typeof vi.fn>): PgStatement {
+function getLastStatement(query: ReturnType<typeof vi.fn>): DatabaseStatement {
   const [statement] = query.mock.calls.at(-1) || []
 
   if (!statement || typeof statement === 'string') {
-    throw new Error('Expected a PgStatement query')
+    throw new Error('Expected a DatabaseStatement query')
   }
 
   return statement

src/internal/database/migration-admin-store-pg.ts

@@ -1,5 +1,5 @@
 import { QueryResultRow } from 'pg'
-import { PgExecutor } from './pg-connection'
+import type { DatabaseExecutor } from './connection'
 import { quoteIdentifier } from './sql'
 import { TenantCursorRow } from './tenant-store-pg'
 
@@ -9,7 +9,7 @@ export class MigrationAdminStorePg {
   private readonly jobTable: string
 
   constructor(
-    private db: PgExecutor,
+    private db: DatabaseExecutor,
     pgBossSchema: string
   ) {
     this.jobTable = `${quoteIdentifier(pgBossSchema)}.job`
@@ -112,7 +112,7 @@ export class MigrationAdminStorePg {
   }
 
   private query<T extends QueryResultRow = QueryResultRow>(
-    statement: Parameters<PgExecutor['query']>[0]
+    statement: Parameters<DatabaseExecutor['query']>[0]
   ) {
     return this.db.query<T>(statement)
   }

src/internal/database/migrations/migrate.ts

@@ -8,8 +8,8 @@ import { validateMigrationHashes } from 'postgres-migrations/dist/validation'
 import SQL from 'sql-template-strings'
 import { getConfig, MultitenantMigrationStrategy } from '../../../config'
 import { logger, logSchema } from '../../monitoring'
+import type { DatabaseExecutor, DatabaseTransaction } from '../connection'
 import { multitenantPgExecutor } from '../multitenant-pg'
-import { PgExecutor, PgTransaction } from '../pg-connection'
 import { searchPath } from '../pool'
 import { getSslSettings } from '../ssl'
 import { getTenantConfig, TenantMigrationStatus } from '../tenant'
@@ -163,7 +163,7 @@ export async function updateTenantMigrationsState(
   options?: {
     migration?: keyof typeof DBMigration
     state: TenantMigrationStatus
-    tnx?: PgExecutor
+    tnx?: DatabaseExecutor
   }
 ) {
   const migrationVersion = options?.migration || (await lastLocalMigrationName())
@@ -196,7 +196,7 @@ export async function areMigrationsUpToDate(tenantId: string) {
 }
 
 export async function obtainLockOnMultitenantDB<T>(
-  fn: (tnx: PgTransaction) => Promise<T>,
+  fn: (tnx: DatabaseTransaction) => Promise<T>,
   options?: { sbReqId?: string }
 ) {
   const trx = await multitenantPgExecutor.beginTransaction()

src/internal/database/multitenant-pg.ts

@@ -1,7 +1,8 @@
 import { logger, logSchema } from '@internal/monitoring'
 import { Pool, PoolConfig } from 'pg'
 import { getConfig } from '../../config'
-import { attachPgPoolErrorHandler, PgPoolExecutor, PgTransactionalExecutor } from './pg-connection'
+import type { DatabaseTransactionalExecutor } from './connection'
+import { attachPgPoolErrorHandler, PgPoolExecutor } from './pg-connection'
 
 function buildMultitenantPgPoolConfig(config: ReturnType<typeof getConfig>): PoolConfig {
   const {
@@ -149,7 +150,7 @@ function getPoolConfigSignature(config: PoolConfig): string {
 
 const multitenantPgPoolOwner = new MultitenantPgPoolOwner()
 
-export const multitenantPgExecutor: PgTransactionalExecutor = {
+export const multitenantPgExecutor: DatabaseTransactionalExecutor = {
   async query(statement, options) {
     return multitenantPgPoolOwner.getExecutor().query(statement, options)
   },

src/internal/database/pg-connection.test.ts

@@ -5,9 +5,9 @@ import { DatabaseError, Pool as PgPool, type Pool, type PoolClient } from 'pg'
 // the same class to keep cancellation pending without opening real sockets.
 import PgConnection from 'pg/lib/connection'
 import { vi } from 'vitest'
+import type { DatabaseExecutor } from './connection'
 import {
   getPgCancelConnectionTarget,
-  type PgExecutor,
   PgPoolExecutor,
   PgPoolManager,
   PgPoolStrategy,
@@ -1536,7 +1536,7 @@ describe('PgTenantConnection', () => {
     )
     const executor = {
       query: vi.fn().mockResolvedValue({ rows: [] }),
-    } as unknown as PgExecutor
+    } as unknown as DatabaseExecutor
     const stringifySpy = vi.spyOn(JSON, 'stringify')
 
     try {
@@ -2081,7 +2081,7 @@ describe('PgTenantConnection payload serialization', () => {
     expect(superUserToJSON).not.toHaveBeenCalled()
 
     const query = vi.fn().mockResolvedValue({ rows: [] })
-    await superUser.setScope({ query } as unknown as PgExecutor)
+    await superUser.setScope({ query } as unknown as DatabaseExecutor)
 
     expect(userToJSON).not.toHaveBeenCalled()
     expect(superUserToJSON).toHaveBeenCalledTimes(1)
@@ -2119,17 +2119,19 @@ describe('PgTenantConnection payload serialization', () => {
       expect(stringifySpy.mock.calls.length).toBe(afterSuperUser)
 
       const parentQuery = vi.fn().mockResolvedValue({ rows: [] })
-      await parent.setScope({ query: parentQuery } as unknown as PgExecutor)
+      await parent.setScope({ query: parentQuery } as unknown as DatabaseExecutor)
       const afterParentScope = stringifySpy.mock.calls.length
       const siblingQuery = vi.fn().mockResolvedValue({ rows: [] })
-      await sibling.setScope({ query: siblingQuery } as unknown as PgExecutor)
+      await sibling.setScope({ query: siblingQuery } as unknown as DatabaseExecutor)
       expect(stringifySpy.mock.calls.length).toBe(afterParentScope)
 
       const superUserQuery = vi.fn().mockResolvedValue({ rows: [] })
-      await superUser.setScope({ query: superUserQuery } as unknown as PgExecutor)
+      await superUser.setScope({ query: superUserQuery } as unknown as DatabaseExecutor)
       const afterSuperUserScope = stringifySpy.mock.calls.length
       const secondSuperUserQuery = vi.fn().mockResolvedValue({ rows: [] })
-      await secondSuperUser.setScope({ query: secondSuperUserQuery } as unknown as PgExecutor)
+      await secondSuperUser.setScope({
+        query: secondSuperUserQuery,
+      } as unknown as DatabaseExecutor)
       expect(stringifySpy.mock.calls.length).toBe(afterSuperUserScope)
       expect(stringifySpy).toHaveBeenCalledWith(userPayload)
       expect(stringifySpy).toHaveBeenCalledWith(superPayload)