updates to dashboard and proxy for DLQ redrive and such

Author: timgit

packages/dashboard/app/routes/queues.$name.jobs.$jobId.tsx

@@ -273,6 +273,33 @@ export default function JobDetail ({ loaderData }: Route.ComponentProps) {
                 </div>
               )}
 
+              {(job.sourceName || job.sourceId) && (
+                <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-x-6 gap-y-4">
+                  {job.sourceName && (
+                    <div>
+                      <dt className="pgb-eyebrow">Source Queue</dt>
+                      <dd className="mt-1 text-sm text-[var(--text-primary)]">
+                        <DbLink
+                          to={`/queues/${job.sourceName}`}
+                          className="font-mono text-xs text-primary-600 hover:text-primary-700 dark:text-primary-400 dark:hover:text-primary-300"
+                        >
+                          {job.sourceName}
+                        </DbLink>
+                      </dd>
+                    </div>
+                  )}
+                  {job.sourceId && (
+                    <ConfigItem label="Source Job ID" value={job.sourceId} mono />
+                  )}
+                  {job.sourceRetryCount !== null && job.sourceRetryCount !== undefined && (
+                    <ConfigItem label="Source Retry Count" value={job.sourceRetryCount} mono />
+                  )}
+                  {job.sourceCreatedOn && (
+                    <ConfigItem label="Source Created" value={formatDate(new Date(job.sourceCreatedOn))} mono />
+                  )}
+                </div>
+              )}
+
               <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-5 gap-x-6 gap-y-4">
                 <ConfigItem
                   label="Created"

packages/dashboard/tests/server/queries.test.ts

@@ -757,6 +757,39 @@ describe('Job Queries', () => {
       expect(job!.priority).toBe(5)
       expect(job!.data).toEqual({ foo: 'bar' })
     })
+
+    it('returns null source-tracking fields for a job that was not dead-lettered', async () => {
+      await createTestQueue('test-queue')
+      const jobId = await sendTestJob('test-queue', { foo: 'bar' })
+
+      const job = await getJobById(ctx.connectionString, ctx.schema, 'test-queue', jobId!)
+
+      expect(job!.sourceName).toBeNull()
+      expect(job!.sourceId).toBeNull()
+      expect(job!.sourceCreatedOn).toBeNull()
+      expect(job!.sourceRetryCount).toBeNull()
+    })
+
+    it('exposes source-tracking fields for a dead-lettered job', async () => {
+      await createTestQueue('test-dlq')
+      await createTestQueue('test-queue', { deadLetter: 'test-dlq', retryLimit: 0 })
+
+      const jobId = await sendTestJob('test-queue', { foo: 'bar' })
+
+      // fail() routes the job to the dead letter queue synchronously
+      await fetchTestJob('test-queue')
+      await failTestJob('test-queue', jobId!)
+
+      // The dead-lettered copy is a new job in the DLQ with its own id
+      const moved = await fetchTestJob('test-dlq')
+      const dlqJob = await getJobById(ctx.connectionString, ctx.schema, 'test-dlq', moved!.id)
+
+      expect(dlqJob).not.toBeNull()
+      expect(dlqJob!.sourceName).toBe('test-queue')
+      expect(dlqJob!.sourceId).toBe(jobId)
+      expect(dlqJob!.sourceCreatedOn).toBeTruthy()
+      expect(dlqJob!.sourceRetryCount).toBe(0)
+    })
   })
 })
 

packages/proxy/src/contracts.ts

@@ -93,6 +93,12 @@ export const findJobsOptionsSchema = z.object({
   queued: z.boolean().optional(),
 }) satisfies z.ZodType<types.HttpFindJobsOptions>
 
+export const redriveOptionsSchema = z.object({
+  destination: queueNameSchema.optional(),
+  sourceName: queueNameSchema.optional(),
+  limit: z.number().optional(),
+}) satisfies z.ZodType<types.HttpRedriveOptions>
+
 export const insertOptionsSchema = z.object({
   returnId: z.boolean().optional(),
 }) satisfies z.ZodType<types.HttpInsertOptions>
@@ -162,6 +168,10 @@ export const jobWithMetadataSchema = jobSchemaBase.extend({
   policy: z.string(),
   deadLetter: z.string(),
   output: jsonRecordSchema,
+  sourceName: z.string().nullable(),
+  sourceId: z.string().nullable(),
+  sourceCreatedOn: z.iso.datetime().nullable().transform((val) => val ? new Date(val) : null),
+  sourceRetryCount: z.number().nullable(),
 }) satisfies z.ZodType<types.HttpJobWithMetadata>
 
 export const commandResponseSchema = z.object({
@@ -380,6 +390,16 @@ export const deleteJobResponseSchema: z.ZodType<types.HttpDeleteJobResponse> = z
   result: commandResponseSchema
 })
 
