From a3e6c4695a703db390acabd9af96af207874a366 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Sun, 17 May 2026 17:35:37 +0700 Subject: [PATCH] chore: remove test environment configuration file --- .env.test | 30 ------ .gitignore | 3 +- src/index.ts | 8 ++ src/media/screenShareController.ts | 53 ++++++++++- src/media/ytdlp.ts | 2 +- src/streaming/index.ts | 142 +++++++++++++++++++++-------- src/streaming/transcoder.ts | 89 ++++++++++++++++++ src/webserver.ts | 23 +++++ 8 files changed, 278 insertions(+), 72 deletions(-) delete mode 100644 .env.test create mode 100644 src/streaming/transcoder.ts diff --git a/.env.test b/.env.test deleted file mode 100644 index 6dabe2d..0000000 --- a/.env.test +++ /dev/null @@ -1,30 +0,0 @@ -DISCORD_TOKEN=test_token_for_testing -RECORDINGS_DIR=./recordings -RECORDING_SEGMENT_MS=5000 -VERBOSE=false -DECODER_ROTATE_MS=5000 -DECODER_COOLDOWN_MS=30000 -AUDIO_STREAM_SILENCE_DURATION_MS=3000 -PACKET_FILTER_MIN_SIZE=8 -OPUS_FRAME_SIZE=960 -AUDIO_SAMPLE_RATE=48000 -AUDIO_CHANNELS=2 -AVATAR_SIZE=64 -WEBSERVER_PORT=3000 -VOICE_CONNECTION_TIMEOUT_MS=15000 -RECONNECT_TIMEOUT_MS=5000 -LOG_LEVEL=info -NODE_ENV=test -MONITOR_GUILD_ID=test_guild_id -PICSER_UPLOAD_URL=https://picser.asepharyana.tech/api/upload -ATTACHMENT_UPLOAD_TIMEOUT_MS=30000 -ATTACHMENT_MAX_SIZE_MB=100 -ATTACHMENT_RETRY_ATTEMPTS=3 -BACKLOG_SYNC_HOURS=24 -BACKLOG_SYNC_BATCH_SIZE=100 -AI_ANALYSIS_ENABLED=false -AI_LLM_API_KEY=test_key -AI_LLM_BASE_URL=https://9router.asepharyana.tech/v1 -AI_LLM_MODEL=free -AI_ANALYSIS_TIMEOUT_MS=30000 -DATABASE_TYPE=sqlite diff --git a/.gitignore b/.gitignore index 508d2fe..026ab7f 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ recordings dist/ public/app/ .muxer-queue.** -.claude/ \ No newline at end of file +.claude/ +.env.test \ No newline at end of file diff --git a/src/index.ts b/src/index.ts index a839bb7..b32e6c3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -73,6 +73,14 @@ async function initializeApp() { process.exit(1); } + client.on("debug", (msg) => { + if (msg.includes("[VOICE") || msg.includes("[ffmpeg") || msg.toLowerCase().includes("error") || msg.toLowerCase().includes("stream")) { + logger.info({ debugMsg: msg }, "Discord Client Debug"); + } else if (config.VERBOSE) { + logger.debug({ debugMsg: msg }, "Discord Client Debug"); + } + }); + client.on("ready", async () => { logger.info({ user: client.user?.tag }, "Bot logged in"); registerMessageCapture(client); diff --git a/src/media/screenShareController.ts b/src/media/screenShareController.ts index 257dad4..51d1e84 100644 --- a/src/media/screenShareController.ts +++ b/src/media/screenShareController.ts @@ -22,6 +22,9 @@ export interface ScreenShareControllerDependencies { getPlayerOwner?: () => DiscordPlayerOwner; getDirectVideoUrl?: (source: string) => Promise; streamer: Streamer; + useTranscoder?: boolean; + onBeforeStreamStart?: (guildId: string, channelId: string) => Promise | void; + onAfterStreamEnd?: (guildId: string, channelId: string) => Promise | void; onStreamStart?: () => void; onStreamEnd?: () => void; } @@ -44,6 +47,14 @@ export function createScreenShareController( async start(source: string): Promise { const status = dependencies.getVoiceStatus(); + let voiceReleased = false; + let voiceRestored = false; + + const restoreVoice = async () => { + if (voiceRestored || !voiceReleased || !guildId || !channelId) return; + voiceRestored = true; + await dependencies.onAfterStreamEnd?.(guildId, channelId); + }; if (active) { active.stop(); @@ -62,6 +73,9 @@ export function createScreenShareController( ); } + const guildId = status.activeGuildId; + const channelId = status.activeChannelId; + // If another media owner (e.g. music) holds the shared player, reject const owner = getPlayerOwner(); if (owner === "music") { @@ -70,15 +84,28 @@ export function createScreenShareController( try { const directUrl = await getDirectVideoUrl(source); + logger.info( + { + guildId, + channelId, + }, + "Creating screen share session", + ); + await dependencies.onBeforeStreamStart?.(guildId, channelId); + voiceReleased = true; const session = await dependencies.streamer.createSession( - status.activeGuildId, - status.activeChannelId, + guildId, + channelId, ); dependencies.onStreamStart?.(); let stopped = false; - const done = playPreparedStream(directUrl, session, { + const playFn = dependencies.useTranscoder + ? (await import("../streaming")).playTranscodedPreparedStream + : (await import("../streaming")).playPreparedStream; + + const done = playFn(directUrl, session, { fps: 30, bitrate: 2500, includeAudio: true, @@ -86,8 +113,16 @@ export function createScreenShareController( }).finally(() => { active = null; dependencies.onStreamEnd?.(); + return restoreVoice(); }); done.catch(() => undefined); + logger.info( + { + guildId, + channelId, + }, + "Screen share session started", + ); active = { done, @@ -96,11 +131,23 @@ export function createScreenShareController( stopped = true; session.stop(); active = null; + void restoreVoice(); }, }; return active; } catch (error) { active = null; + if (voiceReleased) { + await restoreVoice(); + } + logger.error( + { + error, + guildId, + channelId, + }, + "Screen share startup failed", + ); throw new AppError( error instanceof Error ? error.message : "Screen stream failed", "SCREEN_STREAM_FAILED", diff --git a/src/media/ytdlp.ts b/src/media/ytdlp.ts index b17cad7..cf5c360 100644 --- a/src/media/ytdlp.ts +++ b/src/media/ytdlp.ts @@ -61,7 +61,7 @@ export function createYtDlp(dependencies: YtDlpDependencies = {}): YtDlpClient { "--no-warnings", "--quiet", ]); - return value.trim().split("\n")[0] || url; + return value.trim(); }, }; } diff --git a/src/streaming/index.ts b/src/streaming/index.ts index 2cd3cb5..9db9ae9 100644 --- a/src/streaming/index.ts +++ b/src/streaming/index.ts @@ -1,7 +1,8 @@ -import { spawn } from "node:child_process"; import { EventEmitter } from "node:events"; import { PassThrough } from "node:stream"; import type { Readable } from "node:stream"; +import type { ChildProcess } from "node:child_process"; +import { prepareTranscoder, TranscoderOptions } from "./transcoder"; import type { Client } from "discord.js-selfbot-v13"; type VoiceConnectionLike = { @@ -53,11 +54,21 @@ export class Streamer { } async joinVoice(guildId: string, channelId: string): Promise { - const channel = (this.client.channels.resolve(channelId) ?? this.client.channels.cache.get(channelId)) as any; + const guild = this.client.guilds.cache.get(guildId); + const channel = (guild?.channels.cache.get(channelId) ?? + (await guild?.channels.fetch(channelId).catch(() => null))) as any; if (!channel || channel.guild?.id !== guildId) { throw new Error("VOICE_CHANNEL_NOT_FOUND"); } + const existingConnection = (this.client.voice as any).connection as + | VoiceConnectionLike + | undefined; + if (existingConnection?.channel?.id === channelId) { + (existingConnection as any).setVideoCodec?.("H264"); + return existingConnection; + } + const voiceConnection = (await this.client.voice.joinChannel(channel as any, { selfMute: true, selfDeaf: true, @@ -65,6 +76,8 @@ export class Streamer { videoCodec: "H264", })) as unknown as VoiceConnectionLike; + (voiceConnection as any).setVideoCodec?.("H264"); + return voiceConnection; } @@ -108,15 +121,62 @@ export class Streamer { connection, stream, async play(source: string | Readable, options: StreamPlayOptions = {}) { - const videoOptions = { + const videoOptions: Record = { fps: options.fps ?? 30, bitrate: options.bitrate ?? 2500, presetH26x: options.presetH26x ?? "superfast", }; - activeVideo = stream.playVideo(source, videoOptions); + const audioOptions: Record = { + volume: false, + }; + + let videoSource: string | Readable; + let audioSource: string | Readable; + + if (typeof source === "string" && source.includes("\n")) { + // yt-dlp returns multiple URLs (e.g., video\n audio\n) + const urls = source.split("\n").filter((u) => u.trim()); + videoSource = urls[0] ?? source; + audioSource = urls[1] ?? urls[0] ?? source; + } else if (typeof source !== "string") { + // If source is a Readable (e.g. ffmpeg stdout) and audio+video + // need to be played separately, tee the stream into two PassThroughs. + if (options.includeAudio !== false) { + const videoTee = new PassThrough(); + const audioTee = new PassThrough(); + // Pipe to both tees; allow consumers to read independently. + (source as Readable).pipe(videoTee); + (source as Readable).pipe(audioTee); + videoSource = videoTee; + audioSource = audioTee; + } else { + // audio excluded — single video stream + const videoTee = new PassThrough(); + (source as Readable).pipe(videoTee); + videoSource = videoTee; + audioSource = videoTee; + } + } else { + videoSource = source; + audioSource = source; + } + + const inputFFmpegArgs = [ + "-headers", + "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.3\r\nConnection: keep-alive\r\n", + ]; + + if (typeof videoSource === "string" && videoSource.startsWith("http")) { + videoOptions.inputFFmpegArgs = inputFFmpegArgs; + } + if (typeof audioSource === "string" && audioSource.startsWith("http")) { + audioOptions.inputFFmpegArgs = inputFFmpegArgs; + } + + activeVideo = stream.playVideo(videoSource, videoOptions); if (options.includeAudio !== false) { - activeAudio = stream.playAudio(source, { volume: false }); + activeAudio = stream.playAudio(audioSource, audioOptions); } try { @@ -131,40 +191,15 @@ export class Streamer { } export function prepareStream(source: string, _options: any): { - command: ReturnType | { kill?: (signal: NodeJS.Signals) => unknown }; + command: ChildProcess | { kill?: (signal: NodeJS.Signals) => unknown }; output: Readable; } { - // Spawn ffmpeg to transcode the source into a simple container with - // H264 video + Opus audio and pipe to stdout. Options are simplified and - // intentionally conservative to keep parity with prior behavior. - const args = [ - "-hide_banner", - "-loglevel", - "warning", - "-i", - source, - "-c:v", - "libx264", - "-preset", - "superfast", - "-r", - "30", - "-s", - "1280x720", - "-b:v", - "2500k", - "-maxrate", - "4000k", - "-c:a", - "libopus", - "-f", - "matroska", - "-", - ]; - - const command = spawn("ffmpeg", args, { stdio: ["ignore", "pipe", "pipe"] }); - const output = command.stdout ?? new PassThrough(); - + const opts: TranscoderOptions = { + fps: _options?.fps ?? 30, + bitrate: _options?.bitrate ?? "2500k", + preset: _options?.presetH26x ?? _options?.preset ?? "superfast", + }; + const { command, output } = prepareTranscoder(source, opts); return { command, output }; } @@ -197,5 +232,38 @@ export async function playPreparedStream( session: StreamSession, options: StreamPlayOptions = {}, ): Promise { + // Default behavior: forward resource (string or Readable) to session.play. + await session.play(source, options); +} + +export async function playTranscodedPreparedStream( + source: string | Readable, + session: StreamSession, + options: StreamPlayOptions = {}, +): Promise { + if (typeof source === "string" && /^(https?:)?\/\//.test(source)) { + const { command, output } = prepareStream(source, options); + const globalAny: any = globalThis; + const onData = (chunk: Buffer) => { + try { + globalAny.broadcastVideoToWeb?.(chunk); + } catch { + // ignore errors broadcasting + } + }; + output.on("data", onData); + try { + await session.play(output, options); + } finally { + output.off("data", onData); + try { + command.kill?.("SIGKILL"); + } catch (e) { + // ignore + } + } + return; + } + await session.play(source, options); } diff --git a/src/streaming/transcoder.ts b/src/streaming/transcoder.ts new file mode 100644 index 0000000..b005ca1 --- /dev/null +++ b/src/streaming/transcoder.ts @@ -0,0 +1,89 @@ +import { spawn, ChildProcess } from "node:child_process"; +import { PassThrough } from "node:stream"; +import type { Readable } from "node:stream"; +import { retryWithBackoff } from "../retry"; +import { createChildLogger } from "../logger"; + +const logger = createChildLogger("transcoder"); + +export interface TranscoderOptions { + fps?: number; + bitrate?: string | number; + preset?: string; +} + +export class Transcoder { + proc: ChildProcess | null = null; + output: Readable | null = null; + + constructor(private source: string, private opts: TranscoderOptions = {}) {} + + start(): { command: ChildProcess; output: Readable } { + const fps = this.opts.fps ?? 30; + const bitrate = String(this.opts.bitrate ?? "2500k"); + const preset = this.opts.preset ?? "superfast"; + + const args = [ + "-hide_banner", + "-loglevel", + "warning", + "-i", + this.source, + "-c:v", + "libx264", + "-preset", + preset, + "-r", + String(fps), + "-s", + "1280x720", + "-b:v", + String(bitrate), + "-maxrate", + "4000k", + "-c:a", + "libopus", + "-f", + "matroska", + "-", + ]; + + const cmd = spawn("ffmpeg", args, { stdio: ["ignore", "pipe", "pipe"] }); + const out = cmd.stdout ?? new PassThrough(); + + this.proc = cmd; + this.output = out; + + cmd.on("error", (err) => { + logger.error({ err }, "transcoder process error"); + }); + cmd.on("exit", (code, signal) => { + logger.info({ code, signal }, "transcoder exited"); + }); + + return { command: cmd, output: out }; + } + + stop(): void { + try { + if (this.proc && !this.proc.killed) this.proc.kill("SIGKILL"); + } catch (e) { + logger.warn({ e }, "failed to kill transcoder"); + } + this.proc = null; + this.output = null; + } + + async startWithRetry(retries = 2) { + return retryWithBackoff(() => Promise.resolve(this.start()), { + retries, + logger, + }); + } +} + +export function prepareTranscoder(source: string, options: TranscoderOptions = {}) { + const t = new Transcoder(source, options); + const { command, output } = t.start(); + return { transcoder: t, command, output }; +} diff --git a/src/webserver.ts b/src/webserver.ts index 736f496..c4a3249 100644 --- a/src/webserver.ts +++ b/src/webserver.ts @@ -44,6 +44,7 @@ const activeUsers = new Map< type VoiceGlobals = typeof globalThis & { moderationBroadcaster?: ModerationBroadcaster; broadcastPcmToWeb?: (chunk: Buffer, userId: string) => void; + broadcastVideoToWeb?: (chunk: Buffer) => void; updateActiveUser?: ( userId: string, data: { username: string; avatar: string; speaking: boolean }, @@ -211,6 +212,15 @@ export async function startWebserver( const screenController = createScreenShareController({ getVoiceStatus: () => voiceController.getStatus(), streamer, + useTranscoder: true, + onBeforeStreamStart: async () => { + await voiceController.disconnect(); + }, + onAfterStreamEnd: async (guildId: string, channelId: string) => { + const current = voiceController.getStatus(); + if (current.connected && current.activeGuildId === guildId) return; + await voiceController.connect(guildId, channelId); + }, }); const mediaController = new MediaController({ @@ -338,6 +348,19 @@ export async function startWebserver( } }; + // Outbound: server video stream (matroska chunks) -> browser clients + (globalThis as VoiceGlobals).broadcastVideoToWeb = (chunk: Buffer) => { + for (const client of broadcaster.getClients()) { + if (client.readyState === 1) { + try { + client.send(chunk); + } catch (err) { + wsLogger.warn({ err }, "Failed to send video chunk"); + } + } + } + }; + (globalThis as VoiceGlobals).updateActiveUser = ( userId: string, data: { username: string; avatar: string; speaking: boolean },