diff --git a/app/backend/src/job-queue/job-queue.service.ts b/app/backend/src/job-queue/job-queue.service.ts index 7f3b3d340..4d7745c78 100644 --- a/app/backend/src/job-queue/job-queue.service.ts +++ b/app/backend/src/job-queue/job-queue.service.ts @@ -75,29 +75,19 @@ export class JobQueueService { async enqueue( type: JobType, payload: TPayload, + idempotencyKey?: string, ): Promise { - return this.enqueueDelayed(type, payload, new Date()); + return this.enqueueDelayed(type, payload, new Date(), idempotencyKey); } /** - * Enqueue a job for delayed execution - * - * Creates a new job with a future scheduledAt timestamp. - * The job will be picked up by the JobExecutor when scheduledAt is reached. - * - * @param type - Job type (must be registered) - * @param payload - Job-specific payload data - * @param scheduledAt - When the job should execute - * @returns The created job ID - * @throws UnregisteredJobTypeError if job type is not registered - * @throws PayloadValidationError if payload validation fails - * - * **Validates: Requirements 2.2, 2.3, 2.4, 2.5, 2.6, 1.5, 15.2, 15.3** + * Enqueue a job for delayed execution with optional idempotency key */ async enqueueDelayed( type: JobType, payload: TPayload, scheduledAt: Date, + idempotencyKey?: string, ): Promise { // Requirement 1.5: Reject enqueue for unregistered job types if (!this.registry.isRegistered(type)) { @@ -105,6 +95,20 @@ export class JobQueueService { throw new UnregisteredJobTypeError(type); } + // Idempotency check: if idempotencyKey provided, check if existing non-cancelled job exists + if (idempotencyKey) { + const existing = await this.repository.findByIdempotencyKey(idempotencyKey); + if (existing && existing.status !== JobStatus.CANCELLED) { + this.logger.log({ + message: 'Duplicate job submission suppressed by idempotency key', + jobId: existing.id, + type, + idempotencyKey, + }); + return existing.id; + } + } + // Requirement 15.2, 15.3: Validate payload against job type schema try { const handler = this.registry.getHandler(type); @@ -126,6 +130,7 @@ export class JobQueueService { payload, policy.maxAttempts, scheduledAt, + idempotencyKey, ); // Increment jobs_enqueued_total metric @@ -139,6 +144,7 @@ export class JobQueueService { message: 'Job enqueued', jobId: job.id, type, + idempotencyKey, scheduledAt: scheduledAt.toISOString(), }); diff --git a/app/backend/src/job-queue/job.repository.ts b/app/backend/src/job-queue/job.repository.ts index 84536194d..d28ae3633 100644 --- a/app/backend/src/job-queue/job.repository.ts +++ b/app/backend/src/job-queue/job.repository.ts @@ -28,6 +28,8 @@ interface JobRow { completed_at: string | null; failure_reason: string | null; visibility_timeout: string | null; + idempotency_key?: string | null; + retry_metadata?: Record | null; } /** @@ -87,17 +89,28 @@ export class JobRepository { payload: TPayload, maxAttempts: number, scheduledAt: Date = new Date(), + idempotencyKey?: string, + retryMetadata?: Record, ): Promise> { + const insertRow: Record = { + type, + payload: payload as unknown, + status: JobStatus.PENDING, + attempts: 0, + max_attempts: maxAttempts, + scheduled_at: scheduledAt.toISOString(), + }; + + if (idempotencyKey) { + insertRow.idempotency_key = idempotencyKey; + } + if (retryMetadata) { + insertRow.retry_metadata = retryMetadata; + } + const { data, error } = await this.client .from('jobs') - .insert({ - type, - payload: payload as unknown, - status: JobStatus.PENDING, - attempts: 0, - max_attempts: maxAttempts, - scheduled_at: scheduledAt.toISOString(), - }) + .insert(insertRow) .select() .single(); @@ -313,6 +326,33 @@ export class JobRepository { return count; } + /** + * Find a job by its idempotency key + * + * @param idempotencyKey - Idempotency key + * @returns The job, or null if not found + */ + async findByIdempotencyKey( + idempotencyKey: string, + ): Promise | null> { + const { data, error } = await this.client + .from('jobs') + .select('*') + .eq('idempotency_key', idempotencyKey) + .maybeSingle(); + + if (error?.code === 'PGRST116' || !data) { + return null; + } + + if (error) { + this.logger.error(`Failed to find job by idempotency key ${idempotencyKey}: ${error.message}`, error); + throw error; + } + + return this.mapRowToJob(data as JobRow); + } + /** * Map a database row to a Job object * @@ -333,6 +373,8 @@ export class JobRepository { completedAt: row.completed_at ? new Date(row.completed_at) : null, failureReason: row.failure_reason, visibilityTimeout: row.visibility_timeout ? new Date(row.visibility_timeout) : null, + idempotencyKey: row.idempotency_key ?? null, + retryMetadata: row.retry_metadata ?? null, }; } } diff --git a/app/backend/src/job-queue/types/job.types.ts b/app/backend/src/job-queue/types/job.types.ts index 3a7342fba..50f7a54e4 100644 --- a/app/backend/src/job-queue/types/job.types.ts +++ b/app/backend/src/job-queue/types/job.types.ts @@ -68,6 +68,12 @@ export interface Job { /** Lock expiry timestamp - prevents concurrent execution */ visibilityTimeout: Date | null; + + /** Optional idempotency key to prevent duplicate execution */ + idempotencyKey?: string | null; + + /** Structured retry metadata for debugging and operator inspection */ + retryMetadata?: Record | null; } /** diff --git a/app/backend/test/queue.integration.spec.ts b/app/backend/test/queue.integration.spec.ts new file mode 100644 index 000000000..e6f15820f --- /dev/null +++ b/app/backend/test/queue.integration.spec.ts @@ -0,0 +1,175 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { JobQueueService } from '../src/job-queue/job-queue.service'; +import { JobRepository } from '../src/job-queue/job.repository'; +import { JobRegistry } from '../src/job-queue/job-registry.service'; +import { CancellationStore } from '../src/job-queue/cancellation-token'; +import { JobQueueMetricsService } from '../src/job-queue/job-queue-metrics.service'; +import { SupabaseService } from '../src/supabase/supabase.service'; +import { JobType } from '../src/job-queue/types/job.types'; + +interface MockJobRow { + id: string; + type: string; + payload: unknown; + status: string; + attempts: number; + max_attempts: number; + created_at: string; + scheduled_at: string; + started_at: string | null; + completed_at: string | null; + failure_reason: string | null; + visibility_timeout: string | null; + idempotency_key: string | null; + retry_metadata: Record | null; +} + +describe('Job Queue Integration & Idempotency', () => { + let service: JobQueueService; + let repository: JobRepository; + let registry: JobRegistry; + + const mockJobsStore = new Map(); + const mockIdempotencyStore = new Map(); + + beforeEach(async () => { + mockJobsStore.clear(); + mockIdempotencyStore.clear(); + + const mockSupabase = { + getClient: () => ({ + from: () => ({ + insert: (row: Record) => ({ + select: () => ({ + single: async () => { + const id = `job-${Date.now()}-${Math.random()}`; + const jobRow: MockJobRow = { + id, + type: String(row.type), + payload: row.payload, + status: String(row.status), + attempts: Number(row.attempts), + max_attempts: Number(row.max_attempts), + created_at: new Date().toISOString(), + scheduled_at: String(row.scheduled_at), + started_at: null, + completed_at: null, + failure_reason: null, + visibility_timeout: null, + idempotency_key: (row.idempotency_key as string) || null, + retry_metadata: (row.retry_metadata as Record) || null, + }; + mockJobsStore.set(id, jobRow); + if (row.idempotency_key) { + mockIdempotencyStore.set(row.idempotency_key as string, jobRow); + } + return { data: jobRow, error: null }; + }, + }), + }), + select: () => ({ + eq: (col: string, val: unknown) => ({ + maybeSingle: async () => { + if (col === 'id') { + const job = mockJobsStore.get(String(val)); + return { data: job || null, error: job ? null : { code: 'PGRST116' } }; + } + if (col === 'idempotency_key') { + const job = mockIdempotencyStore.get(String(val)); + return { data: job || null, error: job ? null : { code: 'PGRST116' } }; + } + return { data: null, error: null }; + }, + }), + }), + }), + }), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + JobQueueService, + JobRepository, + JobRegistry, + CancellationStore, + { + provide: JobQueueMetricsService, + useValue: { + incrementJobsEnqueued: jest.fn(), + updateJobsPendingCount: jest.fn(), + incrementJobsCancelled: jest.fn(), + }, + }, + { provide: SupabaseService, useValue: mockSupabase }, + ], + }).compile(); + + service = module.get(JobQueueService); + repository = module.get(JobRepository); + registry = module.get(JobRegistry); + + // Register a test job handler + registry.registerHandler({ + type: JobType.EXPORT_GENERATION, + handler: { + execute: jest.fn().mockResolvedValue(undefined), + validate: jest.fn().mockResolvedValue(undefined), + onFailure: jest.fn().mockResolvedValue(undefined), + }, + policy: { + maxAttempts: 3, + backoffStrategy: 'exponential', + initialDelayMs: 1000, + maxDelayMs: 30000, + visibilityTimeoutMs: 60000, + }, + }); + }); + + it('should enqueue job successfully and return unique ID', async () => { + const jobId = await service.enqueue(JobType.EXPORT_GENERATION, { exportId: 'exp-123' }); + expect(jobId).toBeDefined(); + expect(typeof jobId).toBe('string'); + }); + + it('should suppress duplicate job submissions when matching idempotencyKey is supplied', async () => { + const idempotencyKey = 'export-key-999'; + + const jobId1 = await service.enqueue( + JobType.EXPORT_GENERATION, + { exportId: 'exp-123' }, + idempotencyKey, + ); + + const jobId2 = await service.enqueue( + JobType.EXPORT_GENERATION, + { exportId: 'exp-123' }, + idempotencyKey, + ); + + expect(jobId1).toEqual(jobId2); + expect(mockJobsStore.size).toBe(1); + }); + + it('should store and retrieve structured retry metadata on job record', async () => { + const retryMetadata = { + attemptCount: 2, + lastError: 'Network timeout connecting to export storage provider', + lastFailureAt: new Date().toISOString(), + nextBackoffDelayMs: 5000, + inDlq: false, + }; + + const job = await repository.createJob( + JobType.EXPORT_GENERATION, + { exportId: 'exp-456' }, + 3, + new Date(), + 'idem-key-777', + retryMetadata, + ); + + expect(job.idempotencyKey).toBe('idem-key-777'); + expect(job.retryMetadata).toEqual(retryMetadata); + }); +});