29148700684

Committed 11 Jul 2026 10:02AM UTC coverage: 79.824% (+0.4%) from 79.376%

Build # 29148700684

Build Type

Pull #1043

github

Committed by

web-flow

Commit Message

Merge be6c21cdc into b6b815fbd

Pull Request Pull Request #1043: feat: add queue management handlers

Coverage Stats

5537 of 7500 branches covered (73.83%)

Branch coverage included in aggregate %.

282 of 311 new or added lines in 11 files covered. (90.68%)

10767 of 12925 relevant lines covered (83.3%)

417.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.36

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

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 { 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: {
    type: 'object',
    properties: {
      fromQueue: {
        type: 'string',
      },
      toQueue: {
        type: 'string',
      },
      deleteJobsFromOriginalQueue: {
        type: 'boolean',
        default: false,
      },
    },
    required: ['fromQueue', 'toQueue'],
  },
} 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)

  fastify.post<MoveJobsRequestInterface>(
    '/move',
    { schema: { ...moveJobsSchema, tags: ['queue'] } },
    async (req, reply) => {
      if (!pgQueueEnable) {
        return reply.status(400).send({ message: 'Queue is not enabled' })
      }

      const fromQueue = req.body.fromQueue
      const toQueue = req.body.toQueue
      const deleteJobsFromOriginalQueue = req.body.deleteJobsFromOriginalQueue || false

      await MoveJobs.send({
        fromQueue,
        toQueue,
        deleteJobsFromOriginalQueue,
        sbReqId: req.sbReqId,
        tenant: SYSTEM_TENANT,
      })

      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)
    }
  )
}

