email-amazon/email-worker-nodejs/metrics.ts

/**
 * Prometheus metrics collection
 *
 * Uses prom-client. Falls back gracefully if not available.
 */

import { log } from './logger.js';
import type * as PromClientTypes from 'prom-client';

// prom-client is optional — import dynamically
let promClient: typeof PromClientTypes | null = null;
try {
  promClient = require('prom-client') as typeof PromClientTypes;
} catch {
  // not installed
}

// ---------------------------------------------------------------------------
// Metric instances (created lazily if prom-client is available)
// ---------------------------------------------------------------------------
let emailsProcessed: any;
let emailsInFlight: any;
let processingTime: any;
let queueSize: any;
let bouncesProcessed: any;
let autorepliesSent: any;
let forwardsSent: any;
let blockedSenders: any;

function initMetrics(): void {
  if (!promClient) return;
  const { Counter, Gauge, Histogram } = promClient;

  emailsProcessed = new Counter({
    name: 'emails_processed_total',
    help: 'Total emails processed',
    labelNames: ['domain', 'status'],
  });
  emailsInFlight = new Gauge({
    name: 'emails_in_flight',
    help: 'Emails currently being processed',
  });
  processingTime = new Histogram({
    name: 'email_processing_seconds',
    help: 'Time to process email',
    labelNames: ['domain'],
  });
  queueSize = new Gauge({
    name: 'queue_messages_available',
    help: 'Messages in queue',
    labelNames: ['domain'],
  });
  bouncesProcessed = new Counter({
    name: 'bounces_processed_total',
    help: 'Bounce notifications processed',
    labelNames: ['domain', 'type'],
  });
  autorepliesSent = new Counter({
    name: 'autoreplies_sent_total',
    help: 'Auto-replies sent',
    labelNames: ['domain'],
  });
  forwardsSent = new Counter({
    name: 'forwards_sent_total',
    help: 'Forwards sent',
    labelNames: ['domain'],
  });
  blockedSenders = new Counter({
    name: 'blocked_senders_total',
    help: 'Emails blocked by blacklist',
    labelNames: ['domain'],
  });
}

// ---------------------------------------------------------------------------
// MetricsCollector
// ---------------------------------------------------------------------------
export class MetricsCollector {
  public readonly enabled: boolean;

  constructor() {
    this.enabled = !!promClient;
    if (this.enabled) initMetrics();
  }

  incrementProcessed(domain: string, status: string): void {
    emailsProcessed?.labels(domain, status).inc();
  }

  incrementInFlight(): void {
    emailsInFlight?.inc();
  }

  decrementInFlight(): void {
    emailsInFlight?.dec();
  }

  observeProcessingTime(domain: string, seconds: number): void {
    processingTime?.labels(domain).observe(seconds);
  }

  setQueueSize(domain: string, size: number): void {
    queueSize?.labels(domain).set(size);
  }

  incrementBounce(domain: string, bounceType: string): void {
    bouncesProcessed?.labels(domain, bounceType).inc();
  }

  incrementAutoreply(domain: string): void {
    autorepliesSent?.labels(domain).inc();
  }

  incrementForward(domain: string): void {
    forwardsSent?.labels(domain).inc();
  }

  incrementBlocked(domain: string): void {
    blockedSenders?.labels(domain).inc();
  }
}

// ---------------------------------------------------------------------------
// Start metrics HTTP server
// ---------------------------------------------------------------------------
export async function startMetricsServer(port: number): Promise<MetricsCollector | null> {
  if (!promClient) {
    log('⚠ Prometheus client not installed, metrics disabled', 'WARNING');
    return null;
  }

  try {
    const { createServer } = await import('node:http');
    const { register } = promClient;

    const server = createServer(async (_req, res) => {
      try {
        res.setHeader('Content-Type', register.contentType);
        res.end(await register.metrics());
      } catch {
        res.statusCode = 500;
        res.end();
      }
    });

    server.listen(port, () => {
      log(`Prometheus metrics on port ${port}`);
    });

    return new MetricsCollector();
  } catch (err: any) {
    log(`Failed to start metrics server: ${err.message ?? err}`, 'ERROR');
    return null;
  }
}