Structured Logging: Standards That Stick at Scale

import pino from 'pino'

const logger = pino({
  formatters: {
    level(label) {
      return { 'log.level': label }
    },
  },
  messageKey: 'message',
  timestamp: () => `,"@timestamp":"${new Date().toISOString()}"`,
  base: {
    'service.name': process.env.SERVICE_NAME,
    'service.version': process.env.SERVICE_VERSION,
    'service.environment': process.env.NODE_ENV,
    'service.node.name': process.env.HOSTNAME || process.env.POD_NAME,
  },
})

export { logger }

logger.info({
  'http.request.method': 'POST',
  'http.request.path': '/api/orders',
  'http.response.status_code': 201,
  'user.id': 'usr_12345',
  'event.category': ['web'],
  'event.action': 'order_created',
  'event.outcome': 'success',
}, 'Order created successfully')

// Correlation ID hierarchy
interface CorrelationContext {
  // Request-level: entire user request across all services
  requestId: string       // Generated at edge/API gateway

  // Trace-level: distributed trace context (OpenTelemetry)
  traceId: string         // Propagated via W3C Trace Context
  spanId: string          // Current operation within trace
  parentSpanId?: string   // Parent operation

  // Session-level: user session across multiple requests
  sessionId?: string      // From session cookie or JWT

  // Business transaction: logical operation across requests
  transactionId?: string  // e.g., order fulfillment saga

  // Causation: what triggered this operation
  causationId?: string    // ID of the event/message that caused this
}

// Example log showing all correlation levels
const logEntry = {
  '@timestamp': '2024-01-15T14:30:00.123Z',
  'message': 'Payment authorized',

  // All correlation IDs present
  'trace.id': '4bf92f3577b34da6a3ce929d0e0e4736',
  'span.id': '00f067aa0ba902b7',
  'transaction.id': 'order-12345',           // Business transaction
  'request.id': 'req_abc123',                // Edge-assigned request ID
  'session.id': 'sess_xyz789',               // User session
  'event.causation_id': 'order_created_evt', // What triggered this
}

// This single log entry enables:
// - trace.id query: all logs for this request across services
// - transaction.id query: entire order fulfillment saga
// - session.id query: everything this user did in their session
// - causation_id query: what event triggered this payment

// HTTP header propagation
const correlationHeaders = {
  // W3C Trace Context (standard)
  traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',
  tracestate: 'vendor=value',

  // Custom headers for non-trace IDs
  'x-request-id': 'req_abc123',
  'x-session-id': 'sess_xyz789',
  'x-transaction-id': 'order-12345',
  'x-causation-id': 'order_created_evt',
}

// Express middleware for extraction and injection
import { AsyncLocalStorage } from 'async_hooks'

interface RequestContext {
  traceId: string
  spanId: string
  requestId: string
  sessionId?: string
  transactionId?: string
}

const contextStorage = new AsyncLocalStorage<RequestContext>()

function correlationMiddleware(req: Request, res: Response, next: NextFunction) {
  // Extract or generate IDs
  const traceHeader = req.headers['traceparent'] as string
  const { traceId, spanId } = parseTraceParent(traceHeader) || {
    traceId: generateTraceId(),
    spanId: generateSpanId()
  }

  const context: RequestContext = {
    traceId,
    spanId,
    requestId: req.headers['x-request-id'] as string || generateRequestId(),
    sessionId: req.headers['x-session-id'] as string,
    transactionId: req.headers['x-transaction-id'] as string,
  }

  // Set response headers for debugging
  res.setHeader('x-trace-id', traceId)
  res.setHeader('x-request-id', context.requestId)

  // Run request with context
  contextStorage.run(context, () => next())
}

// Get current context from anywhere
function getCorrelationContext(): RequestContext | undefined {
  return contextStorage.getStore()
}

// HTTP client that propagates context
async function fetchWithCorrelation(url: string, options: RequestInit = {}) {
  const context = getCorrelationContext()

  const headers = new Headers(options.headers)

  if (context) {
    // Generate new span ID for outbound call (same trace, new span)
    headers.set('traceparent', formatTraceParent(context.traceId, generateSpanId()))
    headers.set('x-request-id', context.requestId)
    if (context.sessionId) headers.set('x-session-id', context.sessionId)
    if (context.transactionId) headers.set('x-transaction-id', context.transactionId)
  }

  return fetch(url, { ...options, headers })
}

// Message envelope carries correlation through queues
interface MessageEnvelope<T> {
  messageId: string
  timestamp: string
  type: string
  correlation: {
    traceId: string
    causationId: string       // What triggered this message
    originRequestId?: string  // Original HTTP request that started the chain
  }
  payload: T
}

