javis_bot/bot/src/stream/selfbot.ts

/**
 * Selfbot live-stream backend (default).
 *
 * Streams the VNC X display (:1) into the voice channel as a real Discord
 * "Go Live" broadcast. Discord blocks video from *bot* accounts, so this path
 * requires a USER account token (a "selfbot"), which violates Discord ToS and
 * can get the account banned. Use a throwaway/burner account, never your main.
 *
 * Dependencies are optional (native) and dynamically imported so the core bot
 * installs/runs without them:
 *   bun add discord.js-selfbot-v13 @dank074/discord-video-stream
 *   bun pm trust @dank074/node-av node-datachannel   # build native deps
 *
 * API targets @dank074/discord-video-stream v6 (verified against its d.ts):
 *   new Streamer(client) -> joinVoice(guildId, channelId)
 *   prepareStream(input, opts, signal) -> { command, output }
 *   playStream(output, streamer, { type: "go-live" }, signal)
 */
import { spawn, type ChildProcess } from "node:child_process";
import type { AppConfig } from "../config.ts";
import type { ScreenStreamer, StreamContext } from "./index.ts";
import { VncKeepalive, resolveVncPassword, vncPortForDisplay } from "./vnc-keepalive.ts";

export class SelfbotStreamer implements ScreenStreamer {
  readonly kind = "selfbot" as const;
  private streamer: any = null;
  private capture: ChildProcess | null = null;
  private keepalive: VncKeepalive | null = null;
  private helper: ChildProcess | null = null;
  private controller: AbortController | null = null;
  // `starting` is the in-flight lock (start() is mid-handshake); `live` is the
  // REAL Go-Live state — true only once Discord's stream WebRTC actually reaches
  // "connected". isActive() reports `live` so the brain's search routing and the
  // auto-start retry react to whether we are genuinely broadcasting, not to a
  // local "we kicked off ffmpeg" guess.
  private starting = false;
  private live = false;
  // Whether WE own the streamer's client (dedicated/separate account). When we
  // ride the conversation's shared session we must NOT destroy its client or
  // leave its voice on teardown — only stop the Go-Live stream.
  private ownsClient = false;

  /** Tear down the streamer's Discord side: full leave+destroy when we own the
   *  client, otherwise just stop the Go-Live stream (the conversation owns the
   *  client/voice). */
  private teardownStreamer(streamer: any): void {
    try {
      if (this.ownsClient) {
        streamer?.leaveVoice?.();
        streamer?.client?.destroy?.();
      } else {
        streamer?.stopStream?.();
      }
    } catch {
      /* ignore */
    }
  }

  constructor(private config: AppConfig) {}

  isActive() {
    return this.live;
  }

  /** How long to wait for the Go-Live stream WebRTC to reach "connected". */
  private get streamReadyTimeoutMs(): number {
    return (this.config as any).streamReadyTimeoutMs ?? 25_000;
  }

  /**
   * Poll the streaming library for the REAL Go-Live readiness signal: the
   * stream connection's WebRTC peer reaching "connected" state (set only after
   * STREAM_CREATE -> STREAM_SERVER_UPDATE -> protocol/DAVE handshake). Resolves
   * true when live, false on timeout or abort.
   */
  private async waitStreamReady(streamer: any, signal: AbortSignal, timeoutMs: number): Promise<boolean> {
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {
      if (signal.aborted) return false;
      if (streamer?.voiceConnection?.streamConnection?.webRtcConn?.ready === true) return true;
      await new Promise((r) => setTimeout(r, 250));
    }
    return false;
  }

  /** A compact snapshot of the connection state, logged when readiness times out. */
  private streamDiag(streamer: any): string {
    const vc = streamer?.voiceConnection;
    const sc = vc?.streamConnection;
    try {
      return JSON.stringify({
        voiceReady: vc?.webRtcConn?.ready ?? null,
        voiceDave: vc?.daveReady ?? null,
        streamConnCreated: !!sc,
        streamReady: sc?.webRtcConn?.ready ?? null,
        streamDave: sc?.daveReady ?? null,
      });
    } catch {
      return "{unavailable}";
    }
  }

