GMW/services/discord-gateway/src/goLive/Streamer.ts

/**
 * Streamer — gateway-level GoLive controller. Ported from
 * @dank074/discord-video-stream Streamer.js.
 *
 * Drives the Discord gateway (VOICE_STATE_UPDATE, STREAM_CREATE, ...) and
 * hands back a VoiceConnection / StreamConnection once the media server
 * session is ready.
 */

import { EventEmitter } from "node:events";
import { GatewayOpCodes } from "./GatewayOpCodes.js";
import type { NativePeerConnection } from "./native.js";
import { StreamConnection } from "./StreamConnection.js";
import { generateStreamKey, parseStreamKey } from "./utils.js";
import { VoiceConnection } from "./VoiceConnection.js";
import type { WebRtcConnWrapper } from "./WebRtcWrapper.js";

/** Minimal surface of a discord.js-selfbot-v13 client used by Streamer. */
export interface StreamerClientLike {
  user: { id: string; username?: string } | null;
  token: string | null;
  on(
    event: "raw",
    listener: (packet: { t: string; d: unknown }) => void,
  ): unknown;
  ws: {
    broadcast(data: { op: number; d: unknown }): void;
  };
  guilds?: {
    // biome-ignore lint/suspicious/noExplicitAny: discord.js-selfbot client shape is dynamic
    fetch(id: string): Promise<any>;
  };
}

/** Minimal channel shape accepted by joinVoiceChannel. */
export interface VoiceChannelLike {
  id: string;
  type: string;
  guildId?: string | null;
}

export class Streamer {
  _voiceConnection: VoiceConnection | null = null;
  _client: StreamerClientLike;
  _gatewayEmitter = new EventEmitter();

  constructor(client: StreamerClientLike) {
    this._client = client;
    // listen for gateway dispatch events
    this.client.on("raw", (packet) => {
      const t = packet.t as string;
      if (
        t === "STREAM_CREATE" ||
        t === "STREAM_SERVER_UPDATE" ||
        t === "VOICE_STATE_UPDATE" ||
        t === "VOICE_SERVER_UPDATE"
      ) {
        console.log(
          `[goLive:Streamer] raw dispatch ${t}`,
          JSON.stringify(packet.d).slice(0, 220),
        );
      }
      this._gatewayEmitter.emit(t, packet.d);
    });
  }

  get client(): StreamerClientLike {
    return this._client;
  }

  get opts(): Record<string, unknown> {
    return {};
  }

  get voiceConnection(): VoiceConnection | null {
    return this._voiceConnection;
  }

  sendOpcode(code: number, data: unknown): void {
    // Direct instrumentation — bypasses the bootstrap debug filter (which
    // drops messages without [VOICE / [ffmpeg / error / stream).
    console.log(
      `[goLive:Streamer] sendOpcode op=${code} d=${JSON.stringify(data)}`,
    );
    this.client.ws.broadcast({ op: code, d: data });
  }

  joinVoiceChannel(channel: VoiceChannelLike): Promise<WebRtcConnWrapper> {
    let guildId: string | null = null;
    if (
      channel.type === "GUILD_STAGE_VOICE" ||
      channel.type === "GUILD_VOICE"
    ) {
      guildId = channel.guildId ?? null;
    }
    return this.joinVoice(guildId, channel.id);
  }

  /**
   * Joins a voice channel and resolves with the WebRtcConnWrapper when the
   * media session is ready.
   */
  joinVoice(
    guild_id: string | null,
    channel_id: string,
  ): Promise<WebRtcConnWrapper> {
    return new Promise((resolve, reject) => {
      if (!this.client.user) {
        reject(new Error("Client not logged in"));
        return;
      }
      const user_id = this.client.user.id;
      const voiceConn = new VoiceConnection(
        this,
        guild_id,
        user_id,
        channel_id,
        (conn) => {
          resolve(conn);
        },
      );
      this._voiceConnection = voiceConn;
      this._gatewayEmitter.on(
        "VOICE_STATE_UPDATE",
        (d: { user_id: string; session_id: string }) => {
          if (user_id !== d.user_id) return;
          voiceConn.setSession(d.session_id);
        },
      );
      this._gatewayEmitter.on(
        "VOICE_SERVER_UPDATE",
        (d: {
          guild_id: string | null;
          channel_id?: string;
          endpoint: string;
          token: string;
        }) => {
          if (guild_id !== d.guild_id) return;
          // channel_id is not set for guild voice calls
          if (d.channel_id && channel_id !== d.channel_id) return;
          voiceConn.setTokens(d.endpoint, d.token);
        },
      );
      this.signalVideo(false);
    });
  }