// Producer: wrap payload with correlation from current context
async function publishEvent<T>(type: string, payload: T): Promise<void> {
  const ctx = getCorrelationContext()
  const envelope: MessageEnvelope<T> = {
    messageId: generateMessageId(),
    timestamp: new Date().toISOString(),
    type,
    correlation: {
      traceId: ctx?.traceId || generateTraceId(),
      causationId: ctx?.causationId || 'http_request',
      originRequestId: ctx?.requestId,
    },
    payload,
  }
  await messageQueue.publish(type, envelope)
}

// Consumer: restore correlation before processing
async function handleMessage<T>(envelope: MessageEnvelope<T>): Promise<void> {
  const context: RequestContext = {
    traceId: envelope.correlation.traceId,
    spanId: generateSpanId(),
    causationId: envelope.messageId,  // This message is now the cause of downstream work
  }

  await contextStorage.run(context, () => processMessage(envelope.payload))
}

// Type definitions enforce ECS compliance at compile time
interface HttpLogFields {
  'http.request.method': string
  'http.request.path': string
  'http.response.status_code': number
  'http.response.time_ms': number
}

class StructuredLogger {
  private baseFields: Record<string, unknown>

  constructor(config: { serviceName: string; serviceVersion: string }) {
    this.baseFields = {
      'service.name': config.serviceName,
      'service.version': config.serviceVersion,
    }
  }

  private log(level: string, message: string, fields: Record<string, unknown>) {
    const context = getCorrelationContext()
    const entry = {
      '@timestamp': new Date().toISOString(),
      'log.level': level,
      message,
      ...this.baseFields,
      ...(context && { 'trace.id': context.traceId, 'span.id': context.spanId }),
      ...fields,
    }
    console.log(JSON.stringify(entry))
  }

  info(message: string, fields?: Record<string, unknown>) {
    this.log('info', message, fields || {})
  }

  error(message: string, error: Error, fields?: Record<string, unknown>) {
    this.log('error', message, {
      'error.type': error.name,
      'error.message': error.message,
      'error.stack_trace': error.stack,
      ...fields,
    })
  }

  // Typed method: missing http.response.status_code = compile error
  httpRequest(message: string, fields: HttpLogFields & Record<string, unknown>) {
    this.log('info', message, { 'event.category': ['web'], ...fields })
  }

  // Child logger inherits base fields + adds new ones
  child(fields: Record<string, unknown>): StructuredLogger {
    const child = Object.create(this)
    child.baseFields = { ...this.baseFields, ...fields }
    return child
  }
}

// Usage
const logger = new StructuredLogger({
  serviceName: 'order-service',
  serviceVersion: '1.2.3',
})

const orderLogger = logger.child({ 'order.id': 'ord_12345' })

orderLogger.httpRequest('Order created', {
  'http.request.method': 'POST',
  'http.request.path': '/api/orders',
  'http.response.status_code': 201,
  'http.response.time_ms': 45,
  'event.action': 'order_created',
})

// Middleware creates a pre-configured logger for each request
function loggingMiddleware(req: Request, res: Response, next: NextFunction) {
  const context = getCorrelationContext()
  const startTime = Date.now()

  // Create logger with all request context pre-bound
  req.log = logger.child({
    'http.request.method': req.method,
    'http.request.path': req.path,
    'trace.id': context?.traceId,
    'span.id': context?.spanId,
    ...(req.user && { 'user.id': req.user.id }),
  })

  req.log.info('Request started', { 'event.action': 'http_request_started' })

  res.on('finish', () => {
    req.log.httpRequest('Request completed', {
      'http.request.method': req.method,
      'http.request.path': req.path,
      'http.response.status_code': res.statusCode,
      'http.response.time_ms': Date.now() - startTime,
      'event.outcome': res.statusCode < 400 ? 'success' : 'failure',
    })
  })

  next()
}

app.post('/api/orders', async (req, res) => {
  req.log.info('Processing order', { 'order.item_count': req.body.items.length })

  const order = await createOrder(req.body)
  req.log.info('Order created', { 'order.id': order.id, 'event.action': 'order_created' })

  res.status(201).json(order)
})

class LogRedactor {
  private sensitivePatterns = /password|secret|key|token|auth|credential/i

  redact(obj: Record<string, unknown>): Record<string, unknown> {
    const result: Record<string, unknown> = {}

    for (const [key, value] of Object.entries(obj)) {
      if (this.sensitivePatterns.test(key)) {
        result[key] = '[REDACTED]'
      } else if (typeof value === 'object' && value !== null) {
        result[key] = this.redact(value as Record<string, unknown>)
      } else if (typeof value === 'string') {
        result[key] = this.maskValue(key, value)
      } else {
        result[key] = value
      }
    }
    return result
  }

