pubsub.ts

/**
 * pubsub.ts — Kafka-to-AsyncIterator bridge for Apollo Server subscriptions.
 *
 * This module creates an EventEmitter-based PubSub that is fed by Kafka consumers.
 * When a subscription resolver calls asyncIterator('NOTIFICATION_CREATED'),
 * it receives events consumed from the Kafka `notification-events` topic.
 *
 * Why EventEmitter and not graphql-subscriptions PubSub?
 * - graphql-subscriptions PubSub is in-memory only and deprecated for production.
 * - We need Kafka as the event source for multi-pod distribution.
 * - This bridge consumes from Kafka → emits to local EventEmitter → AsyncIterator.
 */

import { EventEmitter } from "events";
import {
  createConsumer,
  TOPIC_NOTIFICATIONS,
  TOPIC_SYSTEM_ALERTS,
} from "./kafka.js";
import type { Consumer } from "kafkajs";

// Increase default max listeners (each subscription adds a listener)
const emitter = new EventEmitter();
emitter.setMaxListeners(100);

// ── Event names (used by subscription resolvers) ──
export const NOTIFICATION_CREATED = "NOTIFICATION_CREATED";
export const SYSTEM_ALERT = "SYSTEM_ALERT";

let consumer: Consumer | null = null;

/**
 * Start consuming from Kafka topics and bridge events to the local EventEmitter.
 * Called once at server startup.
 */
export async function startKafkaBridge(): Promise<void> {
  consumer = createConsumer("notifications-subgraph-group");
  await consumer.connect();
  console.log("✅ Kafka consumer connected");

  await consumer.subscribe({
    topics: [TOPIC_NOTIFICATIONS, TOPIC_SYSTEM_ALERTS],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      if (!message.value) return;

      try {
        const payload = JSON.parse(message.value.toString());
        console.log(
          `📨 Kafka [${topic}] partition=${partition} offset=${message.offset}:`,
          payload.id
        );

        if (topic === TOPIC_NOTIFICATIONS) {
          emitter.emit(NOTIFICATION_CREATED, payload);
        } else if (topic === TOPIC_SYSTEM_ALERTS) {
          emitter.emit(SYSTEM_ALERT, payload);
        }
      } catch (err) {
        console.error("❌ Failed to parse Kafka message:", err);
      }
    },
  });

  console.log(
    `✅ Kafka bridge started — listening on [${TOPIC_NOTIFICATIONS}, ${TOPIC_SYSTEM_ALERTS}]`
  );
}

/**
 * Creates an AsyncIterableIterator that yields events for a given event name.
 * Used by subscription resolvers to return an async stream.
 *
 * @param eventName - The event name to listen for (e.g., NOTIFICATION_CREATED)
 * @param filterFn - Optional filter predicate applied to each event
 */
export function asyncIterator<T>(
  eventName: string,
  filterFn?: (payload: T) => boolean
): AsyncIterableIterator<T> {
  const pullQueue: Array<(value: IteratorResult<T>) => void> = [];
  const pushQueue: T[] = [];
  let done = false;

  const pushValue = (payload: T) => {
    // Apply filter if provided
    if (filterFn && !filterFn(payload)) return;

    if (pullQueue.length > 0) {
      // A consumer is waiting — resolve immediately
      const resolve = pullQueue.shift()!;
      resolve({ value: payload, done: false });
    } else {
      // Buffer the event until someone pulls
      pushQueue.push(payload);
    }
  };

  emitter.on(eventName, pushValue);

  return {
    next(): Promise<IteratorResult<T>> {
      if (done) return Promise.resolve({ value: undefined as any, done: true });

      if (pushQueue.length > 0) {
        return Promise.resolve({ value: pushQueue.shift()!, done: false });
      }

      return new Promise((resolve) => pullQueue.push(resolve));
    },

    return(): Promise<IteratorResult<T>> {
      done = true;
      emitter.removeListener(eventName, pushValue);
      // Drain any waiting consumers
      for (const resolve of pullQueue) {
        resolve({ value: undefined as any, done: true });
      }
      pullQueue.length = 0;
      pushQueue.length = 0;
      return Promise.resolve({ value: undefined as any, done: true });
    },

    throw(err: Error): Promise<IteratorResult<T>> {
      done = true;
      emitter.removeListener(eventName, pushValue);
      return Promise.reject(err);
    },

    [Symbol.asyncIterator]() {
      return this;
    },
  };
}

/**
 * Stop the Kafka consumer. Called during graceful shutdown.
 */
export async function stopKafkaBridge(): Promise<void> {
  if (consumer) {
    await consumer.disconnect();
    console.log("🔌 Kafka consumer disconnected");
  }
}