Request Correlation ID Middleware for Distributed Tracing

middleware/src/monitoring/correlation-exception-filter.ts

@@ -0,0 +1,33 @@
+import {
+  ExceptionFilter,
+  Catch,
+  ArgumentsHost,
+  HttpException,
+  HttpStatus,
+} from '@nestjs/common';
+import { Response } from 'express';
+import { CorrelationIdStorage } from './correlation-id.storage';
+
+@Catch()
+export class CorrelationExceptionFilter implements ExceptionFilter {
+  catch(exception: any, host: ArgumentsHost) {
+    const ctx = host.switchToHttp();
+    const response = ctx.getResponse<Response>();
+    const status =
+      exception instanceof HttpException
+        ? exception.getStatus()
+        : HttpStatus.INTERNAL_SERVER_ERROR;
+
+    const correlationId = CorrelationIdStorage.getCorrelationId();
+    const userId = CorrelationIdStorage.getUserId();
+
+    response.status(status).json({
+      statusCode: status,
+      timestamp: new Date().toISOString(),
+      correlationId: correlationId || 'N/A',
+      userId: userId || 'N/A',
+      message: exception?.message || 'Internal server error',
+      path: ctx.getRequest().url,
+    });
+  }
+}

middleware/src/monitoring/correlation-http-interceptor.service.ts

@@ -0,0 +1,15 @@
+import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
+import { Observable } from 'rxjs';
+import { CorrelationIdStorage } from '../monitoring/correlation-id.storage';
+
+@Injectable()
+export class CorrelationHttpInterceptor implements NestInterceptor {
+  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
+    const correlationId = CorrelationIdStorage.getCorrelationId();
+    if (correlationId) {
+      // If we are dealing with standard req/res, 
+      // the middleware already handled it. 
+    }
+    return next.handle();
+  }
+}

middleware/src/monitoring/correlation-id.middleware.ts

@@ -0,0 +1,28 @@
+import { Injectable, NestMiddleware } from '@nestjs/common';
+import { Request, Response, NextFunction } from 'express';
+import { randomUUID } from 'node:crypto';
+import { CorrelationIdStorage } from './correlation-id.storage';
+
+@Injectable()
+export class CorrelationIdMiddleware implements NestMiddleware {
+  private readonly HEADER_NAME = 'X-Correlation-ID';
+
+  use(req: Request, res: Response, next: NextFunction) {
+    // 1. Extract from header or generate new UUID v4
+    const correlationId = (req.header(this.HEADER_NAME) || randomUUID()) as string;
+
+    // 2. Attach to request headers for propagation
+    req.headers[this.HEADER_NAME.toLowerCase()] = correlationId;
+
+    // 3. Attach to response headers
+    res.setHeader(this.HEADER_NAME, correlationId);
+
+    // 4. Run the rest of the request lifecycle within CorrelationIdStorage context
+    const userId = (req as any).user?.id || (req as any).userId;
+    CorrelationIdStorage.run(correlationId, userId, () => {
+      // 5. Store in request object as well for easy access without storage if needed
+      (req as any).correlationId = correlationId;
+      next();
+    });
+  }
+}

middleware/src/monitoring/correlation-id.storage.ts

@@ -0,0 +1,22 @@
+import { AsyncLocalStorage } from 'node:async_hooks';
+
+export interface CorrelationContext {
+  correlationId: string;
+  userId?: string;
+}
+
+export class CorrelationIdStorage {
+  private static readonly storage = new AsyncLocalStorage<CorrelationContext>();
+
+  static run<R>(correlationId: string, userId: string | undefined, fn: () => R): R {
+    return this.storage.run({ correlationId, userId }, fn);
+  }
+
+  static getCorrelationId(): string | undefined {
+    return this.storage.getStore()?.correlationId;
+  }
+
+  static getUserId(): string | undefined {
+    return this.storage.getStore()?.userId;
+  }
+}

middleware/src/monitoring/correlation-logger.service.ts

@@ -0,0 +1,51 @@
+import { Injectable, LoggerService, Scope } from '@nestjs/common';
+import { CorrelationIdStorage } from './correlation-id.storage';
+
+@Injectable({ scope: Scope.TRANSIENT })
+export class CorrelationLoggerService implements LoggerService {
+  private context?: string;
+
+  constructor(context?: string) {
+    this.context = context;
+  }
+
+  setContext(context: string) {
+    this.context = context;
+  }
+
+  log(message: any, context?: string) {
+    this.printLog('info', message, context);
+  }
+
+  error(message: any, trace?: string, context?: string) {
+    this.printLog('error', message, context, trace);
+  }
+
+  warn(message: any, context?: string) {
+    this.printLog('warn', message, context);
+  }
+
+  debug(message: any, context?: string) {
+    this.printLog('debug', message, context);
+  }
+
+  verbose(message: any, context?: string) {
+    this.printLog('verbose', message, context);
+  }
+
+  private printLog(level: string, message: any, context?: string, trace?: string) {
+    const correlationId = CorrelationIdStorage.getCorrelationId();
+    const userId = CorrelationIdStorage.getUserId();
+    const logOutput = {
+      timestamp: new Date().toISOString(),
+      level,
+      correlationId: correlationId || 'N/A',
+      userId: userId || 'N/A',
+      context: context || this.context,
+      message: typeof message === 'string' ? message : JSON.stringify(message),
+      ...(trace ? { trace } : {}),
+    };
+
+    console.log(JSON.stringify(logOutput));
+  }
+}

middleware/src/monitoring/correlation-propagation.utils.ts

@@ -0,0 +1,30 @@
+import { CorrelationIdStorage } from './correlation-id.storage';
+
+/**
+ * Utility to get headers for downstream calls to propagate correlation ID.
+ */
+export const getCorrelationHeaders = (headers: Record<string, string> = {}) => {
+  const correlationId = CorrelationIdStorage.getCorrelationId();
+  if (correlationId) {
+    return {
+      ...headers,
+      'X-Correlation-ID': correlationId,
+    };
+  }
+  return headers;
+};
+
+/**
+ * Wraps a function to execute within the current correlation context.
+ * Useful for async workers, emitters, and timers.
+ */
+export const withCorrelation = <T extends (...args: any[]) => any>(fn: T): T => {
+  const correlationId = CorrelationIdStorage.getCorrelationId();
+  const userId = CorrelationIdStorage.getUserId();
+  if (!correlationId) {
+    return fn;
+  }
+  return ((...args: any[]) => {
+    return CorrelationIdStorage.run(correlationId, userId, () => fn(...args));
+  }) as T;
+};

middleware/src/monitoring/correlation.module.ts

@@ -0,0 +1,14 @@
+import { Module, Global } from '@nestjs/common';
+import { CorrelationIdMiddleware } from './correlation-id.middleware';
+import { CorrelationLoggerService } from './correlation-logger.service';
+
+@Global()
+@Module({
+  providers: [
+    CorrelationLoggerService,
+  ],
+  exports: [
+    CorrelationLoggerService,
+  ],
+})
+export class CorrelationModule {}

middleware/src/monitoring/index.ts

@@ -1,3 +1,6 @@
-// Placeholder: monitoring middleware exports will live here.
-
-export const __monitoringPlaceholder = true;
+export * from './correlation-id.middleware';
+export * from './correlation-id.storage';
+export * from './correlation-logger.service';
+export * from './correlation.module';
+export * from './correlation-exception-filter';
+export * from './correlation-propagation.utils';