RedditBot/src/web/events/express-sse-adapter.ts

import type { RequestHandler, Response } from 'express';
import {
  systemEventClock,
  type EventClock,
} from './event-dependencies.js';
import type { RealtimeEvent } from './event-types.js';
import type {
  RealtimeEventHub,
  RealtimeEventSubscription,
} from './realtime-event-hub.js';

const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000;
const DEFAULT_MAX_PENDING_FRAMES = 100;

export interface RecurringTask {
  cancel(): void;
}

export interface RecurringTaskScheduler {
  every(intervalMs: number, callback: () => void): RecurringTask;
}

export const systemRecurringTaskScheduler: RecurringTaskScheduler = {
  every(intervalMs, callback) {
    const timer = setInterval(callback, intervalMs);
    timer.unref();
    return { cancel: () => clearInterval(timer) };
  },
};

export interface ExpressSseHandlerOptions {
  hub: RealtimeEventHub;
  scheduler?: RecurringTaskScheduler;
  clock?: EventClock;
  heartbeatIntervalMs?: number;
  maxPendingFrames?: number;
}

function isResponseClosed(response: Response): boolean {
  return response.destroyed || response.writableEnded;
}

function serializeEvent(event: RealtimeEvent): string {
  const safeId = event.id.replace(/[\r\n\0]/g, '');
  return `id: ${safeId}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`;
}

/**
 * Adapts the transport-agnostic realtime hub to an Express SSE endpoint.
 * The returned handler owns only the lifetime of its individual connection.
 */
export function createExpressSseHandler(
  options: ExpressSseHandlerOptions,
): RequestHandler {
  const heartbeatIntervalMs =
    options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS;
  const maxPendingFrames =
    options.maxPendingFrames ?? DEFAULT_MAX_PENDING_FRAMES;
  if (!Number.isInteger(heartbeatIntervalMs) || heartbeatIntervalMs <= 0) {
    throw new RangeError('heartbeatIntervalMs must be a positive integer');
  }
  if (!Number.isInteger(maxPendingFrames) || maxPendingFrames <= 0) {
    throw new RangeError('maxPendingFrames must be a positive integer');
  }

  const scheduler = options.scheduler ?? systemRecurringTaskScheduler;
  const clock = options.clock ?? systemEventClock;

  return (request, response): void => {
    if (isResponseClosed(response)) {
      return;
    }

    let active = true;
    let waitingForDrain = false;
    let subscription: RealtimeEventSubscription | undefined;
    let heartbeat: RecurringTask | undefined;
    const pendingFrames: string[] = [];

    const removeConnectionListeners = (): void => {
      request.removeListener('close', handleConnectionClose);
      response.removeListener('close', handleConnectionClose);
      response.removeListener('drain', drainPendingFrames);
    };

    const close = (endResponse: boolean): void => {
      if (!active) {
        return;
      }
      active = false;
      waitingForDrain = false;
      pendingFrames.length = 0;
      removeConnectionListeners();
      heartbeat?.cancel();
      subscription?.unsubscribe();
      if (endResponse && !isResponseClosed(response)) {
        try {
          response.end();
        } catch {
          // The socket is already unusable; all owned resources are released.
        }
      }
    };

    const handleConnectionClose = (): void => close(false);

    const waitForDrain = (): void => {
      if (active && !waitingForDrain) {
        waitingForDrain = true;
        response.once('drain', drainPendingFrames);
      }
    };

    const writeFrame = (frame: string): boolean => {
      if (!active || isResponseClosed(response)) {
        close(false);
        return false;
      }
      try {
        if (!response.write(frame)) {
          waitForDrain();
        }
        return true;
      } catch {
        close(true);
        return false;
      }
    };

    const enqueueOrWrite = (frame: string): void => {
      if (!active) {
        return;
      }
      if (!waitingForDrain) {
        writeFrame(frame);
        return;
      }
      if (pendingFrames.length >= maxPendingFrames) {
        close(true);
        return;
      }
      pendingFrames.push(frame);
    };

    function drainPendingFrames(): void {
      if (!active || isResponseClosed(response)) {
        close(false);
        return;
      }
      waitingForDrain = false;
      while (active && !waitingForDrain && pendingFrames.length > 0) {
        const frame = pendingFrames.shift();
        if (frame !== undefined) {
          writeFrame(frame);
        }
      }
    }

    request.once('close', handleConnectionClose);
    response.once('close', handleConnectionClose);

    try {
      response.statusCode = 200;
      response.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
      response.setHeader('Cache-Control', 'no-cache, no-transform');
      response.setHeader('Connection', 'keep-alive');
      response.setHeader('X-Accel-Buffering', 'no');
      response.flushHeaders();
    } catch {
      close(true);
      return;
    }

    enqueueOrWrite(': connected\n\n');
    if (!active) {
      return;
    }

    try {
      heartbeat = scheduler.every(heartbeatIntervalMs, () => {
        if (!active || waitingForDrain) {
          return;
        }
        try {
          enqueueOrWrite(`: heartbeat ${clock.now().toISOString()}\n\n`);
        } catch {
          close(true);
        }
      });
    } catch {
      close(true);
      return;
    }
    if (!active) {
      heartbeat.cancel();
      return;
    }

    try {
      // A first-time subscriber repairs authoritative dashboard snapshots on
      // open and only needs events published from this point onwards. Replaying
      // the entire buffer here can overflow the bounded response queue and
      // force EventSource into a reconnect loop. A genuine reconnect supplies
      // Last-Event-ID and still receives every buffered event after its cursor.
      const requestedLastEventId = request.get('Last-Event-ID');
      const replayCursor = requestedLastEventId
        ?? options.hub.recent(1).at(-1)?.id;
      subscription = options.hub.subscribe(
        (event) => {
          if (!active) {
            return;
          }
          try {
            enqueueOrWrite(serializeEvent(event));
          } catch {
            close(true);
          }
        },
        { lastEventId: replayCursor },
      );
    } catch {
      close(true);
      return;
    }

    // Replay is synchronous, so a failed replay can close before assignment.
    if (!active) {
      subscription.unsubscribe();
    }
  };
}