feat: add queue management handlers

.github/workflows/acceptance.yml

@@ -79,7 +79,7 @@ jobs:
           ACCEPTANCE_INFRA_RESTART_SCRIPT: ${{ matrix.database == 'oriole' && 'infra:restart:ci:oriole:pgvector' || matrix.database == 'multigres' && 'infra:restart:ci:multigres' || 'infra:restart:ci' }}
           ACCEPTANCE_PROFILE: ${{ inputs.profile || 'full' }}
           ACCEPTANCE_X_FORWARDED_HOST: ${{ matrix.tenancy == 'multitenant' && 'bjhaohmqunupljrqypxz.local.dev' || '' }}
-          DATABASE_ENGINE: ${{ matrix.database == 'multigres' && 'multigres' || 'postgres' }}
+          DATABASE_ENGINE: ${{ matrix.database == 'pg' && 'postgres' || matrix.database }}
           MULTI_TENANT: ${{ matrix.tenancy == 'multitenant' && 'true' || 'false' }}
           PG_QUEUE_ENABLE: ${{ matrix.tenancy == 'multitenant' && 'true' || 'false' }}
           REQUEST_X_FORWARDED_HOST_REGEXP: ${{ matrix.tenancy == 'multitenant' && '^([a-z]{20})[.]local[.](?:com|dev)$' || '' }}

package.json

@@ -12,6 +12,9 @@
     "migrations:types": "tsx src/scripts/migrations-types.ts",
     "docs:export": "tsx ./src/scripts/export-docs.ts",
     "docs:export-ui": "tsx ./src/scripts/export-docs-ui.ts",
+    "jobs:list": "tsx src/scripts/jobs-client.ts list",
+    "jobs:backup": "tsx src/scripts/jobs-client.ts backup",
+    "jobs:restore": "tsx src/scripts/jobs-client.ts restore",
     "pprof:capture": "tsx src/scripts/pprof-client.ts",
     "test:dummy-data": "tsx -r dotenv/config ./src/test/db/import-dummy-data.ts",
     "test:unit": "vitest run --config vitest.unit.config.ts",
@@ -24,12 +27,12 @@
     "acceptance:run": "tsx acceptance/scripts/run.ts",
     "acceptance:typecheck": "tsc -p acceptance/tsconfig.json --noEmit",
     "test": "npm run test:unit && npm run infra:restart && npm run test:dummy-data && npm run test:integration",
-    "test:oriole": "npm run infra:restart:oriole && npm run test:dummy-data && npm run test:integration",
+    "test:oriole": "npm run infra:restart:oriole && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole npm run test:integration",
     "test:multigres": "npm run infra:restart:multigres && DATABASE_ENGINE=multigres npm run test:dummy-data && DATABASE_ENGINE=multigres npm run test:integration",
     "test:coverage": "npm run infra:restart && npm run test:dummy-data && npm run test:integration:coverage",
     "test:coverage:ci": "npm run infra:restart:ci && npm run test:dummy-data && npm run test:integration:coverage",
-    "test:oriole:ci": "npm run infra:restart:ci:oriole && npm run test:dummy-data && npm run test:integration",
-    "test:vector:oriole:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:oriole:pgvector && npm run test:dummy-data && VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts",
+    "test:oriole:ci": "npm run infra:restart:ci:oriole && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole npm run test:integration",
+    "test:vector:oriole:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:oriole:pgvector && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts",
     "test:vector:multigres:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_DATABASE_CREATE=false VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:multigres && npm run test:dummy-data && VECTOR_BUCKET_PROVIDER=pgvector VECTOR_DATABASE_CREATE=false VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts",
     "test:multigres:ci": "npm run infra:restart:ci:multigres && DATABASE_ENGINE=multigres npm run test:dummy-data && DATABASE_ENGINE=multigres npm run test:integration",
     "infra:stop": "docker compose --project-directory . --profile monitoring -f ./.docker/docker-compose-infra.yml down --remove-orphans",

src/config.test.ts

@@ -7,6 +7,7 @@ const CONFIG_ENV_KEYS = [
   'TENANT_POOL_CACHE_HIT_LOG_SAMPLE_RATE',
   'TENANT_POOL_CACHE_MISS_LOG_SAMPLE_RATE',
   'DATABASE_POOL_DRAIN_TIMEOUT',
+  'DATABASE_ENGINE',
   'REQUEST_HARD_LIMITS_ENABLED',
   'STORAGE_S3_REQUEST_CHECKSUM_CALCULATION',
   'STORAGE_S3_RESPONSE_CHECKSUM_VALIDATION',
@@ -62,6 +63,19 @@ describe('tenant pool cache config parsing', () => {
     expect(config.requestHardLimitsEnabled).toBe(false)
   })
 
+  test.each([
+    'postgres',
+    'oriole',
+    'multigres',
+  ] as const)('parses the %s database engine', async (databaseEngine) => {
+    setConfigEnv({ DATABASE_ENGINE: databaseEngine })
+
+    const { getConfig } = await import('./config')
+    const config = getConfig({ reload: true })
+
+    expect(config.databaseEngine).toBe(databaseEngine)
+  })
+
   test('parses request hard limits as disabled by default', async () => {
     setConfigEnv({})
 

src/config.ts

@@ -5,7 +5,7 @@ import { SignJWT } from 'jose'
 export type StorageBackendType = 'file' | 's3'
 export type VectorBucketProvider = 's3' | 'pgvector'
 export type IcebergCatalogAuthType = 'sigv4' | 'token'
-export type DatabaseEngine = 'postgres' | 'multigres'
+export type DatabaseEngine = 'postgres' | 'oriole' | 'multigres'
 export type StorageS3ChecksumConfig = 'WHEN_SUPPORTED' | 'WHEN_REQUIRED'
 const DEFAULT_S3_UPLOAD_PART_SIZE = 16 * 1024 * 1024
 const MIN_S3_UPLOAD_PART_SIZE = 5 * 1024 * 1024
@@ -256,11 +256,13 @@ export function normalizeDatabaseEngine(engine: string | null | undefined): Data
     return 'postgres'
   }
 
-  if (engine === 'postgres' || engine === 'multigres') {
+  if (engine === 'postgres' || engine === 'oriole' || engine === 'multigres') {
     return engine
   }
 
-  throw new Error(`Invalid database engine "${engine}". Expected "postgres" or "multigres".`)
+  throw new Error(
+    `Invalid database engine "${engine}". Expected "postgres", "oriole", or "multigres".`
+  )
 }
 
 function normalizeStorageS3ChecksumConfig(

src/http/routes/admin/queue.test.ts

@@ -19,7 +19,8 @@ describe('admin queue routes', () => {
   it('passes sbReqId to the move jobs task', async () => {
     vi.resetModules()
 
-    const { mergeConfig } = await import('../../../config')
+    const { getConfig, mergeConfig } = await import('../../../config')
+    getConfig()
     mergeConfig({
       pgQueueEnable: true,
       adminApiKeys: 'test-admin-key',
@@ -71,7 +72,8 @@ describe('admin queue routes', () => {
   it('rejects move jobs requests without queue names', async () => {
     vi.resetModules()
 
-    const { mergeConfig } = await import('../../../config')
+    const { getConfig, mergeConfig } = await import('../../../config')
+    getConfig()
     mergeConfig({
       pgQueueEnable: true,
       adminApiKeys: 'test-admin-key',
@@ -100,4 +102,42 @@ describe('admin queue routes', () => {
       await app.close()
     }
   })
+
+  it('rejects queue overflow restore on OrioleDB', async () => {
+    vi.resetModules()
+
+    const { getConfig, mergeConfig } = await import('../../../config')
+    getConfig()
+    mergeConfig({
+      pgQueueEnable: true,
+      adminApiKeys: 'test-admin-key',
+      databaseEngine: 'oriole',
+    })
+
+    const fastify = (await import('fastify')).default
+    const { default: routes } = await import('./queue')
+
+    const app = fastify()
+    app.register(routes, { prefix: '/queue' })
+
+    try {
+      const response = await app.inject({
+        method: 'POST',
+        url: '/queue/overflow/restore',
+        headers: {
+          apikey: 'test-admin-key',
+        },
+        payload: {
+          name: 'webhooks',
+        },
+      })
+
+      expect(response.statusCode).toBe(400)
+      expect(response.json()).toEqual({
+        message: 'Queue overflow restore is not supported on OrioleDB',
+      })
+    } finally {
+      await app.close()
+    }
+  })
 })

src/http/routes/admin/queue.ts

@@ -1,11 +1,41 @@
 import { SYSTEM_TENANT } from '@internal/queue/constants'
+import {
+  JOB_OVERFLOW_LIST_LIMIT_DEFAULT,
+  JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT,
+  QueueOverflowStorePg,
+} from '@internal/queue/overflow'
+import { Queue } from '@internal/queue/queue'
+import { parseCommaSeparatedList } from '@internal/strings'
 import { MoveJobs } from '@storage/events'
 import { FastifyInstance, RequestGenericInterface } from 'fastify'
 import { FromSchema } from 'json-schema-to-ts'
 import { getConfig } from '../../../config'
 import { registerApiKeyAuth } from '../../plugins/apikey'
 
-const { pgQueueEnable } = getConfig()
+const { databaseEngine, pgQueueEnable } = getConfig()
+
+function getQueueOverflowStore() {
+  return new QueueOverflowStorePg(Queue.getDb())
+}
+
+const nonBlankStringSchema = {
+  type: 'string',
+  minLength: 1,
+  pattern: '\\S',
+} as const
+
+const stringListSchema = {
+  type: 'array',
+  minItems: 1,
+  maxItems: 1000,
+  items: nonBlankStringSchema,
+} as const
+
+const positiveSafeIntegerSchema = {
+  type: 'integer',
+  minimum: 1,
+  maximum: Number.MAX_SAFE_INTEGER,
+} as const
 
 const moveJobsSchema = {
   body: {
@@ -26,10 +56,106 @@ const moveJobsSchema = {
   },
 } as const
 
+const listQueueOverflowSchema = {
+  description: 'List created pgBoss jobs from the live queue table or overflow backup table.',
+  querystring: {
+    type: 'object',
+    properties: {
+      source: {
+        type: 'string',
+        enum: ['job', 'backup'],
+        default: 'job',
+      },
+      groupBy: {
+        type: 'string',
+        enum: ['summary', 'tenant'],
+        default: 'summary',
+      },
+      name: nonBlankStringSchema,
+      eventTypes: {
+        ...nonBlankStringSchema,
+        description: 'Comma-separated event types to filter on.',
+      },
+      tenantRefs: {
+        ...nonBlankStringSchema,
+        description: 'Comma-separated tenant refs to filter on.',
+      },
+      limit: {
+        ...positiveSafeIntegerSchema,
+        default: JOB_OVERFLOW_LIST_LIMIT_DEFAULT,
+      },
+    },
+    additionalProperties: false,
+  },
+} as const
+
+const countQueueOverflowSchema = {
+  description: 'Count created pgBoss jobs in the live queue table.',
+} as const
+
+const backupQueueOverflowSchema = {
+  description: 'Move created pgBoss jobs into the overflow backup table.',
+  body: {
+    type: 'object',
+    properties: {
+      name: nonBlankStringSchema,
+      eventTypes: stringListSchema,
+      tenantRefs: stringListSchema,
+      limit: positiveSafeIntegerSchema,
+      confirmAll: {
+        type: 'boolean',
+        description: 'Required when no queue, event type, or tenant filter is supplied.',
+      },
+    },
+    anyOf: [
+      { required: ['name'] },
+      { required: ['eventTypes'] },
+      { required: ['tenantRefs'] },
+      {
+        properties: {
+          confirmAll: { type: 'boolean', enum: [true] },
+        },
+        required: ['confirmAll'],
+      },
+    ],
+    additionalProperties: false,
+  },
+} as const
+
+const restoreQueueOverflowSchema = {
+  description:
+    'Restore created pgBoss jobs from the overflow backup table in batches. Conflicting rows are dropped from the backup table because the live job table wins.',
+  body: {
+    type: 'object',
+    properties: {
+      name: nonBlankStringSchema,
+      eventTypes: stringListSchema,
+      tenantRefs: stringListSchema,
+      limit: {
+        ...positiveSafeIntegerSchema,
+        default: JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT,
+      },
+    },
+    additionalProperties: false,
+  },
+} as const
+
 interface MoveJobsRequestInterface extends RequestGenericInterface {
   Body: FromSchema<typeof moveJobsSchema.body>
 }
 
+interface ListQueueOverflowRequestInterface extends RequestGenericInterface {
+  Querystring: FromSchema<typeof listQueueOverflowSchema.querystring>
+}
+
+interface BackupQueueOverflowRequestInterface extends RequestGenericInterface {
+  Body: FromSchema<typeof backupQueueOverflowSchema.body>
+}
+
+interface RestoreQueueOverflowRequestInterface extends RequestGenericInterface {
+  Body: FromSchema<typeof restoreQueueOverflowSchema.body>
+}
+
 export default async function routes(fastify: FastifyInstance) {
   registerApiKeyAuth(fastify)
 
@@ -56,4 +182,81 @@ export default async function routes(fastify: FastifyInstance) {
       return reply.send({ message: 'Move jobs scheduled' })
     }
   )
+
+  fastify.get<ListQueueOverflowRequestInterface>(
+    '/overflow',
+    { schema: { ...listQueueOverflowSchema, tags: ['queue'] } },
+    async (req, reply) => {
+      if (!pgQueueEnable) {
+        return reply.status(400).send({ message: 'Queue is not enabled' })
+      }
+
+      const store = getQueueOverflowStore()
+      const data = await store.list({
+        source: req.query.source,
+        groupBy: req.query.groupBy,
+        name: req.query.name,
+        eventTypes: parseCommaSeparatedList(req.query.eventTypes),
+        tenantRefs: parseCommaSeparatedList(req.query.tenantRefs),
+        limit: req.query.limit,
+        signal: req.signals.disconnect.signal,
+      })
+
+      return reply.send(data)
+    }
+  )
+
+  fastify.get(
+    '/overflow/count',
+    { schema: { ...countQueueOverflowSchema, tags: ['queue'] } },
+    async (req, reply) => {
+      if (!pgQueueEnable) {
+        return reply.status(400).send({ message: 'Queue is not enabled' })
+      }
+
+      const store = getQueueOverflowStore()
+      const data = await store.countCreated({ signal: req.signals.disconnect.signal })
+      return reply.send(data)
+    }
+  )
+
+  fastify.post<BackupQueueOverflowRequestInterface>(
+    '/overflow/backup',
+    { schema: { ...backupQueueOverflowSchema, tags: ['queue'] } },
+    async (req, reply) => {
+      if (!pgQueueEnable) {
+        return reply.status(400).send({ message: 'Queue is not enabled' })
+      }
+
+      const store = getQueueOverflowStore()
+      const data = await store.backup({
+        ...req.body,
+        signal: req.signals.disconnect.signal,
+      })
+      return reply.send(data)
+    }
+  )
+
+  fastify.post<RestoreQueueOverflowRequestInterface>(
+    '/overflow/restore',
+    { schema: { ...restoreQueueOverflowSchema, tags: ['queue'] } },
+    async (req, reply) => {
+      if (!pgQueueEnable) {
+        return reply.status(400).send({ message: 'Queue is not enabled' })
+      }
+
+      if (databaseEngine === 'oriole') {
+        return reply
+          .status(400)
+          .send({ message: 'Queue overflow restore is not supported on OrioleDB' })
+      }
+
+      const store = getQueueOverflowStore()
+      const data = await store.restore({
+        ...req.body,
+        signal: req.signals.disconnect.signal,
+      })
+      return reply.send(data)
+    }
+  )
 }