1	import { SYSTEM_TENANT } from '@internal/queue/constants'
2	import {
3	JOB_OVERFLOW_LIST_LIMIT_DEFAULT,
4	JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT,
5	QueueOverflowStorePg,
6	} from '@internal/queue/overflow'
7	import { Queue } from '@internal/queue/queue'
8	import { parseCommaSeparatedList } from '@internal/strings'
9	import { MoveJobs } from '@storage/events'
10	import { FastifyInstance, RequestGenericInterface } from 'fastify'
11	import { FromSchema } from 'json-schema-to-ts'
12	import { getConfig } from '../../../config'
13	import { registerApiKeyAuth } from '../../plugins/apikey'
14
15	const { databaseEngine, pgQueueEnable } = getConfig()	36✔
16
17	function getQueueOverflowStore() {
18	return new QueueOverflowStorePg(Queue.getDb())	16✔
19	}
20
21	const nonBlankStringSchema = {	36✔
22	type: 'string',
23	minLength: 1,
24	pattern: '\\S',
25	} as const
26
27	const stringListSchema = {	36✔
28	type: 'array',
29	minItems: 1,
30	maxItems: 1000,
31	items: nonBlankStringSchema,
32	} as const
33
34	const positiveSafeIntegerSchema = {	36✔
35	type: 'integer',
36	minimum: 1,
37	maximum: Number.MAX_SAFE_INTEGER,
38	} as const
39
40	const moveJobsSchema = {	36✔
41	body: {
42	type: 'object',
43	properties: {
44	fromQueue: {
45	type: 'string',
46	},
47	toQueue: {
48	type: 'string',
49	},
50	deleteJobsFromOriginalQueue: {
51	type: 'boolean',
52	default: false,
53	},
54	},
55	required: ['fromQueue', 'toQueue'],
56	},
57	} as const
58
59	const listQueueOverflowSchema = {	36✔
60	description: 'List created pgBoss jobs from the live queue table or overflow backup table.',
61	querystring: {
62	type: 'object',
63	properties: {
64	source: {
65	type: 'string',
66	enum: ['job', 'backup'],
67	default: 'job',
68	},
69	groupBy: {
70	type: 'string',
71	enum: ['summary', 'tenant'],
72	default: 'summary',
73	},
74	name: nonBlankStringSchema,
75	eventTypes: {
76	...nonBlankStringSchema,
77	description: 'Comma-separated event types to filter on.',
78	},
79	tenantRefs: {
80	...nonBlankStringSchema,
81	description: 'Comma-separated tenant refs to filter on.',
82	},
83	limit: {
84	...positiveSafeIntegerSchema,
85	default: JOB_OVERFLOW_LIST_LIMIT_DEFAULT,
86	},
87	},
88	additionalProperties: false,
89	},
90	} as const
91
92	const countQueueOverflowSchema = {	36✔
93	description: 'Count created pgBoss jobs in the live queue table.',
94	} as const
95
96	const backupQueueOverflowSchema = {	36✔
97	description: 'Move created pgBoss jobs into the overflow backup table.',
98	body: {
99	type: 'object',
100	properties: {
101	name: nonBlankStringSchema,
102	eventTypes: stringListSchema,
103	tenantRefs: stringListSchema,
104	limit: positiveSafeIntegerSchema,
105	confirmAll: {
106	type: 'boolean',
107	description: 'Required when no queue, event type, or tenant filter is supplied.',
108	},
109	},
110	anyOf: [
111	{ required: ['name'] },
112	{ required: ['eventTypes'] },
113	{ required: ['tenantRefs'] },
114	{
115	properties: {
116	confirmAll: { type: 'boolean', enum: [true] },
117	},
118	required: ['confirmAll'],
119	},
120	],
121	additionalProperties: false,
122	},
123	} as const
124
125	const restoreQueueOverflowSchema = {	36✔
126	description:
127	'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.',
128	body: {
129	type: 'object',
130	properties: {
131	name: nonBlankStringSchema,
132	eventTypes: stringListSchema,
133	tenantRefs: stringListSchema,
134	limit: {
135	...positiveSafeIntegerSchema,
136	default: JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT,
137	},
138	},
139	additionalProperties: false,
140	},
141	} as const
142
143	interface MoveJobsRequestInterface extends RequestGenericInterface {
144	Body: FromSchema<typeof moveJobsSchema.body>
145	}
146
147	interface ListQueueOverflowRequestInterface extends RequestGenericInterface {
148	Querystring: FromSchema<typeof listQueueOverflowSchema.querystring>
149	}
150
151	interface BackupQueueOverflowRequestInterface extends RequestGenericInterface {
152	Body: FromSchema<typeof backupQueueOverflowSchema.body>
153	}
154
155	interface RestoreQueueOverflowRequestInterface extends RequestGenericInterface {
156	Body: FromSchema<typeof restoreQueueOverflowSchema.body>
157	}
158
159	export default async function routes(fastify: FastifyInstance) {
160	registerApiKeyAuth(fastify)	19✔
161
162	fastify.post<MoveJobsRequestInterface>(	19✔
163	'/move',
164	{ schema: { ...moveJobsSchema, tags: ['queue'] } },
165	async (req, reply) => {
166	if (!pgQueueEnable) {	1!
167	return reply.status(400).send({ message: 'Queue is not enabled' })	×
168	}
169
170	const fromQueue = req.body.fromQueue	1✔
171	const toQueue = req.body.toQueue	1✔
172	const deleteJobsFromOriginalQueue = req.body.deleteJobsFromOriginalQueue \|\| false	1!
173
174	await MoveJobs.send({	1✔
175	fromQueue,
176	toQueue,
177	deleteJobsFromOriginalQueue,
178	sbReqId: req.sbReqId,
179	tenant: SYSTEM_TENANT,
180	})
181
182	return reply.send({ message: 'Move jobs scheduled' })	1✔
183	}
184	)
185
186	fastify.get<ListQueueOverflowRequestInterface>(	19✔
187	'/overflow',
188	{ schema: { ...listQueueOverflowSchema, tags: ['queue'] } },
189	async (req, reply) => {
190	if (!pgQueueEnable) {	3!
NEW 191	return reply.status(400).send({ message: 'Queue is not enabled' })	×
192	}
193
194	const store = getQueueOverflowStore()	3✔
195	const data = await store.list({	3✔
196	source: req.query.source,
197	groupBy: req.query.groupBy,
198	name: req.query.name,
199	eventTypes: parseCommaSeparatedList(req.query.eventTypes),
200	tenantRefs: parseCommaSeparatedList(req.query.tenantRefs),
201	limit: req.query.limit,
202	signal: req.signals.disconnect.signal,
203	})
204
205	return reply.send(data)	3✔
206	}
207	)
208
209	fastify.get(	19✔
210	'/overflow/count',
211	{ schema: { ...countQueueOverflowSchema, tags: ['queue'] } },
212	async (req, reply) => {
213	if (!pgQueueEnable) {	2!
NEW 214	return reply.status(400).send({ message: 'Queue is not enabled' })	×
215	}
216
217	const store = getQueueOverflowStore()	2✔
218	const data = await store.countCreated({ signal: req.signals.disconnect.signal })	2✔
219	return reply.send(data)	2✔
220	}
221	)
222
223	fastify.post<BackupQueueOverflowRequestInterface>(	19✔
224	'/overflow/backup',
225	{ schema: { ...backupQueueOverflowSchema, tags: ['queue'] } },
226	async (req, reply) => {
227	if (!pgQueueEnable) {	5!
NEW 228	return reply.status(400).send({ message: 'Queue is not enabled' })	×
229	}
230
231	const store = getQueueOverflowStore()	5✔
232	const data = await store.backup({	5✔
233	...req.body,
234	signal: req.signals.disconnect.signal,
235	})
236	return reply.send(data)	5✔
237	}
238	)
239
240	fastify.post<RestoreQueueOverflowRequestInterface>(	19✔
241	'/overflow/restore',
242	{ schema: { ...restoreQueueOverflowSchema, tags: ['queue'] } },
243	async (req, reply) => {
244	if (!pgQueueEnable) {	7!
NEW 245	return reply.status(400).send({ message: 'Queue is not enabled' })	×
246	}
247
248	if (databaseEngine === 'oriole') {	7✔
249	return reply	1✔
250	.status(400)
251	.send({ message: 'Queue overflow restore is not supported on OrioleDB' })
252	}
253
254	const store = getQueueOverflowStore()	6✔
255	const data = await store.restore({	6✔
256	...req.body,
257	signal: req.signals.disconnect.signal,
258	})
259	return reply.send(data)	5✔
260	}
261	)
262	}

supabase / storage / 29148700684

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous