From 0d8ee9f4acc3b1888affb6bfea1784384fb817ce Mon Sep 17 00:00:00 2001 From: Qoder-Undefined Date: Sat, 22 Aug 2026 17:46:36 +0100 Subject: [PATCH 1/2] logging interceptor --- app/backend/README.md | 37 +++++ app/backend/src/app.module.ts | 2 + .../interceptors/logging.interceptor.ts | 5 +- .../middleware/correlation-id.middleware.ts | 16 +- .../src/job-queue/job-executor.service.ts | 151 ++++++++++-------- .../job-executor.service.unit.spec.ts | 9 ++ .../src/job-queue/job-queue.service.ts | 8 + .../job-queue/job-queue.service.unit.spec.ts | 15 ++ app/backend/src/job-queue/job.repository.ts | 6 + app/backend/src/job-queue/types/job.types.ts | 7 + app/backend/src/main.ts | 4 +- .../src/metrics/metrics.interceptor.ts | 44 ++++- 12 files changed, 226 insertions(+), 78 deletions(-) diff --git a/app/backend/README.md b/app/backend/README.md index e3a51744c..18cd5c86f 100644 --- a/app/backend/README.md +++ b/app/backend/README.md @@ -114,6 +114,43 @@ try { --- +## Request Correlation & Distributed Tracing + +Every inbound HTTP request receives a unique **correlation ID** (`x-request-id`). This ID is automatically propagated across all layers: + +| Layer | How it works | +|---|---| +| **HTTP middleware** | `CorrelationIdMiddleware` extracts or generates a UUID and sets it on the request, response headers, and `AsyncLocalStorage` context. | +| **Logging** | `LoggingInterceptor` and `MetricsInterceptor` emit structured JSON logs containing `correlationId`. | +| **Metrics** | Prometheus histograms/counters include the correlation ID in log output for Grafana correlation. | +| **Job queue** | `JobQueueService.enqueue()` captures the caller's correlation ID and persists it in the `jobs` table. `JobExecutor` restores it into `AsyncLocalStorage` when the job runs. | +| **Error responses** | `GlobalHttpExceptionFilter` includes `request_id` / `correlationId` in every error envelope. | +| **External calls** | `@TraceExternalCall` decorator reads the correlation ID from `AsyncLocalStorage` for downstream call tracing. | + +### Tracing a request end-to-end + +1. **Client sends HTTP request** with `x-request-id: ` (optional — generated if absent). +2. **Middleware** stores the ID in `AsyncLocalStorage` and sets response headers. +3. **Interceptors** log structured JSON with the correlation ID on every request. +4. **If a background job is enqueued**, the correlation ID is stored in the `jobs.correlation_id` column. +5. **When the job executor picks up the job**, it restores the original correlation ID into `AsyncLocalStorage`, so all downstream service calls (DB, external APIs) carry the same ID. +6. **To trace a full request lifecycle**, grep logs for the correlation ID: + ```bash + # Find all log entries for a specific request + grep 'correlationId.*abc-123-...' logs/combined.log + ``` + +### Correlation ID headers + +| Header | Direction | Purpose | +|---|---|---| +| `x-request-id` | Request → Response | Canonical correlation identifier | +| `x-correlation-id` | Request → Response | Legacy alias (backward compatible) | + +Both headers are set on every response so clients can correlate responses to requests. + +--- + ## Architecture ```text diff --git a/app/backend/src/app.module.ts b/app/backend/src/app.module.ts index 36dea5359..79fc5499d 100644 --- a/app/backend/src/app.module.ts +++ b/app/backend/src/app.module.ts @@ -23,6 +23,7 @@ import { PaymentsModule } from "./payments/payments.module"; import { MetricsMiddleware } from "./metrics/metrics.middleware"; import { MetricsInterceptor } from "./metrics/metrics.interceptor"; import { CorrelationIdMiddleware } from "./common/middleware/correlation-id.middleware"; +import { CorrelationContextModule } from "./common/correlation/correlation-context.module"; import { OrganizationContextMiddleware } from "./common/middleware/organization-context.middleware"; import { ShadowTrafficMiddleware } from "./environment-parity/shadow-traffic.middleware"; import { IngestionModule } from "./ingestion/ingestion.module"; @@ -56,6 +57,7 @@ const validatedEnv = validateEnv(process.env); @Module({ imports: [ + CorrelationContextModule, SentryModule, AppConfigModule, // ScheduleModule registered once here — shared by NotificationsModule and ReconciliationModule diff --git a/app/backend/src/common/interceptors/logging.interceptor.ts b/app/backend/src/common/interceptors/logging.interceptor.ts index c0ee85236..bac9af806 100644 --- a/app/backend/src/common/interceptors/logging.interceptor.ts +++ b/app/backend/src/common/interceptors/logging.interceptor.ts @@ -8,6 +8,7 @@ import { } from '@nestjs/common'; import { Observable } from 'rxjs'; import { tap } from 'rxjs/operators'; +import { CorrelationContextService } from '../correlation/correlation-context.service'; /** * LoggingInterceptor logs every HTTP request with full request context @@ -24,6 +25,8 @@ import { tap } from 'rxjs/operators'; export class LoggingInterceptor implements NestInterceptor { private readonly logger = new Logger(LoggingInterceptor.name); + constructor(private readonly correlationContext?: CorrelationContextService) {} + /** Fields that must never appear in logs */ private static readonly SENSITIVE_FIELDS = new Set([ 'password', @@ -44,7 +47,7 @@ export class LoggingInterceptor implements NestInterceptor { intercept(context: ExecutionContext, next: CallHandler): Observable { const request = context.switchToHttp().getRequest(); const { method, url, body, route } = request; - const correlationId = request.correlationId || 'N/A'; + const correlationId = this.correlationContext?.getCorrelationId() || request.correlationId || 'N/A'; const userId = this.extractUserId(request); const routePath = route?.path || url; const now = Date.now(); diff --git a/app/backend/src/common/middleware/correlation-id.middleware.ts b/app/backend/src/common/middleware/correlation-id.middleware.ts index 4575af3cf..4440c7838 100644 --- a/app/backend/src/common/middleware/correlation-id.middleware.ts +++ b/app/backend/src/common/middleware/correlation-id.middleware.ts @@ -1,15 +1,29 @@ import { Injectable, NestMiddleware } from '@nestjs/common'; import { Request, Response, NextFunction } from 'express'; import { v4 as uuidv4 } from 'uuid'; +import { CorrelationContextService } from '../correlation/correlation-context.service'; +/** + * Middleware that extracts or generates a correlation ID for every inbound + * HTTP request and makes it available in three places: + * + * 1. Response headers (`x-request-id` / `x-correlation-id`) — echoed back to clients. + * 2. Express request object (`req.correlationId`) — used by filters and interceptors. + * 3. `AsyncLocalStorage` context — used by any service in the same async call stack + * (DB queries, logging, metrics, downstream calls). + */ @Injectable() export class CorrelationIdMiddleware implements NestMiddleware { + constructor(private readonly correlationContext: CorrelationContextService) {} + use(req: Request, res: Response, next: NextFunction) { const correlationId = req.header('x-request-id') || req.header('x-correlation-id') || uuidv4(); // Expose as both the legacy header and the canonical request-id header res.setHeader('x-request-id', correlationId); res.setHeader('x-correlation-id', correlationId); - req['correlationId'] = correlationId; + req['correlationId'] = correlationId; + // Propagate into AsyncLocalStorage so downstream services can read it + this.correlationContext.setCorrelationId(correlationId); next(); } } \ No newline at end of file diff --git a/app/backend/src/job-queue/job-executor.service.ts b/app/backend/src/job-queue/job-executor.service.ts index 186d3dd15..e084e2693 100644 --- a/app/backend/src/job-queue/job-executor.service.ts +++ b/app/backend/src/job-queue/job-executor.service.ts @@ -14,6 +14,7 @@ import { JobRegistry } from './job-registry.service'; import { CancellationStore } from './cancellation-token'; import { JobQueueMetricsService } from './job-queue-metrics.service'; import { Job, JobStatus, RetryPolicy } from './types'; +import { CorrelationContextService } from '../common/correlation/correlation-context.service'; /** * Job Executor Service @@ -39,6 +40,7 @@ export class JobExecutor implements OnModuleInit { private readonly registry: JobRegistry, private readonly cancellationStore: CancellationStore, private readonly metrics: JobQueueMetricsService, + private readonly correlationContext: CorrelationContextService, ) {} /** @@ -144,86 +146,96 @@ export class JobExecutor implements OnModuleInit { private async executeJob(job: Job): Promise { let policy: RetryPolicy; - try { - // Get the retry policy for this job type to determine visibility timeout - policy = this.registry.getPolicy(job.type); - - // Check if this job has an expired visibility timeout - // This indicates the previous execution attempt timed out or crashed - if (job.visibilityTimeout && job.visibilityTimeout < new Date()) { - this.logger.warn( - `Job ${job.id} has expired visibility timeout (type: ${job.type}, timeout: ${job.visibilityTimeout.toISOString()}). Treating as timeout failure.`, - ); - - // Treat expired visibility timeout as a failure - const timeoutError = new Error( - `Visibility timeout expired at ${job.visibilityTimeout.toISOString()}`, - ); - await this.handleJobFailure(job, timeoutError, policy); - return; - } + // Restore the caller's correlation ID into AsyncLocalStorage so that + // any logging, metrics, or downstream service calls within the handler + // carry the same correlation ID as the originating HTTP request. + // Falls back to the job ID for jobs without an origin context. + const jobCorrelationId = job.correlationId || job.id; - // Calculate visibility timeout: current time + policy.visibilityTimeoutMs - const visibilityTimeout = new Date(Date.now() + policy.visibilityTimeoutMs); - const startedAt = new Date(); + await this.correlationContext.run(jobCorrelationId, async () => { + try { + // Get the retry policy for this job type to determine visibility timeout + policy = this.registry.getPolicy(job.type); + + // Check if this job has an expired visibility timeout + // This indicates the previous execution attempt timed out or crashed + if (job.visibilityTimeout && job.visibilityTimeout < new Date()) { + this.logger.warn( + `Job ${job.id} has expired visibility timeout (type: ${job.type}, timeout: ${job.visibilityTimeout.toISOString()}). Treating as timeout failure.`, + ); + + // Treat expired visibility timeout as a failure + const timeoutError = new Error( + `Visibility timeout expired at ${job.visibilityTimeout.toISOString()}`, + ); + await this.handleJobFailure(job, timeoutError, policy); + return; + } - // Lock the job by updating status to 'running' and setting visibility timeout - // This prevents other executor instances from picking up the same job - await this.repository.updateJobStatus(job.id, JobStatus.RUNNING, { - startedAt, - visibilityTimeout, - }); + // Calculate visibility timeout: current time + policy.visibilityTimeoutMs + const visibilityTimeout = new Date(Date.now() + policy.visibilityTimeoutMs); + const startedAt = new Date(); - // Update gauge metrics: pending -> running - this.metrics.updateJobsPendingCount(job.type, -1); - this.metrics.updateJobsRunningCount(job.type, 1); + // Lock the job by updating status to 'running' and setting visibility timeout + // This prevents other executor instances from picking up the same job + await this.repository.updateJobStatus(job.id, JobStatus.RUNNING, { + startedAt, + visibilityTimeout, + }); - // Structured logging: job started at INFO level - this.logger.log({ - message: 'Job started', - jobId: job.id, - type: job.type, - attempts: job.attempts + 1, - }); + // Update gauge metrics: pending -> running + this.metrics.updateJobsPendingCount(job.type, -1); + this.metrics.updateJobsRunningCount(job.type, 1); + + // Structured logging: job started at INFO level + this.logger.log({ + message: 'Job started', + jobId: job.id, + type: job.type, + attempts: job.attempts + 1, + correlationId: jobCorrelationId, + }); - // Retrieve handler from JobRegistry - const handler = this.registry.getHandler(job.type); + // Retrieve handler from JobRegistry + const handler = this.registry.getHandler(job.type); - // Create CancellationToken for the job - const cancellationToken = this.cancellationStore.createToken(job.id); + // Create CancellationToken for the job + const cancellationToken = this.cancellationStore.createToken(job.id); - // Invoke handler.execute() with job and cancellation token - await handler.execute(job, cancellationToken); + // Invoke handler.execute() with job and cancellation token + await handler.execute(job, cancellationToken); - // Success: update status to completed, set completedAt - const completedAt = new Date(); - await this.repository.updateJobStatus(job.id, JobStatus.COMPLETED, { - completedAt, - }); + // Success: update status to completed, set completedAt + const completedAt = new Date(); + await this.repository.updateJobStatus(job.id, JobStatus.COMPLETED, { + completedAt, + }); - // Calculate execution duration in seconds - const durationMs = completedAt.getTime() - startedAt.getTime(); - const durationSeconds = durationMs / 1000; + // Calculate execution duration in seconds + const durationMs = completedAt.getTime() - startedAt.getTime(); + const durationSeconds = durationMs / 1000; - // Update metrics - this.metrics.incrementJobsCompleted(job.type); - this.metrics.updateJobsRunningCount(job.type, -1); - this.metrics.recordJobExecutionDuration(job.type, durationSeconds); - - // Structured logging: job completed at INFO level - this.logger.log({ - message: 'Job completed', - jobId: job.id, - type: job.type, - duration: durationMs, - }); + // Update metrics + this.metrics.incrementJobsCompleted(job.type); + this.metrics.updateJobsRunningCount(job.type, -1); + this.metrics.recordJobExecutionDuration(job.type, durationSeconds); - // Clean up cancellation token - this.cancellationStore.clearCancellation(job.id); - } catch (error) { - // Handle failure: increment attempts, set failureReason, calculate retry delay - await this.handleJobFailure(job, error, policy); - } + // Structured logging: job completed at INFO level + this.logger.log({ + message: 'Job completed', + jobId: job.id, + type: job.type, + correlationId: jobCorrelationId, + duration: durationMs, + }); + + // Clean up cancellation token + this.cancellationStore.clearCancellation(job.id); + } catch (error) { + // Handle failure: increment attempts, set failureReason, schedule retry or move to DLQ + await this.handleJobFailure(job, error, policy); + } + }); } /** @@ -252,6 +264,7 @@ export class JobExecutor implements OnModuleInit { message: 'Job failed', jobId: job.id, type: job.type, + correlationId: job.correlationId || job.id, attempts: newAttempts, failureReason, stack: error.stack, diff --git a/app/backend/src/job-queue/job-executor.service.unit.spec.ts b/app/backend/src/job-queue/job-executor.service.unit.spec.ts index da9c323a6..44340a445 100644 --- a/app/backend/src/job-queue/job-executor.service.unit.spec.ts +++ b/app/backend/src/job-queue/job-executor.service.unit.spec.ts @@ -16,6 +16,7 @@ import { JobRegistry } from './job-registry.service'; import { CancellationStore } from './cancellation-token'; import { JobQueueMetricsService } from './job-queue-metrics.service'; import { Job, JobType, JobStatus } from './types'; +import { CorrelationContextService } from '../common/correlation/correlation-context.service'; describe('JobExecutor', () => { let executor: JobExecutor; @@ -73,6 +74,13 @@ describe('JobExecutor', () => { mockRepository.updateJobStatus.mockResolvedValue(undefined); mockRepository.resetStaleJobs.mockResolvedValue(0); + const mockCorrelationContext = { + getCorrelationId: jest.fn().mockReturnValue('test-correlation-id'), + setCorrelationId: jest.fn(), + run: jest.fn((_id: string, fn: () => Promise) => fn()), + runSync: jest.fn((_id: string, fn: () => unknown) => fn()), + }; + const module: TestingModule = await Test.createTestingModule({ providers: [ JobExecutor, @@ -80,6 +88,7 @@ describe('JobExecutor', () => { { provide: JobRegistry, useValue: mockRegistry }, { provide: CancellationStore, useValue: mockCancellationStore }, { provide: JobQueueMetricsService, useValue: mockMetrics }, + { provide: CorrelationContextService, useValue: mockCorrelationContext }, ], }).compile(); diff --git a/app/backend/src/job-queue/job-queue.service.ts b/app/backend/src/job-queue/job-queue.service.ts index 4d7745c78..9d5114786 100644 --- a/app/backend/src/job-queue/job-queue.service.ts +++ b/app/backend/src/job-queue/job-queue.service.ts @@ -13,6 +13,7 @@ import { JobRegistry } from './job-registry.service'; import { CancellationStore } from './cancellation-token'; import { JobQueueMetricsService } from './job-queue-metrics.service'; import { Job, JobType, JobStatus } from './types'; +import { CorrelationContextService } from '../common/correlation/correlation-context.service'; /** * Error thrown when attempting to enqueue a job with an unregistered type @@ -56,6 +57,7 @@ export class JobQueueService { private readonly registry: JobRegistry, private readonly cancellationStore: CancellationStore, private readonly metrics: JobQueueMetricsService, + private readonly correlationContext: CorrelationContextService, ) {} /** @@ -124,6 +126,9 @@ export class JobQueueService { // Get retry policy for this job type const policy = this.registry.getPolicy(type); + // Capture the caller's correlation ID for distributed tracing + const correlationId = this.correlationContext.getCorrelationId(); + // Requirement 2.3: Persist job with status "pending" const job = await this.repository.createJob( type, @@ -131,6 +136,8 @@ export class JobQueueService { policy.maxAttempts, scheduledAt, idempotencyKey, + undefined, // retryMetadata + correlationId, ); // Increment jobs_enqueued_total metric @@ -145,6 +152,7 @@ export class JobQueueService { jobId: job.id, type, idempotencyKey, + correlationId, scheduledAt: scheduledAt.toISOString(), }); diff --git a/app/backend/src/job-queue/job-queue.service.unit.spec.ts b/app/backend/src/job-queue/job-queue.service.unit.spec.ts index ddba52663..a1a7a5e8d 100644 --- a/app/backend/src/job-queue/job-queue.service.unit.spec.ts +++ b/app/backend/src/job-queue/job-queue.service.unit.spec.ts @@ -13,6 +13,7 @@ import { JobRegistry } from './job-registry.service'; import { CancellationStore } from './cancellation-token'; import { JobQueueMetricsService } from './job-queue-metrics.service'; import { JobType, JobStatus, Job, JobHandler, RetryPolicy } from './types'; +import { CorrelationContextService } from '../common/correlation/correlation-context.service'; describe('JobQueueService', () => { let service: JobQueueService; @@ -77,6 +78,13 @@ describe('JobQueueService', () => { recordJobExecutionDuration: jest.fn(), }; + const mockCorrelationContext = { + getCorrelationId: jest.fn().mockReturnValue('test-correlation-id'), + setCorrelationId: jest.fn(), + run: jest.fn((_id, fn) => fn()), + runSync: jest.fn((_id, fn) => fn()), + }; + const module: TestingModule = await Test.createTestingModule({ providers: [ JobQueueService, @@ -84,6 +92,7 @@ describe('JobQueueService', () => { { provide: JobRegistry, useValue: mockRegistry }, { provide: CancellationStore, useValue: mockCancellationStore }, { provide: JobQueueMetricsService, useValue: mockMetrics }, + { provide: CorrelationContextService, useValue: mockCorrelationContext }, ], }).compile(); @@ -134,6 +143,9 @@ describe('JobQueueService', () => { payload, 5, // maxAttempts from policy expect.any(Date), // scheduledAt should be approximately now + undefined, // idempotencyKey + undefined, // retryMetadata + 'test-correlation-id', // correlationId from context ); }); @@ -247,6 +259,9 @@ describe('JobQueueService', () => { payload, 5, scheduledAt, + undefined, // idempotencyKey + undefined, // retryMetadata + 'test-correlation-id', // correlationId from context ); }); diff --git a/app/backend/src/job-queue/job.repository.ts b/app/backend/src/job-queue/job.repository.ts index d28ae3633..f66892d1f 100644 --- a/app/backend/src/job-queue/job.repository.ts +++ b/app/backend/src/job-queue/job.repository.ts @@ -30,6 +30,7 @@ interface JobRow { visibility_timeout: string | null; idempotency_key?: string | null; retry_metadata?: Record | null; + correlation_id?: string | null; } /** @@ -91,6 +92,7 @@ export class JobRepository { scheduledAt: Date = new Date(), idempotencyKey?: string, retryMetadata?: Record, + correlationId?: string, ): Promise> { const insertRow: Record = { type, @@ -107,6 +109,9 @@ export class JobRepository { if (retryMetadata) { insertRow.retry_metadata = retryMetadata; } + if (correlationId) { + insertRow.correlation_id = correlationId; + } const { data, error } = await this.client .from('jobs') @@ -375,6 +380,7 @@ export class JobRepository { visibilityTimeout: row.visibility_timeout ? new Date(row.visibility_timeout) : null, idempotencyKey: row.idempotency_key ?? null, retryMetadata: row.retry_metadata ?? null, + correlationId: row.correlation_id ?? null, }; } } diff --git a/app/backend/src/job-queue/types/job.types.ts b/app/backend/src/job-queue/types/job.types.ts index 50f7a54e4..535a6ffb7 100644 --- a/app/backend/src/job-queue/types/job.types.ts +++ b/app/backend/src/job-queue/types/job.types.ts @@ -74,6 +74,13 @@ export interface Job { /** Structured retry metadata for debugging and operator inspection */ retryMetadata?: Record | null; + + /** + * Correlation ID propagated from the caller context (HTTP request or + * upstream job). Enables operators to trace a single request across + * the entire NestJS application, queue, and realtime services. + */ + correlationId?: string | null; } /** diff --git a/app/backend/src/main.ts b/app/backend/src/main.ts index 8c85e359a..ea0b98fb0 100644 --- a/app/backend/src/main.ts +++ b/app/backend/src/main.ts @@ -26,6 +26,7 @@ import { mapValidationErrors } from "./common/utils/validation-error.mapper"; import { ErrorCode } from "./common/errors"; import { SentryExceptionFilter, SentryService } from "./sentry"; import { MetricsService } from "./metrics/metrics.service"; +import { CorrelationContextService } from "./common/correlation/correlation-context.service"; import { sanitizeErrorMessage, createConfigSummary, @@ -139,7 +140,8 @@ async function bootstrap() { }), ); - app.useGlobalInterceptors(new LoggingInterceptor()); + const correlationContext = app.get(CorrelationContextService); + app.useGlobalInterceptors(new LoggingInterceptor(correlationContext)); // Register Sentry exception filter FIRST so it captures errors, // then the existing HTTP exception filter handles the response. diff --git a/app/backend/src/metrics/metrics.interceptor.ts b/app/backend/src/metrics/metrics.interceptor.ts index 621c7fd03..5d78e63da 100644 --- a/app/backend/src/metrics/metrics.interceptor.ts +++ b/app/backend/src/metrics/metrics.interceptor.ts @@ -3,41 +3,73 @@ import { NestInterceptor, ExecutionContext, CallHandler, + Logger, } from '@nestjs/common'; import { Observable } from 'rxjs'; import { tap } from 'rxjs/operators'; import { MetricsService } from './metrics.service'; import { Request } from 'express'; + import { CorrelationContextService } from '../common/correlation/correlation-context.service'; @Injectable() export class MetricsInterceptor implements NestInterceptor { - constructor(private metricsService: MetricsService) {} + private readonly logger = new Logger(MetricsInterceptor.name); + + constructor( + private metricsService: MetricsService, + private readonly correlationContext: CorrelationContextService, + ) {} intercept(context: ExecutionContext, next: CallHandler): Observable { const start = Date.now(); const req = context.switchToHttp().getRequest(); const method = req.method; const route = req.route?.path || req.path; + const correlationId = this.correlationContext.getCorrelationId() || req['correlationId'] || 'N/A'; return next.handle().pipe( tap({ next: () => { const res = context.switchToHttp().getResponse(); - const duration = (Date.now() - start) / 1000; + const durationMs = Date.now() - start; + const durationSec = durationMs / 1000; this.metricsService.recordRequestDuration( method, route, res.statusCode, - duration, + durationSec, + ); + this.logger.log( + JSON.stringify({ + correlationId, + event: 'http_request_completed', + method, + route, + status_code: res.statusCode, + duration_ms: durationMs, + }), ); }, error: (err) => { - const duration = (Date.now() - start) / 1000; + const durationMs = Date.now() - start; + const durationSec = durationMs / 1000; + const statusCode = err.status || 500; this.metricsService.recordRequestDuration( method, route, - err.status || 500, - duration, + statusCode, + durationSec, + ); + this.logger.warn( + JSON.stringify({ + correlationId, + event: 'http_request_failed', + method, + route, + status_code: statusCode, + duration_ms: durationMs, + error: err.message, + }), ); }, }), From 30aa1cc7afed07560654fbe0ef1c8a82e09fc625 Mon Sep 17 00:00:00 2001 From: Qoder-Undefined Date: Sat, 22 Aug 2026 17:47:18 +0100 Subject: [PATCH 2/2] external call decorator --- .../correlation/correlation-context.module.ts | 21 +++++ .../correlation-context.service.ts | 87 +++++++++++++++++++ .../trace-external-call.decorator.ts | 9 +- 3 files changed, 116 insertions(+), 1 deletion(-) create mode 100644 app/backend/src/common/correlation/correlation-context.module.ts create mode 100644 app/backend/src/common/correlation/correlation-context.service.ts diff --git a/app/backend/src/common/correlation/correlation-context.module.ts b/app/backend/src/common/correlation/correlation-context.module.ts new file mode 100644 index 000000000..acf7d010c --- /dev/null +++ b/app/backend/src/common/correlation/correlation-context.module.ts @@ -0,0 +1,21 @@ +/** + * Correlation Context Module + * + * Provides the `CorrelationContextService` as a singleton across the + * application. Import this module in any feature module that needs to + * read or write the current correlation identifier. + * + * The service is backed by `AsyncLocalStorage`, so it works across + * HTTP middleware, background job execution, database calls, and any + * other async flow without explicit parameter threading. + */ + +import { Module, Global } from '@nestjs/common'; +import { CorrelationContextService } from './correlation-context.service'; + +@Global() +@Module({ + providers: [CorrelationContextService], + exports: [CorrelationContextService], +}) +export class CorrelationContextModule {} diff --git a/app/backend/src/common/correlation/correlation-context.service.ts b/app/backend/src/common/correlation/correlation-context.service.ts new file mode 100644 index 000000000..18f3691e2 --- /dev/null +++ b/app/backend/src/common/correlation/correlation-context.service.ts @@ -0,0 +1,87 @@ +/** + * Correlation Context Service + * + * Provides a single source of truth for the correlation identifier across + * all execution contexts: HTTP requests, background job processing, database + * operations, and any downstream service calls. + * + * Uses Node.js `AsyncLocalStorage` to propagate the correlation ID through + * the async call stack without explicit parameter threading. This enables + * operators to follow a single request or background action across the + * entire NestJS application, queue, and realtime services. + * + * Usage: + * // Set context (typically done in middleware or job executor) + * correlationContext.setCorrelationId('abc-123'); + * + * // Get current context (works anywhere in the same async call stack) + * const id = correlationContext.getCorrelationId(); + * + * // Run a callback with a specific correlation ID + * await correlationContext.run('abc-123', async () => { + * await someService.doWork(); + * }); + */ + +import { Injectable } from '@nestjs/common'; +import { AsyncLocalStorage } from 'async_hooks'; + +/** + * Shape of the correlation context stored per async execution. + */ +interface CorrelationStore { + correlationId: string; +} + +@Injectable() +export class CorrelationContextService { + private readonly storage = new AsyncLocalStorage(); + + /** + * Get the current correlation ID from the async context. + * Returns `undefined` when called outside any tracked context + * (e.g. during module initialization or in non-request code). + */ + getCorrelationId(): string | undefined { + return this.storage.getStore()?.correlationId; + } + + /** + * Set the correlation ID for the current async context. + * Typically called from the correlation-id middleware (HTTP) or + * the job executor (background jobs). + */ + setCorrelationId(correlationId: string): void { + const store = this.storage.getStore(); + if (store) { + store.correlationId = correlationId; + } else { + // Outside any tracked context — create a new root context. + this.storage.enterWith({ correlationId }); + } + } + + /** + * Run a callback inside a new async context with the given correlation ID. + * The callback can be async — all `await`ed work inherits the context. + * + * @param correlationId - The correlation ID to propagate + * @param fn - The callback to execute in context + * @returns The return value of `fn` + */ + async run(correlationId: string, fn: () => Promise): Promise { + return this.storage.run({ correlationId }, fn); + } + + /** + * Run a synchronous callback inside a new async context with the given + * correlation ID. Useful for non-async initialization paths. + * + * @param correlationId - The correlation ID to propagate + * @param fn - The synchronous callback to execute in context + * @returns The return value of `fn` + */ + runSync(correlationId: string, fn: () => T): T { + return this.storage.run({ correlationId }, fn); + } +} diff --git a/app/backend/src/common/decorators/trace-external-call.decorator.ts b/app/backend/src/common/decorators/trace-external-call.decorator.ts index c99fa896e..7cb468feb 100644 --- a/app/backend/src/common/decorators/trace-external-call.decorator.ts +++ b/app/backend/src/common/decorators/trace-external-call.decorator.ts @@ -1,5 +1,6 @@ import { Logger } from '@nestjs/common'; import { MetricsService } from '../../metrics/metrics.service'; +import { CorrelationContextService } from '../correlation/correlation-context.service'; /** * Decorator to trace external API calls with timing and error tracking. @@ -24,7 +25,13 @@ export function TraceExternalCall(service: string, operation: string) { descriptor.value = async function (...args: unknown[]) { const metricsService: MetricsService = (this as Record).metricsService as MetricsService; const startTime = Date.now(); - const correlationId = (this as Record).correlationId as string || 'N/A'; + // Resolve correlation ID: try AsyncLocalStorage first, then instance property, then fallback + const correlationContext: CorrelationContextService | undefined = + (this as Record).correlationContext as CorrelationContextService | undefined; + const correlationId = + correlationContext?.getCorrelationId() + || (this as Record).correlationId as string + || 'N/A'; logger.debug( JSON.stringify({