abdulive/packages/discord/src/gateway.ts

// Discord-Gateway ueber WebSocket. Nur so viel, wie ein Knopf braucht:
// verbinden, am Leben halten, Interaktionen durchreichen.
//
// Warum ueberhaupt eine dauerhafte Verbindung: Discord liefert Klicks auf
// einen Knopf entweder ans Gateway oder an eine oeffentlich erreichbare
// HTTPS-Adresse. Der Stream-PC hat keine, also Gateway. Die Verbindung haengt
// im Ingest-Daemon mit drin und kostet dort praktisch nichts.
//
// Absichten (Intents) stehen auf 0: Interaktionen kommen immer an, ohne
// Berechtigung fuer Nachrichteninhalte oder Mitgliederlisten. Was wir nicht
// anfordern, koennen wir auch nicht versehentlich mitlesen.

import { logger } from '#core/logger.ts';
import { forever } from '#core/retry.ts';

const log = logger('discord:gateway');

// Betriebsarten des Gateways
const OP = {
  DISPATCH: 0,
  HEARTBEAT: 1,
  IDENTIFY: 2,
  RESUME: 6,
  RECONNECT: 7,
  INVALID_SESSION: 9,
  HELLO: 10,
  HEARTBEAT_ACK: 11,
} as const;

/**
 * Schliessgruende, bei denen ein neuer Versuch nichts bringt: falsches Token,
 * falsche Absichten, falsche Fassung. Dagegen hilft nur eine Aenderung an der
 * App, und bis dahin waere jeder Wiederverbindungsversuch nur Laerm.
 */
const FATAL_CLOSE = new Set([4004, 4010, 4011, 4012, 4013, 4014]);

export type Payload = { op: number; d?: unknown; s?: number | null; t?: string | null };

/** Nur die Felder einer Interaktion, die hier gebraucht werden. */
export type Interaction = {
  id: string;
  token: string;
  type: number;
  application_id: string;
  guild_id?: string;
  channel_id?: string;
  data?: { custom_id?: string; component_type?: number };
  member?: { user: { id: string; username: string }; roles: string[] };
  user?: { id: string; username: string };
};

export type GatewayHandlers = {
  onReady?: (info: { user: { id: string; username: string }; sessionId: string }) => void;
  onInteraction?: (interaction: Interaction) => Promise<void> | void;
  onClosed?: (code: number, reason: string) => void;
};

export type GatewayClient = {
  /** Haelt die Verbindung; kehrt erst bei `signal.abort` oder einem harten Fehler zurueck. */
  run: (signal: AbortSignal) => Promise<void>;
  isConnected: () => boolean;
};

/** Was ueber eine Trennung hinweg erhalten bleibt, damit fortgesetzt statt neu begonnen wird. */
type Session = { id: string; resumeUrl: string; seq: number | null };