  /**
   * Wait a randomised, human-plausible amount of time. Resolves immediately if
   * the stream is aborted (stop()) mid-wait, so teardown never hangs on a pause.
   */
  private humanPause(minMs: number, maxMs: number, signal?: AbortSignal): Promise<void> {
    const ms = Math.floor(minMs + Math.random() * Math.max(0, maxMs - minMs));
    return new Promise((resolve) => {
      if (signal?.aborted) return resolve();
      const onAbort = () => {
        clearTimeout(timer);
        resolve();
      };
      const timer = setTimeout(() => {
        signal?.removeEventListener("abort", onAbort);
        resolve();
      }, ms);
      signal?.addEventListener("abort", onAbort, { once: true });
    });
  }

  private async loadLib() {
    let selfbot: any, vs: any;
    try {
      selfbot = await import("discord.js-selfbot-v13");
      // Optional native dep; resolved at runtime only.
      // @ts-ignore - optional dependency, may be absent until `bun add`ed
      vs = await import("@dank074/discord-video-stream");
    } catch (e) {
      throw new Error(
        "셀프봇 송출 의존성이 없습니다. 설치: bun add discord.js-selfbot-v13 @dank074/discord-video-stream\n" +
          `원본 오류: ${(e as Error).message}`,
      );
    }
    if (!vs.Streamer || !vs.prepareStream || !vs.playStream) {
      throw new Error(
        "@dank074/discord-video-stream v6 API(Streamer/prepareStream/playStream)를 찾지 못했습니다. " +
          "package.json 버전을 ^6.0.0으로 맞추세요.",
      );
    }
    return { selfbot, vs };
  }