+export const redriveRequestSchema: z.ZodType<types.HttpRedriveRequest> = z.object({
+  name: queueNameSchema,
+  options: redriveOptionsSchema.optional()
+})
+
+export const redriveResponseSchema: z.ZodType<types.HttpRedriveResponse> = z.object({
+  ok: z.literal(true),
+  result: z.number()
+})
+
 export const deleteQueuedJobsRequestSchema: z.ZodType<types.HttpDeleteQueuedJobsRequest> = z.object({
   name: queueNameSchema
 })

packages/proxy/src/routes.ts

@@ -35,6 +35,8 @@ import {
   isInstalledResponseSchema,
   publishRequestSchema,
   publishResponseSchema,
+  redriveRequestSchema,
+  redriveResponseSchema,
   resumeRequestSchema,
   resumeResponseSchema,
   retryRequestSchema,
@@ -177,6 +179,7 @@ export const postMethods: RouteEntry[] = [
   post('jobs', 'resume', resumeRequestSchema, resumeResponseSchema, (body) => [body.name, body.id]),
   post('jobs', 'retry', retryRequestSchema, retryResponseSchema, (body) => [body.name, body.id]),
   post('jobs', 'deleteJob', deleteJobRequestSchema, deleteJobResponseSchema, (body) => [body.name, body.id]),
+  post('jobs', 'redrive', redriveRequestSchema, redriveResponseSchema, (body) => withOptionalOptions([body.name], body.options)),
   post('jobs', 'deleteQueuedJobs', deleteQueuedJobsRequestSchema, deleteQueuedJobsResponseSchema, (body) => [body.name]),
   post('jobs', 'deleteStoredJobs', deleteStoredJobsRequestSchema, deleteStoredJobsResponseSchema, (body) => [body.name]),
   post('jobs', 'deleteAllJobs', deleteAllJobsRequestSchema, deleteAllJobsResponseSchema, (body) => (body.name ? [body.name] : [])),

packages/proxy/src/types.ts

@@ -39,6 +39,8 @@ export type HttpFindJobsOptions = Omit<types.FindJobsOptions, 'db' | 'data'> & {
   data?: HttpJsonRecord
 }
 
+export type HttpRedriveOptions = Omit<types.RedriveOptions, 'db'>
+
 export type HttpInsertOptions = Omit<types.InsertOptions, 'db'>
 
 export type HttpCompleteOptions = Omit<types.CompleteOptions, 'db'>
@@ -198,6 +200,16 @@ export type HttpDeleteJobRequest = HttpCancelRequest
 
 export type HttpDeleteJobResponse = HttpCancelResponse
 
+export type HttpRedriveRequest = {
+  name: HttpQueueName
+  options?: HttpRedriveOptions
+}
+
+export type HttpRedriveResponse = {
+  ok: true
+  result: number
+}
+
 export type HttpDeleteQueuedJobsRequest = {
   name: HttpQueueName
 }

packages/proxy/test/apiCallsTest.ts

@@ -440,6 +440,11 @@ describe('proxy api routes', () => {
         body: { name: 'queue', id: '1' },
         expected: ['queue', '1']
       },
+      {
+        method: 'redrive',
+        body: { name: 'dlq', options: { destination: 'dest', sourceName: 'src', limit: 50 } },
+        expected: ['dlq', { destination: 'dest', sourceName: 'src', limit: 50 }]
+      },
       {
         method: 'deleteQueuedJobs',
         body: { name: 'queue' },

packages/proxy/test/typeDriftTest.ts

@@ -52,6 +52,10 @@ describe('Zod schema / HTTP type key drift', () => {
     expect(assertKeysMatch<AssertKeysMatch<SchemaOutput<typeof contracts.findJobsOptionsSchema>, httpTypes.HttpFindJobsOptions>>()).toBe(true)
   })
 
+  it('redriveOptionsSchema matches HttpRedriveOptions', () => {
+    expect(assertKeysMatch<AssertKeysMatch<SchemaOutput<typeof contracts.redriveOptionsSchema>, httpTypes.HttpRedriveOptions>>()).toBe(true)
+  })
+
   it('insertOptionsSchema matches HttpInsertOptions', () => {
     expect(assertKeysMatch<AssertKeysMatch<SchemaOutput<typeof contracts.insertOptionsSchema>, httpTypes.HttpInsertOptions>>()).toBe(true)
   })

src/index.ts

@@ -496,6 +496,7 @@ export type {
   QueueOptions,
   QueuePolicy,
   QueueResult,
+  RedriveOptions,
   Request,
   Schedule,
   ScheduleOptions,