fixup

Signed-off-by: Paolo Insogna <paolo@cowtech.it>

Author: ShogunPanda

src/admin-app.test.ts

@@ -1,4 +1,4 @@
-import { vi } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 function mockAdminAppDependencies() {
   vi.doMock('./config', () => ({

src/database/config.ts

@@ -22,6 +22,7 @@ export type DatabaseConfig = {
   maxResultRows: number
   maxSerializedRequestBytes: number
   maxSqlBytes: number
+  poolIsExternal: boolean
   poolConnectionString?: string
   poolMode?: string
   rootCert?: string
@@ -73,6 +74,7 @@ export function readConfig(env: NodeJS.ProcessEnv = process.env): DatabaseConfig
       10 * 1024 * 1024
     ),
     maxSqlBytes: readPositiveInteger(env.DATABASE_WATT_MAX_SQL_BYTES, 1024 * 1024),
+    poolIsExternal: Boolean(env.DATABASE_POOL_URL),
     poolConnectionString: env.DATABASE_POOL_URL || env.DATABASE_URL,
     poolMode: env.DATABASE_POOL_MODE,
     rootCert: readRootCert(env.DATABASE_SSL_ROOT_CERT),

src/database/destinations.ts

@@ -36,7 +36,7 @@ export class DestinationResolver {
     return {
       connectionString,
       id: destination,
-      isExternalPool: Boolean(process.env.DATABASE_POOL_URL),
+      isExternalPool: this.config.poolIsExternal,
       maxConnections: this.config.destinationMaxConnections,
       poolMode: this.config.poolMode,
     }

src/database/pools.ts

@@ -21,13 +21,17 @@ export class PoolRegistry {
   private readonly config: DatabaseConfig
   private readonly pools = new Map<string, PoolEntry>()
   private pendingGlobalAcquisitions = 0
+  private cachedStats?: PoolRegistryStats
   private readonly evictionInterval: NodeJS.Timeout
 
   constructor(config: DatabaseConfig) {
     this.config = config
-    this.evictionInterval = setInterval(() => {
-      void this.evictIdlePools()
-    }, Math.max(config.idlePoolTimeoutMs, 1_000))
+    this.evictionInterval = setInterval(
+      () => {
+        void this.evictIdlePools()
+      },
+      Math.max(config.idlePoolTimeoutMs, 1_000)
+    )
     this.evictionInterval.unref()
   }
 
@@ -75,6 +79,10 @@ export class PoolRegistry {
   }
 
   getStats(): PoolRegistryStats {
+    if (this.cachedStats) {
+      return this.cachedStats
+    }
+
     let inUseConnections = 0
     let totalConnections = 0
     let waitingRequests = 0
@@ -85,18 +93,25 @@ export class PoolRegistry {
       waitingRequests += entry.pool.waitingCount
     }
 
-    return {
+    const stats = {
       inUseConnections,
       pools: this.pools.size,
       totalConnections,
       waitingRequests,
     }
+    this.cachedStats = stats
+    queueMicrotask(() => {
+      this.cachedStats = undefined
+    })
+
+    return stats
   }
 
   async close(): Promise<void> {
     clearInterval(this.evictionInterval)
     const pools = [...this.pools.values()].map((entry) => entry.pool)
     this.pools.clear()
+    this.cachedStats = undefined
     await Promise.allSettled(pools.map((pool) => pool.end()))
   }
 
@@ -132,6 +147,7 @@ export class PoolRegistry {
     }
 
     this.pools.set(destination.id, entry)
+    this.cachedStats = undefined
     return entry
   }
 
@@ -166,6 +182,7 @@ export class PoolRegistry {
       }
 
       this.pools.delete(destination)
+      this.cachedStats = undefined
       toEvict.push(entry.pool)
     }
 

src/database/ssl.ts

@@ -1,7 +1,7 @@
 export function getSslSettings(
   connectionString: string,
   rootCert?: string
-): { ca?: string; rejectUnauthorized?: boolean } | false | undefined {
+): { ca?: string; rejectUnauthorized?: boolean } | boolean | undefined {
   const sslMode = getSslMode(connectionString)
 
   if (sslMode === 'disable') {
@@ -14,7 +14,11 @@ export function getSslSettings(
     }
   }
 
-  if (sslMode === 'require' || sslMode === 'prefer') {
+  if (sslMode === 'require') {
+    return true
+  }
+
+  if (sslMode === 'prefer') {
     return {
       rejectUnauthorized: false,
     }

src/database/tests/destinations.test.ts

@@ -0,0 +1,20 @@
+import { describe, expect, it } from 'vitest'
+import { readConfig } from '../config.js'
+import { DestinationResolver } from '../destinations.js'
+
+describe('database destination resolution', () => {
+  it('derives single-tenant external pool state from parsed config', async () => {
+    const resolver = new DestinationResolver(
+      readConfig({
+        DATABASE_POOL_URL: 'postgres://pooler',
+      })
+    )
+
+    const destination = await resolver.resolve('default')
+
+    expect(destination).toMatchObject({
+      connectionString: 'postgres://pooler',
+      isExternalPool: true,
+    })
+  })
+})

src/database/tests/index.test.ts

@@ -3,19 +3,21 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { DatabaseErrorResponse } from '../errors.js'
 import type { ApplicationContext } from '../index.js'
 
-type Handler = (data: unknown) => Promise<unknown>
-
 type MockClient = {
   queries: Array<{ sql: string; values?: unknown[] }>
   query: ReturnType<typeof vi.fn>
   release: ReturnType<typeof vi.fn>
 }
 
 type MockedEnvironment = {
-  app: ApplicationContext,
+  app: ApplicationContext
   messaging: ReturnType<typeof setupLoopbackMessaging>
   clients: MockClient[]
-  pools: Array<{ config: Record<string, unknown>; ended: boolean; queries: Array<{ sql: string; values?: unknown[] }> }>
+  pools: Array<{
+    config: Record<string, unknown>
+    ended: boolean
+    queries: Array<{ sql: string; values?: unknown[] }>
+  }>
 }
 
 function createMockClient(rows: unknown[]): MockClient {
@@ -31,14 +33,11 @@ function createMockClient(rows: unknown[]): MockClient {
   return client
 }
 
-async function loadApp(options: {
-  env?: Record<string, string>
-  queryRows?: unknown[]
-  tenantRows?: unknown[]
-} = {}): Promise<MockedEnvironment> {
+async function loadApp(
+  options: { env?: Record<string, string>; queryRows?: unknown[]; tenantRows?: unknown[] } = {}
+): Promise<MockedEnvironment> {
   vi.resetModules()
 
-  const handlers = new Map<string, Handler>()
   const clients: MockClient[] = []
   const pools: MockedEnvironment['pools'] = []
   const queryRows = options.queryRows || [{ ok: true }]
@@ -100,7 +99,7 @@ async function loadApp(options: {
   }))
 
   // We need dynamic import due to the mocking of PostgreSQL modules above
-  const {create} = await import('../index.js')
+  const { create } = await import('../index.js')
 
   const messaging = setupLoopbackMessaging('db')
   const app = create()
@@ -120,7 +119,7 @@ describe('database Watt application messaging handlers', () => {
   it('registers prefixed handlers and exposes no server', async () => {
     const { app } = await loadApp()
 
-    expect(app.isBackgroundApplication).toBe(true)    
+    expect(app.isBackgroundApplication).toBe(true)
   })
 
   it('executes stateless queries against the single-tenant destination', async () => {
@@ -138,6 +137,19 @@ describe('database Watt application messaging handlers', () => {
     expect(clients[0].release).toHaveBeenCalledWith(undefined)
   })
 
+  it('marks single-tenant DATABASE_POOL_URL destinations as external pools', async () => {
+    const { messaging, pools } = await loadApp({
+      env: { DATABASE_POOL_URL: 'postgres://pooler' },
+    })
+
+    await messaging.send('database', 'database.query', {
+      destination: 'default',
+      sql: 'SELECT 1',
+    })
+
+    expect(pools[0].config).toMatchObject({ connectionString: 'postgres://pooler' })
+  })
+
   it('returns validation errors from malformed requests', async () => {
     const { messaging } = await loadApp()
 
@@ -200,7 +212,9 @@ describe('database Watt application messaging handlers', () => {
       lockId: begin.lockId,
       sql: 'SELECT 1',
     })
-    const commit = await messaging.send('database', 'database.commitTransaction', { lockId: begin.lockId })
+    const commit = await messaging.send('database', 'database.commitTransaction', {
+      lockId: begin.lockId,
+    })
 
     expect(commit).toEqual({ committed: true })
     expect(clients[0].queries.map((query) => query.sql)).toEqual([
@@ -300,5 +314,3 @@ describe('database Watt application messaging handlers', () => {
     })
   })
 })
-
-

src/database/tests/ssl.test.ts

@@ -0,0 +1,24 @@
+import { describe, expect, it } from 'vitest'
+import { getSslSettings } from '../ssl.js'
+
+describe('database SSL settings', () => {
+  it('disables SSL when sslmode=disable', () => {
+    expect(getSslSettings('postgres://user:pass@localhost/db?sslmode=disable')).toBe(false)
+  })
+
+  it('uses certificate verification for sslmode=require without a root cert', () => {
+    expect(getSslSettings('postgres://user:pass@localhost/db?sslmode=require')).toBe(true)
+  })
+
+  it('uses the configured root certificate when present', () => {
+    expect(getSslSettings('postgres://user:pass@localhost/db?sslmode=require', '<cert>')).toEqual({
+      ca: '<cert>',
+    })
+  })
+
+  it('keeps sslmode=prefer compatible with poolers that use self-signed certificates', () => {
+    expect(getSslSettings('postgres://user:pass@localhost/db?sslmode=prefer')).toEqual({
+      rejectUnauthorized: false,
+    })
+  })
+})

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

@@ -1,30 +1,29 @@
 import { removeGlobals, updateGlobals } from '@platformatic/globals'
-import { describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { TenantConnectionOptions } from './pool'
-import {
-  DatabaseWattPgExecutor,
-  getWattPostgresConnection,
-} from './watt-connection'
+import { DatabaseWattPgExecutor, getWattPostgresConnection } from './watt-connection'
 
 type SentWattMessage = {
   application: string
   data: Record<string, unknown>
   message: string
 }
 
-function installWattMessagingMock(
-  responses: Record<string, unknown> = {}
-): { sent: SentWattMessage[] } {
+function installWattMessagingMock(responses: Record<string, unknown> = {}): {
+  sent: SentWattMessage[]
+} {
   const sent: SentWattMessage[] = []
 
   updateGlobals({
     messaging: {
+      handle: vi.fn(),
+      notify: vi.fn(),
       send: vi.fn(async (application: string, message: string, data: Record<string, unknown>) => {
         sent.push({ application, data, message })
         const response = responses[message]
         return response instanceof Promise ? response : response
       }),
-    } as unknown as any
+    },
   })
 
   return { sent }
@@ -41,7 +40,10 @@ describe('Watt PostgreSQL connection adapter', () => {
     })
     const connection = await getWattPostgresConnection(createConnectionOptions())
 
-    const result = await connection.query<{ id: number }>({ text: 'SELECT $1::int as id', values: [1] })
+    const result = await connection.query<{ id: number }>({
+      text: 'SELECT $1::int as id',
+      values: [1],
+    })
 
     expect(result.rows).toEqual([{ id: 1 }])
     expect(result.rowCount).toBe(1)
@@ -118,13 +120,42 @@ describe('Watt PostgreSQL connection adapter', () => {
     })
   })
 
+  it('preserves timeout response codes on pg DatabaseError', async () => {
+    installWattMessagingMock({
+      'database.query': {
+        code: 'ACQUIRE_TIMEOUT',
+        message: 'Timed out acquiring database connection',
+      },
+    })
+    const connection = await getWattPostgresConnection(createConnectionOptions())
+
+    await expect(connection.query('SELECT 1')).rejects.toMatchObject({
+      code: 'ACQUIRE_TIMEOUT',
+      message: 'Timed out acquiring database connection',
+    })
+  })
+
+  it('maps server timeouts to PostgreSQL query-canceled SQLSTATE', async () => {
+    installWattMessagingMock({
+      'database.query': {
+        code: 'SERVER_TIMEOUT',
+        message: 'canceling statement due to statement timeout',
+      },
+    })
+    const connection = await getWattPostgresConnection(createConnectionOptions())
+
+    await expect(connection.query('SELECT pg_sleep(10)')).rejects.toMatchObject({
+      code: '57014',
+      message: 'canceling statement due to statement timeout',
+    })
+  })
+
   it('sends explicit cancellation when an AbortSignal fires', async () => {
     let resolveQuery!: (value: unknown) => void
     const { sent } = installWattMessagingMock({
-      'database.query':
-        new Promise((resolve) => {
-          resolveQuery = resolve
-        }),
+      'database.query': new Promise((resolve) => {
+        resolveQuery = resolve
+      }),
       'database.cancel': { cancelled: true },
     })
     const connection = await getWattPostgresConnection(createConnectionOptions())