  async start(ctx: StreamContext): Promise<string> {
    if (this.live || this.starting) return "이미 송출 중이거나 시작 중입니다.";
    // Screen-share gate: STREAM_BROWSER=false means voice + API/MCP only, so we
    // never go Live. Enforced HERE (not just in the slash command) so every
    // caller - including stream-hold.ts - respects it.
    if (this.config.screenBrowser === false) {
      return "화면 공유(브라우저) 모드가 꺼져 있습니다 (STREAM_BROWSER=false). 음성 + API/MCP 모드로만 동작합니다.";
    }
    // Broadcast account: a dedicated DISCORD_STREAM_TOKEN if provided, otherwise
    // reuse the conversation's selfbot token. In userbot mode (no normal-bot
    // token) sharing one account fails — the broadcaster's voice connection
    // can't establish alongside the conversation's, so the Go-Live never
    // connects. Warn loudly in that case so the failure is self-explanatory.
    const streamToken = this.config.streamToken || this.config.selfbotToken;
    if (!streamToken) {
      return "DISCORD_SELFBOT_TOKEN(또는 DISCORD_STREAM_TOKEN)이 설정되지 않았습니다 (.env). 버너 계정 토큰을 넣어주세요.";
    }
    const dedicatedStreamAccount =
      !!this.config.streamToken && this.config.streamToken !== this.config.selfbotToken;
    // Single-account userbot: broadcast on the conversation's OWN session — the
    // Go-Live stream is a SEPARATE stream connection on the same session (exactly
    // like a real Discord client), so it never deafens or fights the conversation
    // voice. Only when there is neither a shared session nor a dedicated account
    // is the broadcast impossible: a second login would deafen the conversation,
    // so refuse to protect it.
    const shared = ctx.getSharedSession?.() ?? null;
    if (!shared && !dedicatedStreamAccount && !this.config.botToken) {
      console.warn(
        "[selfbot] ⚠️ 공유 세션도 방송 전용 토큰도 없어 방송을 시작하지 않습니다 (대화 음성 보호).",
      );
      return "방송을 시작할 수 없습니다 (대화 세션 공유 불가 + 전용 계정 없음).";
    }
    if (!ctx.voiceChannelId) {
      return "셀프봇 송출은 음성 채널 안에서 호출해야 합니다.";
    }
    // Lock the starting state BEFORE any await: the human-pause delays below
    // mean start() is in-flight for several seconds, so a second start() call
    // must be rejected by the guard above, and the status must read "starting"
    // rather than live during the wait. `live` is only set once the stream
    // WebRTC is actually connected (see the readiness wait below). Keep
    // controller / streamer / capture as LOCAL refs so an interleaved stop()
    // (which nulls the instance fields) can't turn our own continuation into a
    // null dereference.
    this.starting = true;
    const controller = (this.controller = new AbortController());
    const signal = controller.signal;
    let streamer: any = null;
    let capture: ChildProcess | null = null;
    let keepalive: VncKeepalive | null = null;
    let helper: ChildProcess | null = null;

    try {
      const { selfbot, vs } = await this.loadLib();
      const { Streamer, prepareStream, playStream } = vs;
      signal.throwIfAborted();

      if (shared) {
        // Ride the conversation's existing session: create the Go-Live STREAM
        // connection on it (playStream below calls createStream against this
        // voiceConnection) — no second login / voice join, which would deafen or
        // fight the conversation. type/botId/session_id are required by the
        // stream-create signalling.
        this.ownsClient = false;
        streamer = this.streamer = new Streamer(shared.client);
        (streamer as any)._voiceConnection = {
          guildId: shared.guildId,
          channelId: shared.channelId,
          session_id: shared.sessionId,
          type: "guild",
          botId: shared.botId,
        };
      } else {
        // Dedicated/separate account: log in and join voice ourselves. Act like
        // a person, not a bot: breathe after coming online before joining, then
        // settle before going Live. Randomised; throwIfAborted() after each pause
        // unwinds into the catch if stop() lands mid-wait.
        this.ownsClient = true;
        streamer = this.streamer = new Streamer(new selfbot.Client());
        await streamer.client.login(streamToken);
        signal.throwIfAborted();
        await this.humanPause(2500, 4500, signal);
        signal.throwIfAborted();
        await streamer.joinVoice(ctx.guildId, ctx.voiceChannelId);
        await this.humanPause(6000, 10000, signal);
        signal.throwIfAborted();
      }

      const [w, h] = this.config.vncResolution.split("x").map((n) => parseInt(n, 10));

      // Capture the VNC X display with the SYSTEM ffmpeg (which reliably has
      // x11grab), then pipe that stream into the library. Relying on the lib's
      // bundled libav for the x11grab input device is not portable; piping the
      // system ffmpeg is. (Verified live against a real voice channel.)
      //
      // The SYSTEM ffmpeg produces the final, Discord-ready H264 in one pass:
      // target bitrate (-b:v/-maxrate), no B-frames (WebRTC requires this), a
      // 1s keyframe interval, and yuv420p. The library then only REMUXES it
      // (noTranscoding below) so there is no second decode/scale/encode. With
      // streamHw on (default) this single encode runs on the GPU (h264_nvenc,
      // RTX 5050); otherwise it falls back to software x264.
      const hw = this.config.streamHw;
      const kbps = this.config.vncBitrateKbps;
      // The library advertises a hardcoded max_bitrate of 10 Mbps to Discord
      // (BaseMediaConnection: `max_bitrate: 10000 * 1000`). If the encoder bursts
      // above that negotiated ceiling, WebRTC congestion control drops packets
      // and the viewer sees stutter. Cap -maxrate at 10 Mbps to stay within it.
      const LIB_MAX_BITRATE_KBPS = 10000;
      const maxKbps = Math.min(Math.round(kbps * 1.5), LIB_MAX_BITRATE_KBPS);
      const captureCodecArgs = hw
        ? ["-c:v", "h264_nvenc", "-preset", "p4", "-tune", "ll", "-forced-idr", "1"]
        : ["-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency"];
      // Optionally pull desktop audio (the default sink's PipeWire/Pulse monitor)
      // so the broadcast has sound. We add it as a second input and mux AAC into
      // the mpegts; the library re-encodes it to Opus for Discord. ffmpeg needs
      // XDG_RUNTIME_DIR (inherited) to reach the pulse socket. -map is required
      // once there are two inputs.
      const audioOn = this.config.streamAudio;
      const audioInput = audioOn ? ["-f", "pulse", "-i", this.config.streamAudioSource] : [];
      const audioMap = audioOn ? ["-map", "0:v:0", "-map", "1:a:0"] : [];
      const audioCodec = audioOn ? ["-c:a", "aac", "-b:a", "160k", "-ar", "48000", "-ac", "2"] : [];
      capture = this.capture = spawn("ffmpeg", [
        "-loglevel", "error",
        "-thread_queue_size", "1024",
        "-f", "x11grab",
        "-framerate", String(this.config.vncFramerate),
        "-video_size", this.config.vncResolution,
        "-i", this.config.vncDisplay,
        ...(audioOn ? ["-thread_queue_size", "1024"] : []),
        ...audioInput,
        ...audioMap,
        ...captureCodecArgs,
        "-b:v", `${kbps}k`, "-maxrate", `${maxKbps}k`, "-bufsize", `${kbps}k`,
        "-bf", "0",
        "-pix_fmt", "yuv420p",
        "-g", String(this.config.vncFramerate),
        ...audioCodec,
        "-f", "mpegts", "pipe:1",
      ]);
      capture.stderr?.on("data", (d) => {
        if (!signal.aborted) console.error("[selfbot x11grab]", d.toString().trim());
      });

      // Keep a VNC client attached for the life of the stream. TigerVNC only
      // flushes its framebuffer at full rate while a client pulls updates; the
      // Discord broadcast reads that framebuffer with x11grab (not as a VNC
      // client), so without this the captured screen would idle at ~1.5 fps and
      // the stream would look badly choppy. Fail-open: a missing password just
      // skips it. Matched to the stream framerate so motion stays smooth.
      const vncPw = resolveVncPassword();
      if (vncPw) {
        keepalive = this.keepalive = new VncKeepalive({
          host: "127.0.0.1",
          port: vncPortForDisplay(this.config.vncDisplay),
          password: vncPw,
          fps: this.config.vncFramerate,
        });
        keepalive.start();
      }

      // Browser-side broadcast defaults (ad-skip, subtitle rule, fullscreen
      // toolbar hiding) run in a small CDP helper tied to the stream lifecycle.
      // It attaches to the on-screen Chrome (CDP_PORT) and fail-opens if Chrome
      // or its deps are absent, so it can never break the stream.
      try {
        const helperPath = new URL("../../scripts/stream-test/broadcast-helper.mjs", import.meta.url).pathname;
        helper = this.helper = spawn("node", [helperPath], {
          env: { ...process.env, CDP_PORT: process.env.CDP_PORT ?? "9222" },
          stdio: "ignore",
        });
        helper.on("error", () => {
          /* node/playwright missing: the stream runs without the helper */
        });
      } catch {
        /* ignore */
      }

      const { command, output } = prepareStream(
        capture.stdout,
        {
          // The capture above is already a Discord-ready H264 elementary stream,
          // so the library only remuxes it (no second encode). width/height/
          // frameRate are passed for signalling; encoding options are ignored
          // on the copy path.
          width: w || 1920,
          height: h || 1080,
          frameRate: this.config.vncFramerate,
          videoCodec: "H264",
          noTranscoding: true,
        },
        signal,
      );
      command.on("error", (err: Error) => {
        if (!signal.aborted) console.error("[selfbot] ffmpeg error:", err);
      });
      signal.throwIfAborted();

      playStream(output, streamer, { type: "go-live" }, signal)
        .catch((err: Error) => {
          if (!signal.aborted) console.error("[selfbot] playStream:", err);
        })
        .finally(() => {
          // The stream ended on its own (Discord closed the Go-Live, the voice
          // UDP dropped, or ffmpeg exited) rather than via stop(). If we are
          // still the current attempt, tear the pipeline DOWN: kill the capture
          // ffmpeg and leave voice. Otherwise the x11grab->nvenc encoder keeps
          // running forever feeding a pipe nobody reads, pinning a CPU core
          // while no media is actually transmitted. Skip if a concurrent
          // stop()/start() already replaced the controller (it owns teardown).
          if (this.controller !== controller) return;
          try {
            capture?.kill("SIGKILL");
          } catch {
            /* ignore */
          }
          try {
            keepalive?.stop();
          } catch {
            /* ignore */
          }
          try {
            helper?.kill();
          } catch {
            /* ignore */
          }
          this.teardownStreamer(streamer);
          if (this.capture === capture) this.capture = null;
          if (this.keepalive === keepalive) this.keepalive = null;
          if (this.helper === helper) this.helper = null;
          if (this.streamer === streamer) this.streamer = null;
          this.controller = null;
          this.live = false;
          this.starting = false;
        });

      // Wait for the REAL Go-Live signal before declaring success. playStream
      // above only kicks off the pipeline (it resolves when the stream ENDS, not
      // when it connects), so without this we'd report "live" while the WebRTC
      // handshake is still pending — or never completes. Poll the library's
      // stream connection until its WebRTC reaches "connected".
      const ready = await this.waitStreamReady(streamer, signal, this.streamReadyTimeoutMs);
      if (!ready) {
        // Not actually broadcasting: log why and tear the pipeline down NOW so a
        // dead x11grab/NVENC encoder isn't left pinning the GPU/CPU. We abort and
        // do the teardown with our LOCAL refs here rather than relying on the
        // playStream .finally (it early-returns once we null this.controller).
        console.error(
          `[selfbot] ⚠️ Go-Live did NOT connect within ${this.streamReadyTimeoutMs}ms — ` +
            `not broadcasting. state=${this.streamDiag(streamer)}`,
        );
        controller.abort();
        try { capture?.kill("SIGKILL"); } catch { /* ignore */ }
        try { keepalive?.stop(); } catch { /* ignore */ }
        try { helper?.kill(); } catch { /* ignore */ }
        this.teardownStreamer(streamer);
        if (this.controller === controller) {
          if (this.capture === capture) this.capture = null;
          if (this.keepalive === keepalive) this.keepalive = null;
          if (this.helper === helper) this.helper = null;
          if (this.streamer === streamer) this.streamer = null;
          this.controller = null;
          this.live = false;
          this.starting = false;
        }
        return `방송을 시작하지 못했습니다 (Go-Live 연결 타임아웃 ${this.streamReadyTimeoutMs}ms). 동일 계정 음성 충돌 또는 송출 경로 문제로 보입니다.`;
      }
      this.live = true;
      this.starting = false;
      console.log("🔴 [selfbot] Go-Live WebRTC connected — broadcasting for real.");
      return "🔴 셀프봇으로 VNC 화면을 음성채널에 실시간 송출 중입니다 (Go Live).";
    } catch (e) {
      // Startup was aborted (stop() during a pause) or failed. Tear down using
      // our LOCAL refs, then clear instance state only if it still points at us
      // (a concurrent stop()/start() may already have replaced it).
      try {
        capture?.kill("SIGKILL");
      } catch {
        /* ignore */
      }
      try {
        keepalive?.stop();
      } catch {
        /* ignore */
      }
      try {
        helper?.kill();
      } catch {
        /* ignore */
      }
      this.teardownStreamer(streamer);
      // Only release the lock / clear instance state if WE are still the
      // current attempt. If a concurrent stop()+start() already replaced the
      // controller, a newer start() owns the lock — clearing it here would
      // unlock it mid-startup and let a third start() race in.
      if (this.controller === controller) {
        if (this.capture === capture) this.capture = null;
        if (this.keepalive === keepalive) this.keepalive = null;
        if (this.helper === helper) this.helper = null;
        if (this.streamer === streamer) this.streamer = null;
        this.controller = null;
        this.live = false;
        this.starting = false;
      }
      if (signal.aborted) return "송출을 시작하는 중에 중지했습니다.";
      throw e;
    }
  }

  async stop(): Promise<void> {
    this.controller?.abort();
    this.controller = null;
    try {
      this.capture?.kill("SIGKILL");
    } catch {
      /* ignore */
    }
    this.capture = null;
    try {
      this.keepalive?.stop();
    } catch {
      /* ignore */
    }
    this.keepalive = null;
    try {
      this.helper?.kill();
    } catch {
      /* ignore */
    }
    this.helper = null;
    this.teardownStreamer(this.streamer);
    this.streamer = null;
    this.live = false;
    this.starting = false;
  }
}