  private maskValue(key: string, value: string): string {
    if (key.includes('email')) {
      const [local, domain] = value.split('@')
      return `${local[0]}***@${domain}`
    }
    if (key.includes('phone') || key.includes('card')) {
      return value.replace(/\d(?=\d{4})/g, '*')
    }
    return value
  }
}

class FieldEncryptor {
  constructor(private key: Buffer, private keyId: string) {}

  encrypt(value: string): EncryptedField {
    const iv = randomBytes(16)
    const cipher = createCipheriv('aes-256-gcm', this.key, iv)
    const encrypted = cipher.update(value, 'utf8', 'base64') + cipher.final('base64')

    return {
      _encrypted: true,
      keyId: this.keyId,
      iv: iv.toString('base64'),
      authTag: cipher.getAuthTag().toString('base64'),
      ciphertext: encrypted,
    }
  }
}

// Usage: combine masking and encryption
logger.info('Customer registered', {
  'user.id': userId,
  'customer.email': redactor.maskEmail(email),       // Masked: debuggable
  'customer.address': encryptor.encrypt(address),    // Encrypted: recoverable
})

// Aggregation: count instead of individual logs
interface AggregatedLog {
  '@timestamp': string
  'log.level': 'info'
  'message': string
  'event.action': string
  'aggregation.count': number
  'aggregation.window_seconds': number
}

// Instead of 1000 "connection_opened" logs per minute:
const aggregatedConnectionLog: AggregatedLog = {
  '@timestamp': '2024-01-15T14:31:00Z',
  'log.level': 'info',
  'message': 'Database connections aggregated',
  'event.action': 'db_connection_aggregated',
  'aggregation.count': 1247,
  'aggregation.window_seconds': 60,
}

import { describe, it, expect, beforeEach, vi } from 'vitest'
import Ajv from 'ajv'
import addFormats from 'ajv-formats'

const ecsLogSchema = {
  type: 'object',
  required: ['@timestamp', 'log.level', 'message', 'service.name'],
  properties: { '@timestamp': { type: 'string', format: 'date-time' } },
  additionalProperties: true,
}

describe('Structured Logger Schema Compliance', () => {
  let logOutput: string[] = []
  const validate = new Ajv({ allErrors: true }).addFormats(addFormats).compile(ecsLogSchema)

  beforeEach(() => {
    logOutput = []
    vi.spyOn(console, 'log').mockImplementation((msg: string) => logOutput.push(msg))
  })

  it('emits valid ECS-compliant logs', () => {
    logger.info('Test message', { 'user.id': 'usr_123' })
    const entry = JSON.parse(logOutput[0])

    expect(validate(entry)).toBe(true)
    expect(entry['@timestamp']).toMatch(/^\d{4}-\d{2}-\d{2}T/)
  })

  it('includes trace context when available', () => {
    logger.info('With trace', {
      'trace.id': '4bf92f3577b34da6a3ce929d0e0e4736',
      'span.id': '00f067aa0ba902b7',
    })
    const entry = JSON.parse(logOutput[0])

    expect(entry['trace.id']).toMatch(/^[a-f0-9]{32}$/)
    expect(entry['span.id']).toMatch(/^[a-f0-9]{16}$/)
  })
})

describe('Correlation ID Propagation', () => {
  it('extracts trace context from W3C traceparent header', async () => {
    const req = createMockRequest({
      headers: { traceparent: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01' },
    })

    await new Promise<void>(resolve => {
      correlationMiddleware(req, createMockResponse(), () => {
        expect(getCorrelationContext()?.traceId).toBe('4bf92f3577b34da6a3ce929d0e0e4736')
        resolve()
      })
    })
  })

  it('generates IDs when headers are missing', async () => {
    const req = createMockRequest({ headers: {} })

    await new Promise<void>(resolve => {
      correlationMiddleware(req, createMockResponse(), () => {
        expect(getCorrelationContext()?.traceId).toMatch(/^[a-f0-9]{32}$/)
        resolve()
      })
    })
  })

  it('propagates context to outbound requests', async () => {
    const mockFetch = vi.fn().mockResolvedValue(new Response('{}'))
    const testContext = { traceId: '4bf92f3577b34da6a', spanId: '00f067aa', requestId: 'req_123' }

    await contextStorage.run(testContext, () =>
      fetchWithCorrelation('https://api.example.com', {}, mockFetch)
    )

    expect(mockFetch.mock.calls[0][1].headers.get('traceparent')).toContain(testContext.traceId)
  })
})