export function createGateway(
  token: string,
  restUrl: (path: string) => Promise<{ url: string }>,
  handlers: GatewayHandlers,
): GatewayClient {
  let session: Session | null = null;
  let connected = false;

  return {
    isConnected: () => connected,
    run: async (outer) => {
      // Bei einem harten Fehler brechen wir die Schleife selbst ab, statt
      // Discord alle paar Sekunden erneut mit demselben Token zu behelligen.
      const own = new AbortController();
      const signal = AbortSignal.any([outer, own.signal]);

      await forever(
        async (sig) => {
          try {
            await connectOnce(token, restUrl, handlers, sig, {
              get: () => session,
              set: (s) => { session = s; },
              setConnected: (v) => { connected = v; },
              giveUp: () => own.abort(),
            });
          } finally {
            connected = false;
          }
        },
        {
          signal,
          baseMs: 2000,
          maxMs: 120_000,
          onWait: (ms, n) => {
            if (own.signal.aborted) return;
            log.warn(`Gateway getrennt, neuer Versuch in ${Math.round(ms / 1000)} s (${n}.)`);
          },
        },
      );
    },
  };

  function connectOnce(
    botToken: string,
    getGateway: (path: string) => Promise<{ url: string }>,
    h: GatewayHandlers,
    signal: AbortSignal,
    store: {
      get: () => Session | null;
      set: (s: Session | null) => void;
      setConnected: (v: boolean) => void;
      /** Bricht die Wiederverbindungsschleife ab - nur bei harten Fehlern. */
      giveUp: () => void;
    },
  ): Promise<void> {
    return new Promise<void>((_resolve, reject) => {
      let ws: WebSocket | null = null;
      let beat: ReturnType<typeof setInterval> | null = null;
      let firstBeat: ReturnType<typeof setTimeout> | null = null;
      let acked = true;
      let seq: number | null = store.get()?.seq ?? null;
      let done = false;

      const finish = (why: string): void => {
        if (done) return;
        done = true;
        if (beat) clearInterval(beat);
        if (firstBeat) clearTimeout(firstBeat);
        store.setConnected(false);
        try { ws?.close(); } catch { /* schon zu */ }
        reject(new Error(why));
      };

      signal.addEventListener('abort', () => finish('abgebrochen'), { once: true });

      const send = (payload: Payload): void => {
        if (ws?.readyState === WebSocket.OPEN) ws.send(JSON.stringify(payload));
      };

      const heartbeat = (): void => {
        if (!acked) {
          // Discord hat den letzten Schlag nicht bestaetigt. Die Verbindung
          // sieht offen aus, ist es aber nicht mehr - abreissen und neu.
          log.warn('Kein Herzschlag bestaetigt, Verbindung gilt als tot');
          finish('Herzschlag ohne Antwort');
          return;
        }
        acked = false;
        send({ op: OP.HEARTBEAT, d: seq });
      };

      void (async () => {
        let url: string;
        const resume = store.get();
        try {
          url = resume?.resumeUrl ?? (await getGateway('/gateway/bot')).url;
        } catch (err) {
          finish(`Gateway-Adresse nicht abrufbar: ${(err as Error).message}`);
          return;
        }
        if (done) return;

        ws = new WebSocket(`${url}?v=10&encoding=json`);

        ws.addEventListener('close', (ev) => {
          const code = (ev as CloseEvent).code;
          const reason = (ev as CloseEvent).reason || '';
          h.onClosed?.(code, reason);
          if (FATAL_CLOSE.has(code)) {
            // Nicht fortsetzbar und nicht behebbar: Token oder Absichten.
            log.error(`Gateway hat endgueltig geschlossen (${code} ${reason}). ` +
              'Bot-Token pruefen und den Dienst neu starten.');
            store.set(null);
            store.giveUp();
          } else if (code === 4007 || code === 4009 || code === 1000 || code === 1001) {
            // Sitzung ist hin, aber ein neuer Anlauf lohnt.
            store.set(null);
          }
          finish(`geschlossen (${code})`);
        });

        ws.addEventListener('error', () => finish('Verbindungsfehler'));

        ws.addEventListener('message', (ev) => {
          let msg: Payload;
          try {
            msg = JSON.parse(String((ev as MessageEvent).data)) as Payload;
          } catch {
            return;
          }
          if (typeof msg.s === 'number') seq = msg.s;

          switch (msg.op) {
            case OP.HELLO: {
              const interval = (msg.d as { heartbeat_interval: number }).heartbeat_interval;
              // Der erste Schlag kommt versetzt, damit nicht alle Verbindungen
              // der Welt im selben Moment klopfen - so schreibt es Discord vor.
              firstBeat = setTimeout(() => {
                heartbeat();
                beat = setInterval(heartbeat, interval);
              }, Math.floor(interval * Math.random()));

              const s = store.get();
              if (s) {
                send({ op: OP.RESUME, d: { token: botToken, session_id: s.id, seq: s.seq } });
              } else {
                send({
                  op: OP.IDENTIFY,
                  d: {
                    token: botToken,
                    intents: 0,          // nur Interaktionen, nichts mitlesen
                    properties: { os: process.platform, browser: 'abdulive-toolkit', device: 'abdulive-toolkit' },
                  },
                });
              }
              return;
            }

            case OP.HEARTBEAT:
              // Discord darf jederzeit einen Schlag anfordern.
              send({ op: OP.HEARTBEAT, d: seq });
              return;

            case OP.HEARTBEAT_ACK:
              acked = true;
              return;

            case OP.RECONNECT:
              log.info('Gateway bittet um Wiederverbindung');
              finish('Wiederverbindung angefordert');
              return;

            case OP.INVALID_SESSION:
              // d = true heisst: die Sitzung laesst sich noch fortsetzen.
              if (msg.d !== true) store.set(null);
              finish('Sitzung ungueltig');
              return;

            case OP.DISPATCH:
              onDispatch(msg);
              return;

            default:
              return;
          }
        });

        function onDispatch(msg: Payload): void {
          if (msg.t === 'READY') {
            const d = msg.d as {
              session_id: string; resume_gateway_url: string;
              user: { id: string; username: string };
            };
            store.set({ id: d.session_id, resumeUrl: d.resume_gateway_url, seq });
            store.setConnected(true);
            log.info(`Gateway verbunden als ${d.user.username}`);
            h.onReady?.({ user: d.user, sessionId: d.session_id });
            return;
          }
          if (msg.t === 'RESUMED') {
            store.setConnected(true);
            log.info('Gateway-Sitzung fortgesetzt');
            return;
          }
          if (msg.t === 'INTERACTION_CREATE') {
            const s = store.get();
            if (s) store.set({ ...s, seq });
            void Promise.resolve(h.onInteraction?.(msg.d as Interaction))
              .catch((err: unknown) => log.error(`Interaktion nicht verarbeitet: ${String(err)}`));
          }
        }
      })();
    });
  }
}