RedditBot/apps/dashboard/src/features/operations/realtime-client.ts

import type { QueryKey } from '@tanstack/react-query';
import {
  DashboardEventSchema,
  type DashboardEvent,
} from '../../../../../packages/api-contracts/src/index.js';

export const DASHBOARD_EVENT_TYPES = [
  'bot.status.changed',
  'message.sent',
  'message.failed',
  'conversation.updated',
  'reply.received',
  'sync.started',
  'sync.progress',
  'sync.completed',
  'sync.failed',
  'rate-limit.changed',
  'reddit.session.changed',
  'log.created',
] as const;

export interface DashboardEventSource {
  addEventListener(type: string, listener: (event: unknown) => void): void;
  close(): void;
}

export interface RealtimeConnectionSnapshot {
  status: 'idle' | 'connecting' | 'connected' | 'reconnecting' | 'closed';
  isStale: boolean;
  lastEventId?: string;
}

export interface DashboardRealtimeClient {
  start(): void;
  stop(): void;
  getSnapshot(): RealtimeConnectionSnapshot;
}

export interface DashboardRealtimeClientOptions {
  url?: string;
  sourceFactory(url: string): DashboardEventSource;
  onEvent(event: DashboardEvent): void;
  onConnected?(context: { isReconnect: boolean }): void;
  onStateChange?(snapshot: RealtimeConnectionSnapshot): void;
  onInvalidEvent?(value: unknown): void;
}

function eventData(value: unknown): string | undefined {
  if (typeof value !== 'object' || value === null || !('data' in value)) return undefined;
  return typeof value.data === 'string' ? value.data : undefined;
}

export function createDashboardRealtimeClient(
  options: DashboardRealtimeClientOptions,
): DashboardRealtimeClient {
  let source: DashboardEventSource | undefined;
  let snapshot: RealtimeConnectionSnapshot = { status: 'idle', isStale: true };
  let hasConnected = false;

  function update(next: RealtimeConnectionSnapshot): void {
    snapshot = next;
    options.onStateChange?.(snapshot);
  }

  function receive(value: unknown): void {
    const data = eventData(value);
    if (data === undefined) {
      options.onInvalidEvent?.(value);
      return;
    }
    let decoded: unknown;
    try {
      decoded = JSON.parse(data);
    } catch {
      options.onInvalidEvent?.(data);
      return;
    }
    const result = DashboardEventSchema.safeParse(decoded);
    if (!result.success) {
      options.onInvalidEvent?.(decoded);
      return;
    }
    snapshot = { ...snapshot, lastEventId: result.data.id };
    options.onEvent(result.data);
  }

  return {
    start() {
      if (source !== undefined) return;
      update({ ...snapshot, status: 'connecting', isStale: true });
      source = options.sourceFactory(options.url ?? '/api/events/stream');
      source.addEventListener('open', () => {
        update({ ...snapshot, status: 'connected', isStale: false });
        options.onConnected?.({ isReconnect: hasConnected });
        hasConnected = true;
      });
      source.addEventListener('error', () => {
        update({ ...snapshot, status: 'reconnecting', isStale: true });
      });
      for (const type of DASHBOARD_EVENT_TYPES) source.addEventListener(type, receive);
    },
    stop() {
      source?.close();
      source = undefined;
      update({ ...snapshot, status: 'closed', isStale: true });
    },
    getSnapshot() {
      return snapshot;
    },
  };
}

export function queryKeysForDashboardEvent(event: DashboardEvent): readonly QueryKey[] {
  switch (event.type) {
    case 'bot.status.changed':
      return [['dashboard', 'stats']];
    case 'rate-limit.changed':
      return [['dashboard', 'stats'], ['dashboard', 'timers']];
    case 'message.sent':
    case 'message.failed':
      return [['conversations'], ['performance']];
    case 'conversation.updated':
    case 'reply.received':
      return [['conversations'], ['operations', 'activity']];
    case 'sync.started':
    case 'sync.progress':
    case 'sync.completed':
    case 'sync.failed':
      return [['operations', 'sync-status'], ['conversations'], ['media']];
    case 'reddit.session.changed':
      return [['dashboard', 'login', 'status']];
    case 'log.created':
      return [];
  }
}

export function createBrowserEventSource(url: string): DashboardEventSource {
  const source = new EventSource(url);
  return {
    addEventListener(type, listener) {
      source.addEventListener(type, (event) => listener(event));
    },
    close() {
      source.close();
    },
  };
}