feat(backend): add idempotency and durable retry-state handling to queue layer

app/backend/src/job-queue/job-queue.service.ts

@@ -75,36 +75,40 @@ export class JobQueueService {
   async enqueue<TPayload = unknown>(
     type: JobType,
     payload: TPayload,
+    idempotencyKey?: string,
   ): Promise<string> {
-    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<TPayload = unknown>(
     type: JobType,
     payload: TPayload,
     scheduledAt: Date,
+    idempotencyKey?: string,
   ): Promise<string> {
     // Requirement 1.5: Reject enqueue for unregistered job types
     if (!this.registry.isRegistered(type)) {
       this.logger.error(`Attempted to enqueue unregistered job type: ${type}`);
       throw new UnregisteredJobTypeError(type);
     }
 
+    // Idempotency check: if idempotencyKey provided, check if existing non-cancelled job exists
+    if (idempotencyKey) {
+      const existing = await this.repository.findByIdempotencyKey<TPayload>(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(),
     });
 

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<string, unknown> | null;
 }
 
 /**
@@ -87,17 +89,28 @@ export class JobRepository {
     payload: TPayload,
     maxAttempts: number,
     scheduledAt: Date = new Date(),
+    idempotencyKey?: string,
+    retryMetadata?: Record<string, unknown>,
   ): Promise<Job<TPayload>> {
+    const insertRow: Record<string, unknown> = {
+      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<TPayload = unknown>(
+    idempotencyKey: string,
+  ): Promise<Job<TPayload> | 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<TPayload>(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,
     };
   }
 }

app/backend/src/job-queue/types/job.types.ts

@@ -68,6 +68,12 @@ export interface Job<TPayload = unknown> {
 
   /** 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<string, unknown> | null;
 }
 
 /**

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<string, unknown> | null;
+}
+
+describe('Job Queue Integration & Idempotency', () => {
+  let service: JobQueueService;
+  let repository: JobRepository;
+  let registry: JobRegistry;
+
+  const mockJobsStore = new Map<string, MockJobRow>();
+  const mockIdempotencyStore = new Map<string, MockJobRow>();
+
+  beforeEach(async () => {
+    mockJobsStore.clear();
+    mockIdempotencyStore.clear();
+
+    const mockSupabase = {
+      getClient: () => ({
+        from: () => ({
+          insert: (row: Record<string, unknown>) => ({
+            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<string, unknown>) || 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>(JobQueueService);
+    repository = module.get<JobRepository>(JobRepository);
+    registry = module.get<JobRegistry>(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);
+  });
+});