  /** Create a GoLive stream (screen share) on top of the voice connection. */
  createStream(): Promise<WebRtcConnWrapper> {
    return new Promise((resolve, reject) => {
      if (!this.client.user) {
        reject(new Error("Client not logged in"));
        return;
      }
      if (!this.voiceConnection) {
        reject(
          new Error("cannot start stream without first joining voice channel"),
        );
        return;
      }
      const {
        guildId: clientGuildId,
        channelId: clientChannelId,
        session_id,
      } = this.voiceConnection;
      const clientUserId = this.client.user.id;
      if (!session_id) throw new Error("Session doesn't exist yet");
      const streamConn = new StreamConnection(
        this,
        clientGuildId,
        clientUserId,
        clientChannelId,
        (conn) => {
          clearTimeout(streamTimeout);
          clearInterval(retryInterval);
          resolve(conn);
        },
      );
      this.voiceConnection.streamConnection = streamConn;

      // Attach listeners BEFORE the first signal so a fast dispatch can't
      // be lost between signalStream() and listener registration.
      const onStreamCreate = (d: {
        stream_key: string;
        rtc_server_id: string;
      }) => {
        const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
        if (
          clientGuildId !== guildId ||
          clientChannelId !== channelId ||
          clientUserId !== userId
        ) {
          return;
        }
        streamConn.serverId = d.rtc_server_id;
        streamConn.streamKey = d.stream_key;
        streamConn.setSession(session_id);
      };
      const onStreamServerUpdate = (d: {
        stream_key: string;
        endpoint: string;
        token: string;
      }) => {
        const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
        if (
          clientGuildId !== guildId ||
          clientChannelId !== channelId ||
          clientUserId !== userId
        ) {
          return;
        }
        streamConn.setTokens(d.endpoint, d.token);
      };
      this._gatewayEmitter.on("STREAM_CREATE", onStreamCreate);
      this._gatewayEmitter.on("STREAM_SERVER_UPDATE", onStreamServerUpdate);

      const cleanup = () => {
        clearTimeout(streamTimeout);
        clearInterval(retryInterval);
        this._gatewayEmitter.removeListener("STREAM_CREATE", onStreamCreate);
        this._gatewayEmitter.removeListener(
          "STREAM_SERVER_UPDATE",
          onStreamServerUpdate,
        );
      };
      const streamTimeout = setTimeout(() => {
        cleanup();
        reject(
          new Error(
            "Timed out waiting for STREAM_CREATE/STREAM_SERVER_UPDATE from Discord (stream handshake) — voice media session may not be active",
          ),
        );
      }, 12_000);

      // Discord sometimes drops the STREAM_CREATE request silently (upstream
      // issue #217/#219) — resend a few times instead of giving up after one.
      let attempt = 0;
      const retryInterval = setInterval(() => {
        attempt += 1;
        if (attempt >= 4) {
          clearInterval(retryInterval);
          return;
        }
        console.log(
          `[goLive:Streamer] createStream: retrying STREAM_CREATE (attempt ${attempt + 1}/4)`,
        );
        this.signalStream();
      }, 3_000);
      console.log(
        `[goLive:Streamer] createStream: sending STREAM_CREATE (attempt 1/4)`,
      );
      this.signalStream();
    });
  }

  async setStreamPreview(image: Buffer): Promise<void> {
    if (!this.client.token) throw new Error("Please login :)");
    if (!this.voiceConnection?.streamConnection?.guildId) return;
    const data = `data:image/jpeg;base64,${image.toString("base64")}`;
    const { guildId } = this.voiceConnection.streamConnection;
    if (!this.client.guilds) return;
    const server = await this.client.guilds.fetch(guildId);
    // biome-ignore lint/suspicious/noExplicitAny: discord.js-selfbot dynamic
    (server as any).members.me?.voice?.postPreview(data);
  }

  stopStream(): void {
    const stream = this.voiceConnection?.streamConnection;
    if (!stream) return;
    stream.stop();
    this.signalStopStream();
    this.voiceConnection.streamConnection = null;
    this._gatewayEmitter.removeAllListeners("STREAM_CREATE");
    this._gatewayEmitter.removeAllListeners("STREAM_SERVER_UPDATE");
  }

  leaveVoice(): void {
    this.voiceConnection?.stop();
    this.signalLeaveVoice();
    this._voiceConnection = null;
    this._gatewayEmitter.removeAllListeners("VOICE_STATE_UPDATE");
    this._gatewayEmitter.removeAllListeners("VOICE_SERVER_UPDATE");
  }

  signalVideo(video_enabled: boolean): void {
    if (!this.voiceConnection) return;
    const { guildId: guild_id, channelId: channel_id } = this.voiceConnection;
    this.sendOpcode(GatewayOpCodes.VOICE_STATE_UPDATE, {
      guild_id: guild_id,
      channel_id,
      self_mute: false,
      self_deaf: true,
      self_video: video_enabled,
    });
  }

  signalStream(): void {
    if (!this.voiceConnection) return;
    const {
      type,
      guildId: guild_id,
      channelId: channel_id,
      botId: user_id,
    } = this.voiceConnection;
    // Un-deafen before requesting the stream (mimic real client). Do NOT
    // set self_video: true — that flips on the bot's camera in Discord
    // (visible to everyone); screen share should not enable the camera.
    this.sendOpcode(GatewayOpCodes.VOICE_STATE_UPDATE, {
      guild_id,
      channel_id,
      self_mute: false,
      self_deaf: false,
      self_video: false,
    });
    this.sendOpcode(GatewayOpCodes.STREAM_CREATE, {
      type,
      guild_id,
      channel_id,
      preferred_region: null,
    });
    this.sendOpcode(GatewayOpCodes.STREAM_SET_PAUSED, {
      stream_key: generateStreamKey(type, guild_id, channel_id, user_id),
      paused: false,
    });
  }

  signalStopStream(): void {
    if (!this.voiceConnection) return;
    const {
      type,
      guildId: guild_id,
      channelId: channel_id,
      botId: user_id,
    } = this.voiceConnection;
    this.sendOpcode(GatewayOpCodes.STREAM_DELETE, {
      stream_key: generateStreamKey(type, guild_id, channel_id, user_id),
    });
  }

  signalLeaveVoice(): void {
    this.sendOpcode(GatewayOpCodes.VOICE_STATE_UPDATE, {
      guild_id: null,
      channel_id: null,
      self_mute: true,
      self_deaf: false,
      self_video: false,
    });
  }
}

export type { NativePeerConnection };