30089661594

Committed 24 Jul 2026 11:28AM UTC coverage: 80.756% (+0.4%) from 80.352%

Build # 30089661594

Build Type

Pull #1043

github

Committed by

web-flow

Commit Message

Merge 583fb7624 into e7b975a46

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

Coverage Stats

5829 of 7763 branches covered (75.09%)

Branch coverage included in aggregate %.

270 of 296 new or added lines in 6 files covered. (91.22%)

11066 of 13158 relevant lines covered (84.1%)

409.56 hits per line

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

92.5

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

import { SYSTEM_TENANT } from '@internal/queue/constants'
import {
  JOB_OVERFLOW_LIST_LIMIT_DEFAULT,
  JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT,
  parseCommaSeparatedList,
  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'
import { getConfig } from '../../../config'
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',
  finite: true,
  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.addHook('preHandler', async (_req, reply) => {
    if (!pgQueueEnable) {
      return reply.status(400).send({ message: 'Queue is not enabled' })
    }
  })

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

supabase / storage / 30089661594

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