feat: add admin quick queue management

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

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",

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',

src/http/routes/admin/queue.ts

@@ -1,4 +1,11 @@
+import { parseCommaSeparatedList } from '@internal/parser'
 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 { MoveJobs } from '@storage/events'
 import { FastifyInstance, RequestGenericInterface } from 'fastify'
 import { FromSchema } from 'json-schema-to-ts'
@@ -7,6 +14,29 @@ import { registerApiKeyAuth } from '../../plugins/apikey'
 
 const { 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: {
     type: 'object',
@@ -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,75 @@ 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' })
+      }
+
+      const store = getQueueOverflowStore()
+      const data = await store.restore({
+        ...req.body,
+        signal: req.signals.disconnect.signal,
+      })
+      return reply.send(data)
+    }
+  )
 }

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

@@ -325,6 +325,28 @@ describe('getPgCancelConnectionTarget', () => {
 })
 
 describe('PgPoolExecutor', () => {
+  it('keeps no-signal transaction checkout on the direct pool path', async () => {
+    const addEventListener = vi.spyOn(AbortSignal.prototype, 'addEventListener')
+    const client = Object.assign(new EventEmitter(), {
+      query: vi.fn().mockResolvedValue({ rows: [] }),
+      release: vi.fn(),
+    }) as unknown as PoolClient & EventEmitter
+    const pool = {
+      connect: vi.fn().mockResolvedValue(client),
+    } as unknown as Pool
+    const executor = new PgPoolExecutor(pool)
+
+    try {
+      const transaction = await executor.beginTransaction()
+      await transaction.commit()
+
+      expect(pool.connect).toHaveBeenCalledOnce()
+      expect(addEventListener).not.toHaveBeenCalled()
+    } finally {
+      addEventListener.mockRestore()
+    }
+  })
+
   it('tracks checked-out client errors during direct queries', async () => {
     const socketError = new Error('socket reset')
     const client = Object.assign(new EventEmitter(), {
@@ -558,6 +580,50 @@ describe('PgPoolExecutor', () => {
     expect(pool.connect).not.toHaveBeenCalled()
   })
 
+  it('rejects an aborted transaction checkout and releases a late client', async () => {
+    const controller = new AbortController()
+    let resolveConnect: (client: PoolClient) => void = () => undefined
+    const connectPromise = new Promise<PoolClient>((resolve) => {
+      resolveConnect = resolve
+    })
+    const client = Object.assign(new EventEmitter(), {
+      query: vi.fn().mockResolvedValue({ rows: [] }),
+      release: vi.fn(),
+    }) as unknown as PoolClient & EventEmitter
+    const pool = {
+      connect: vi.fn(() => connectPromise),
+    } as unknown as Pool
+    const executor = new PgPoolExecutor(pool)
+
+    const transactionPromise = executor.beginTransaction({
+      signal: controller.signal,
+    })
+    controller.abort()
+
+    const result = await Promise.race([
+      transactionPromise.then(
+        () => ({ status: 'resolved' as const }),
+        (error) => ({ status: 'rejected' as const, error })
+      ),
+      waitForDrainCheck().then(() => ({ status: 'pending' as const })),
+    ])
+
+    resolveConnect(client)
+    await waitForDrainCheck()
+
+    expect(result).toMatchObject({
+      status: 'rejected',
+      error: {
+        name: 'AbortError',
+        code: 'ABORT_ERR',
+        message: 'Query was aborted',
+      },
+    })
+    expect(client.query).not.toHaveBeenCalled()
+    expect(client.release).toHaveBeenCalledOnce()
+    expect(client.release).toHaveBeenCalledWith()
+  })
+
   it('rejects aborted queries without waiting for the pg query to settle', async () => {
     const controller = new AbortController()
     const release = vi.fn()