MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(gateway): remove screen-share / GoLive feature entirely

Browse Source 
Drop the Discord Go Live (screen share) stack across the discord-gateway:
- delete src/goLive/ (19 modules: Streamer, Demuxer, encoders, WebRTC wrapper, native loader, etc.)
- delete native/libdatachannel-min/ N-API binding + flake native build + LD_LIBRARY_PATH wiring
- delete screenShareController.ts and screen-share tests (goLive-port, golive-*, demuxerNut, screenShareInput)
- mediaSource.ts: remove Invidious helpers + downloadScreenInput (YouTube full-file download)
- mediaTypes.ts: drop ScreenShare* types, narrow MediaMode to 'music' and DiscordPlayerOwner to non-screen
- media.handler.ts: remove screen branch, screenController/screenPlayback, voice-disconnect/reconnect accessor
- commandHandler.ts: stop passing getVoiceStatus / setVoiceController into MediaHandler
- media handler now only handles music; music queue/playback/status untouched

Verification: tsc --noEmit clean, biome clean on touched files, no lingering goLive/screenShare refs in BE/FE/gateway.
This commit is contained in:
asepharyana
2026-08-15 21:20:20 +07:00
parent 9ae26b8ec9
commit 0164444dd7
42 changed files with 14 additions and 5957 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/discord-gateway/src/goLive/AnnexBBitstreamReaderWriter.ts
-155
View File
@@ -1,155 +0,0 @@
/**
* AnnexB bitstream reader/writer (RBSP + emulation prevention) — ported
* from @dank074/discord-video-stream AnnexBBitstreamReaderWriter.js.
*/
export class AnnexBBitstreamReader {
private _buffer: Uint8Array;
private _byteOffset = 0;
private _bitOffset = 0;
constructor(buffer: Uint8Array) {
this._buffer = buffer;
}
readBits(count: number): number {
if (count === 0) return 0;
let result = 0;
while (count > 0) {
if (this._byteOffset >= this._buffer.length) {
throw new Error("Bad byte offset");
}
if (
this._bitOffset === 0 &&
this._byteOffset >= 2 &&
this._buffer[this._byteOffset - 2] === 0 &&
this._buffer[this._byteOffset - 1] === 0 &&
this._buffer[this._byteOffset] === 3
) {
// Skip over emulation prevention
this._byteOffset++;
}
if (this._bitOffset === 0 && count >= 8) {
result = (result << 8) | this._buffer[this._byteOffset++];
count -= 8;
} else {
const numBitsToRead = Math.min(count, 8 - this._bitOffset);
const mask = (1 << numBitsToRead) - 1;
const newBits =
(this._buffer[this._byteOffset] >>
(8 - this._bitOffset - numBitsToRead)) &
mask;
result = (result << numBitsToRead) | newBits;
count -= numBitsToRead;
this._bitOffset += numBitsToRead;
if (this._bitOffset === 8) {
this._bitOffset = 0;
this._byteOffset++;
}
}
}
return result;
}
readUnsigned(bits: number): number {
return this.readBits(bits);
}
readSigned(bits: number): number {
const unsigned = this.readUnsigned(bits);
if (unsigned & (1 << (bits - 1))) return unsigned - (1 << bits);
return unsigned;
}
readUnsignedExpGolomb(): number {
let leading0 = 0;
while (this.readBits(1) === 0) leading0++;
return (1 << leading0) + this.readBits(leading0) - 1;
}
readSignedExpGolomb(): number {
const unsigned = this.readUnsignedExpGolomb();
if (unsigned % 2 === 0) return unsigned / -2;
return (unsigned + 1) / 2;
}
}
export class AnnexBBitstreamWriter {
private _arr: number[] = [];
private _pendingByte = 0;
private _bitOffset = 0;
toBuffer(): Buffer {
return Buffer.from(this._arr);
}
flush(): void {
// Emulation prevention: insert 0x03 before 00 00
if (
this._pendingByte <= 3 &&
this._arr[this._arr.length - 1] === 0 &&
this._arr[this._arr.length - 2] === 0
) {
this._arr.push(3);
}
this._arr.push(this._pendingByte);
this._pendingByte = 0;
this._bitOffset = 0;
}
writeBits(bits: number, count: number): void {
while (count > 0) {
if (this._bitOffset === 0) {
if (count >= 8) {
this._pendingByte = (bits >> (count - 8)) & 0xff;
count -= 8;
this.flush();
} else {
const mask = (1 << count) - 1;
this._pendingByte |= (bits & mask) << (8 - count);
this._bitOffset = count;
count = 0;
}
} else {
const numBitsToWrite = Math.min(8 - this._bitOffset, count);
const bitsToWrite =
(bits >> (count - numBitsToWrite)) & ((1 << numBitsToWrite) - 1);
this._pendingByte |=
bitsToWrite << (8 - this._bitOffset - numBitsToWrite);
count -= numBitsToWrite;
this._bitOffset += numBitsToWrite;
if (this._bitOffset === 8) {
this._bitOffset = 0;
this.flush();
}
}
}
}
writeUnsigned(num: number, count: number): void {
if (num < 0) throw new Error("Expected a non-negative number");
this.writeBits(num, count);
}
writeSigned(num: number, count: number): void {
if (count <= 0) return;
if (count > 32) throw new Error("writeSigned supports up to 32 bits");
const mask =
count === 32 ? 0xffffffff >>> 0 : (((1 << count) >>> 0) - 1) >>> 0;
const unsigned = (num & mask) >>> 0;
this.writeBits(unsigned, count);
}
writeUnsignedExpGolomb(num: number): void {
if (num < 0) throw new Error("Expected a non-negative number");
num++;
const bitCount = 32 - Math.clz32(num >>> 0);
this.writeBits(0, bitCount - 1);
this.writeBits(num, bitCount);
}
writeSignedExpGolomb(num: number): void {
if (num < 0) this.writeUnsignedExpGolomb(-2 * num);
else this.writeUnsignedExpGolomb(2 * num - 1);
}
}
services/discord-gateway/src/goLive/AnnexBHelper.ts
-112
View File
@@ -1,112 +0,0 @@
/**
* H264/H265 NAL helpers — ported from @dank074/discord-video-stream
* AnnexBHelper.js. Only the H264 parts are used by GoLive (H264 encoder),
* H265 constants kept for completeness of the port.
*/
export enum H264NalUnitTypes {
Unspecified = 0,
CodedSliceNonIDR = 1,
CodedSlicePartitionA = 2,
CodedSlicePartitionB = 3,
CodedSlicePartitionC = 4,
CodedSliceIdr = 5,
SEI = 6,
SPS = 7,
PPS = 8,
AccessUnitDelimiter = 9,
EndOfSequence = 10,
EndOfStream = 11,
FillerData = 12,
SEIExtenstion = 13,
PrefixNalUnit = 14,
SubsetSPS = 15,
}
export enum H265NalUnitTypes {
TRAIL_N = 0,
TRAIL_R = 1,
TSA_N = 2,
TSA_R = 3,
STSA_N = 4,
STSA_R = 5,
RADL_N = 6,
RADL_R = 7,
RASL_N = 8,
RASL_R = 9,
RSV_VCL_N10 = 10,
RSV_VCL_R11 = 11,
RSV_VCL_N12 = 12,
RSV_VCL_R13 = 13,
RSV_VCL_N14 = 14,
RSV_VCL_R15 = 15,
BLA_W_LP = 16,
BLA_W_RADL = 17,
BLA_N_LP = 18,
IDR_W_RADL = 19,
IDR_N_LP = 20,
CRA_NUT = 21,
RSV_IRAP_VCL22 = 22,
RSV_IRAP_VCL23 = 23,
RSV_VCL24 = 24,
RSV_VCL25 = 25,
RSV_VCL26 = 26,
RSV_VCL27 = 27,
RSV_VCL28 = 28,
RSV_VCL29 = 29,
RSV_VCL30 = 30,
RSV_VCL31 = 31,
VPS_NUT = 32,
SPS_NUT = 33,
PPS_NUT = 34,
AUD_NUT = 35,
EOS_NUT = 36,
EOB_NUT = 37,
FD_NUT = 38,
PREFIX_SEI_NUT = 39,
SUFFIX_SEI_NUT = 40,
}
export const H264Helpers = {
getUnitType(frame: Uint8Array): number {
return frame[0] & 0x1f;
},
splitHeader(frame: Uint8Array): [Uint8Array, Uint8Array] {
return [frame.subarray(0, 1), frame.subarray(1)];
},
isAUD(unitType: number): boolean {
return unitType === H264NalUnitTypes.AccessUnitDelimiter;
},
};
export const H265Helpers = {
getUnitType(frame: Uint8Array): number {
return (frame[0] >> 1) & 0x3f;
},
splitHeader(frame: Uint8Array): [Uint8Array, Uint8Array] {
return [frame.subarray(0, 2), frame.subarray(2)];
},
isAUD(unitType: number): boolean {
return unitType === H265NalUnitTypes.AUD_NUT;
},
};
export const startCode3 = Buffer.from([0, 0, 1]);
/** Split an AnnexB bitstream into NAL units (start codes stripped). */
export function splitNalu(buf: Buffer): Buffer[] {
let temp: Buffer | null = buf;
const nalus: Buffer[] = [];
while (temp?.byteLength) {
let pos: number = temp.indexOf(startCode3);
let length = 3;
if (pos > 0 && temp[pos - 1] === 0) {
pos--;
length++;
}
const nalu = pos === -1 ? temp : temp.subarray(0, pos);
temp = pos === -1 ? null : temp.subarray(pos + length);
if (nalu.byteLength) nalus.push(nalu);
}
return nalus;
}
services/discord-gateway/src/goLive/AudioStream.ts
-20
View File
@@ -1,20 +0,0 @@
/**
* AudioStream — feeds encoded opus frames into the WebRTC connection.
* Ported from @dank074/discord-video-stream AudioStream.js.
*/
import { BaseMediaStream } from "./BaseMediaStream.js";
import type { WebRtcConnWrapper } from "./WebRtcWrapper.js";
export class AudioStream extends BaseMediaStream {
_conn: WebRtcConnWrapper;
constructor(conn: WebRtcConnWrapper, noSleep = false) {
super("audio", noSleep);
this._conn = conn;
}
async _sendFrame(frame: Buffer, frametime: number): Promise<void> {
this._conn.sendAudioFrame(frame, frametime);
}
}
services/discord-gateway/src/goLive/BaseMediaConnection.ts
-674
View File
@@ -1,674 +0,0 @@
/**
* Base media connection for Discord GoLive — ported from
* @dank074/discord-video-stream BaseMediaConnection.js.
*
* Owns the voice WebSocket (identify/select_protocol/heartbeat/resume),
* SDP negotiation against Discord's media server, DAVE E2E voice
* (via @snazzah/davey), and speaking/video attribute signaling.
*/
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import Davey from "@snazzah/davey";
import { CodecPayloadType } from "./CodecPayloadType.js";
import type { NativePeerConnection } from "./native.js";
import { isNativeAvailable } from "./native.js";
import { STREAMS_SIMULCAST } from "./utils.js";
import { VoiceOpCodes, VoiceOpCodesBinary } from "./VoiceOpCodes.js";
import { WebRtcConnWrapper } from "./WebRtcWrapper.js";
export interface MediaConnectionStatus {
hasSession: boolean;
hasToken: boolean;
started: boolean;
resuming: boolean;
}
export interface VideoAttribute {
fps: number;
width: number;
height: number;
}
export interface StreamerLike {
opts: Record<string, unknown>;
}
export class BaseMediaConnection extends EventEmitter {
interval: ReturnType<typeof setInterval> | null = null;
guildId: string | null = null;
channelId: string;
botId: string;
ws: WebSocket | null = null;
status: MediaConnectionStatus;
server: string | null = null; // websocket url
token: string | null = null;
session_id: string | null = null;
protected _webRtcWrapper: WebRtcConnWrapper;
_webRtcParams: {
address: string;
port: number;
audioSsrc: number;
videoSsrc: number;
rtxSsrc: number;
supportedEncryptionModes: string[];
} | null = null;
protected _closed = false;
ready: ((conn: WebRtcConnWrapper) => void) | null;
protected _streamer: StreamerLike;
protected _sequenceNumber = -1;
protected _daveSession: Davey.DAVESession | null = null;
protected _connectedUsers = new Set<string>();
protected _daveProtocolVersion = 0;
protected _davePendingTransitions = new Map<number, number>();
protected _daveDowngraded = false;
constructor(
streamer: StreamerLike,
guildId: string | null,
botId: string,
channelId: string,
callback: ((conn: WebRtcConnWrapper) => void) | null,
) {
super();
this._streamer = streamer;
this.status = {
hasSession: false,
hasToken: false,
started: false,
resuming: false,
};
this.guildId = guildId;
this.channelId = channelId;
this.botId = botId;
this.ready = callback;
this._webRtcWrapper = new WebRtcConnWrapper(this);
}
get type(): "guild" | "call" {
return this.guildId ? "guild" : "call";
}
get webRtcConn(): WebRtcConnWrapper {
return this._webRtcWrapper;
}
get webRtcParams(): BaseMediaConnection["_webRtcParams"] {
return this._webRtcParams;
}
get streamer(): StreamerLike {
return this._streamer;
}
/** daveChannelId — overridden in VoiceConnection (channelId) and StreamConnection (serverId - 1n). */
get daveChannelId(): string {
throw new Error("daveChannelId not implemented");
}
stop(): void {
this._closed = true;
this._webRtcWrapper.close();
this.ws?.close();
}
setSession(session_id: string): void {
this.session_id = session_id;
this.status.hasSession = true;
this.start();
}
setTokens(server: string, token: string): void {
this.token = token;
this.server = server;
this.status.hasToken = true;
this.start();
}
start(): void {
if (this.status.hasSession && this.status.hasToken) {
if (this.status.started) return;
this.status.started = true;
this.ws = new WebSocket(`wss://${this.server}/?v=8`);
this.ws.binaryType = "arraybuffer";
this.ws.addEventListener("open", () => {
if (this.status.resuming) {
this.status.resuming = false;
this.resume();
} else {
this.identify();
}
});
this.ws.addEventListener("error", (err) => {
console.error(err);
});
this.ws.addEventListener("close", (e) => {
const wasStarted = this.status.started;
this.interval && clearInterval(this.interval);
this.status.started = false;
const canResume = e.code === 4015 || e.code < 4000;
if (canResume && wasStarted) {
this.status.resuming = true;
this.start();
} else {
this._closed = true;
this._webRtcWrapper?.close();
}
});
this.setupEvents();
}
}
handleReady(d: {
ip: string;
port: number;
ssrc: number;
streams: { ssrc: number; rtx_ssrc: number }[];
modes: string[];
}): void {
// we hardcoded STREAMS_SIMULCAST, which will always be array of 1
const stream = d.streams[0];
console.log(
`[goLive:${this.constructor.name}] READY ssrc=${d.ssrc} ip=${d.ip} port=${d.port} streams=${JSON.stringify(d.streams)}`,
);
this._webRtcParams = {
address: d.ip,
port: d.port,
audioSsrc: d.ssrc,
videoSsrc: stream.ssrc,
rtxSsrc: stream.rtx_ssrc,
supportedEncryptionModes: d.modes,
};
}
async handleProtocolAck(d: {
sdp?: string;
dave_protocol_version?: number;
}): Promise<void> {
if (!("sdp" in d)) throw new Error("Only WebRTC connections are allowed");
// DEBUG: dump Discord's real answer SDP — which payload types did it select?
console.log(
`[goLive:${this.constructor.name}] DISCORD_ANSWER_SDP ${JSON.stringify(d.sdp ?? "").slice(0, 900)}`,
);
this._daveProtocolVersion = d.dave_protocol_version ?? 0;
this.initDave();
// Discord's SDP is garbage — generate our own from its pieces
let ip = "";
let port = "";
let iceUsername = "";
let icePassword = "";
let fingerprint = "";
let candidate = "";
for (const line of (d.sdp ?? "").split("\n")) {
if (line.startsWith("c=")) ip = line;
else if (line.startsWith("a=rtcp")) port = line.split(":")[1];
else if (line.startsWith("a=ice-ufrag")) iceUsername = line;
else if (line.startsWith("a=ice-pwd")) icePassword = line;
else if (line.startsWith("a=fingerprint")) fingerprint = line;
else if (line.startsWith("a=candidate")) candidate = line;
}
const audioPayloadType = CodecPayloadType.opus.payload_type;
const audioSection = `
m=audio ${port} UDP/TLS/RTP/SAVPF ${audioPayloadType}
${ip}
a=extmap:1 urn:ietf:params:rtp-hdrext:ssrc-audio-level
a=extmap:3 http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01
a=setup:passive
a=mid:0
a=maxptime:60
a=inactive
${iceUsername}
${icePassword}
${fingerprint}
${candidate}
a=rtcp-mux
a=rtpmap:${audioPayloadType} opus/48000/2
a=fmtp:${audioPayloadType} minptime=10;useinbandfec=1;usedtx=1
a=rtcp-fb:${audioPayloadType} transport-cc
a=rtcp-fb:${audioPayloadType} nack
a=ice-lite
`.trim();
const videoPayloads = Object.values(CodecPayloadType).filter(
(el) => el.type === "video",
);
const videoPayloadTypes = videoPayloads.flatMap((el) => [
el.payload_type,
el.rtx_payload_type ?? 0,
]);
const videoSection = `
m=video ${port} UDP/TLS/RTP/SAVPF ${videoPayloadTypes.join(" ")}
${ip}
a=extmap:2 http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time
a=extmap:3 http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01
a=extmap:14 urn:ietf:params:rtp-hdrext:toffset
a=extmap:13 urn:3gpp:video-orientation
a=extmap:5 http://www.webrtc.org/experiments/rtp-hdrext/playout-delay
a=setup:passive
a=mid:1
a=inactive
${iceUsername}
${icePassword}
${fingerprint}
${candidate}
a=rtcp-mux
a=ice-lite
`.trim();
const videoRtpMap = videoPayloads
.flatMap((el) => [
`a=rtpmap:${el.payload_type} ${el.name}/90000`,
`a=rtpmap:${el.rtx_payload_type} rtx/90000`,
`a=fmtp:${el.rtx_payload_type} apt=${el.payload_type}`,
// CRITICAL: H264 MUST advertise packetization-mode=1. The encoder
// emits baseline slices up to 8KB (>RTP MTU), so the packetizer
// fragments them into FU-A units (RFC 6184). Discord's receiver only
// reassembles FU-A when packetization-mode=1 is negotiated — without
// it the slices (type 28) are DROPPED while SPS/PPS (small single
// NALs) and Opus audio (no fragmentation) still arrive → black video
// with working audio. profile-level-id=42e01f (constrained baseline
// 3.1) matches the -profile:v baseline encoder + SPS VUI rewriter.
...(el.name === "H264"
? [
`a=fmtp:${el.payload_type} level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f`,
]
: []),
`a=rtcp-fb:${el.payload_type} ccm fir`,
`a=rtcp-fb:${el.payload_type} nack`,
`a=rtcp-fb:${el.payload_type} nack pli`,
`a=rtcp-fb:${el.payload_type} goog-remb`,
`a=rtcp-fb:${el.payload_type} transport-cc`,
])
.join("\n");
const builtAnswer = [audioSection, videoSection, videoRtpMap].join("\n");
this._webRtcWrapper.webRtcConn?.setRemoteDescription(builtAnswer, "answer");
console.log(
`[goLive:${this.constructor.name}] SELECT_PROTOCOL_ACK processed — remote answer set (${builtAnswer.length}B) video_mline=${videoPayloadTypes.join(" ")}`,
);
this.emit("select_protocol_ack");
}
initDave(): void {
if (this._daveProtocolVersion) {
if (this._daveSession) {
this._daveSession.reinit(
this._daveProtocolVersion,
this.botId,
this.daveChannelId,
);
} else {
this._daveSession = new Davey.DAVESession(
this._daveProtocolVersion,
this.botId,
this.daveChannelId,
);
}
this.sendOpcodeBinary(
VoiceOpCodesBinary.MLS_KEY_PACKAGE,
this._daveSession.getSerializedKeyPackage(),
);
} else if (this._daveSession) {
this._daveSession.reset();
this._daveSession.setPassthroughMode(true, 10);
}
}
processInvalidCommit(transitionId: number): void {
this.sendOpcode(VoiceOpCodes.MLS_INVALID_COMMIT_WELCOME, {
transition_id: transitionId,
});
this.initDave();
}
executePendingTransition(transitionId: number): void {
const newVersion = this._davePendingTransitions.get(transitionId);
if (newVersion === undefined) {
console.error("Unrecognized transition ID", { transitionId });
return;
}
const oldVersion = this._daveProtocolVersion;
this._daveProtocolVersion = newVersion;
if (oldVersion !== newVersion && newVersion === 0) {
// Downgraded
this._daveDowngraded = true;
} else if (transitionId > 0 && this._daveDowngraded) {
this._daveDowngraded = false;
this._daveSession?.setPassthroughMode(true, 10);
}
this._davePendingTransitions.delete(transitionId);
}
setupEvents(): void {
this.ws?.addEventListener("message", async (e) => {
if (e.data instanceof ArrayBuffer) {
this.handleBinaryMessages(Buffer.from(e.data));
return;
}
const { op, d, seq } = JSON.parse(e.data as string) as {
op: number;
// biome-ignore lint/suspicious/noExplicitAny: Discord voice WS payload is dynamically typed
d: any;
seq?: number;
};
if (seq) this._sequenceNumber = seq;
if (op === VoiceOpCodes.READY) {
this.handleReady(d);
this.setProtocols()
.then(() => this.ready?.(this._webRtcWrapper))
.catch((err: unknown) => {
// PC can be closed while setProtocols is in flight (stream
// teardown) — don't let that become an unhandledRejection.
console.log(
`[goLive:${this.constructor.name}] setProtocols rejected during teardown: ${err instanceof Error ? err.message : String(err)}`,
);
});
this.setVideoAttributes(false);
} else if (op >= 4000) {
console.error(`${this.constructor.name} connection error`, d);
} else if (op === VoiceOpCodes.HELLO) {
this.setupHeartbeat(d.heartbeat_interval);
} else if (op === VoiceOpCodes.SELECT_PROTOCOL_ACK) {
await this.handleProtocolAck(d);
} else if (op === VoiceOpCodes.SPEAKING) {
// ignore speaking updates
} else if (op === VoiceOpCodes.HEARTBEAT_ACK) {
// ignore heartbeat acknowledgements
} else if (op === VoiceOpCodes.RESUMED) {
this.status.started = true;
} else if (op === VoiceOpCodes.CLIENTS_CONNECT) {
d.user_ids.forEach((id: string) => {
this._connectedUsers.add(id);
});
} else if (op === VoiceOpCodes.CLIENT_DISCONNECT) {
this._connectedUsers.delete(d.user_id);
} else if (op === VoiceOpCodes.DAVE_PREPARE_TRANSITION) {
this._davePendingTransitions.set(d.transition_id, d.protocol_version);
if (d.transition_id === 0) {
this.executePendingTransition(d.transition_id);
} else {
if (d.protocol_version === 0) {
this._daveSession?.setPassthroughMode(true, 120);
}
this.sendOpcode(VoiceOpCodes.DAVE_TRANSITION_READY, {
transition_id: d.transition_id,
});
}
} else if (op === VoiceOpCodes.DAVE_EXECUTE_TRANSITION) {
this.executePendingTransition(d.transition_id);
} else if (op === VoiceOpCodes.DAVE_PREPARE_EPOCH) {
if (d.epoch === 1) {
this._daveProtocolVersion = d.protocol_version;
this.initDave();
}
}
});
}
handleBinaryMessages(msg: Buffer): void {
this._sequenceNumber = msg.readUint16BE(0);
const op = msg.readUint8(2);
switch (op) {
case VoiceOpCodesBinary.MLS_EXTERNAL_SENDER: {
this._daveSession?.setExternalSender(msg.subarray(3));
break;
}
case VoiceOpCodesBinary.MLS_PROPOSALS: {
const optype = msg.readUint8(3);
if (!this._daveSession) break;
const { commit, welcome } = this._daveSession.processProposals(
optype,
msg.subarray(4),
[...this._connectedUsers],
);
if (commit) {
this.sendOpcodeBinary(
VoiceOpCodesBinary.MLS_COMMIT_WELCOME,
welcome ? Buffer.concat([commit, welcome]) : commit,
);
}
break;
}
case VoiceOpCodesBinary.MLS_ANNOUNCE_COMMIT_TRANSITION: {
const transitionId = msg.readUInt16BE(3);
try {
this._daveSession?.processCommit(msg.subarray(5));
if (transitionId) {
this._davePendingTransitions.set(
transitionId,
this._daveProtocolVersion,
);
this.sendOpcode(VoiceOpCodes.DAVE_TRANSITION_READY, {
transition_id: transitionId,
});
}
} catch (e) {
console.debug("MLS commit errored", e);
this.processInvalidCommit(transitionId);
}
break;
}
case VoiceOpCodesBinary.MLS_WELCOME: {
const transitionId = msg.readUInt16BE(3);
try {
this._daveSession?.processWelcome(msg.subarray(5));
if (transitionId) {
this._davePendingTransitions.set(
transitionId,
this._daveProtocolVersion,
);
this.sendOpcode(VoiceOpCodes.DAVE_TRANSITION_READY, {
transition_id: transitionId,
});
}
} catch (e) {
console.debug("MLS welcome errored", e);
this.processInvalidCommit(transitionId);
}
break;
}
}
}
get daveReady(): boolean {
return !!this._daveProtocolVersion && !!this._daveSession?.ready;
}
get daveSession(): Davey.DAVESession | null {
return this._daveSession;
}
setupHeartbeat(interval: number): void {
if (this.interval) {
clearInterval(this.interval);
}
this.interval = setInterval(() => {
try {
this.sendOpcode(VoiceOpCodes.HEARTBEAT, {
t: Date.now(),
seq_ack: this._sequenceNumber,
});
} catch {
/* ignore */
}
}, interval);
}
sendOpcode(code: number, data: unknown): void {
if (this.ws?.readyState !== WebSocket.OPEN) return;
this.ws.send(JSON.stringify({ op: code, d: data }));
}
sendOpcodeBinary(code: number, data: Uint8Array): void {
if (this.ws?.readyState !== WebSocket.OPEN) return;
const buf = Buffer.allocUnsafe(data.length + 1);
buf.writeUInt8(code);
Buffer.from(data).copy(buf, 1);
this.ws.send(buf);
}
/** serverId — overridden in VoiceConnection (guildId ?? channelId) and StreamConnection (rtc_server_id). */
get serverId(): string | null {
throw new Error("serverId not implemented");
}
/** identifies with media server with credentials */
identify(): void {
if (!this.serverId) throw new Error("Server ID is null or empty");
if (!this.session_id) throw new Error("Session ID is null or empty");
if (!this.token) throw new Error("Token is null or empty");
this.sendOpcode(VoiceOpCodes.IDENTIFY, {
server_id: this.serverId,
user_id: this.botId,
session_id: this.session_id,
token: this.token,
video: true,
streams: STREAMS_SIMULCAST,
max_dave_protocol_version: Davey.DAVE_PROTOCOL_VERSION ?? 0,
});
}
resume(): void {
if (!this.serverId) throw new Error("Server ID is null or empty");
if (!this.session_id) throw new Error("Session ID is null or empty");
if (!this.token) throw new Error("Token is null or empty");
this.sendOpcode(VoiceOpCodes.RESUME, {
server_id: this.serverId,
session_id: this.session_id,
token: this.token,
seq_ack: this._sequenceNumber,
});
}
/** Sets protocols and ip data used for video and audio (vp8 video, opus audio). */
async setProtocols(): Promise<void> {
if (!this._webRtcParams) throw new Error("WebRTC parameters not set");
if (!isNativeAvailable()) {
throw new Error(
"libdatachannel-min native binding not built — cannot start GoLive",
);
}
const reconnect = () => {
const webRtcConn = this._webRtcWrapper.initWebRtc();
webRtcConn.onStateChange((state) => {
console.log(`[goLive:${this.constructor.name}] pc state => ${state}`);
if (state === "closed" && !this._closed) reconnect();
});
this._webRtcWrapper.onLocalDescription = (sdp) => {
const rtc_connection_id = randomUUID();
console.log(
`[goLive:${this.constructor.name}] sending SELECT_PROTOCOL (offer ${sdp.length}B, rtc_connection_id=${rtc_connection_id.slice(0, 8)})`,
);
this.sendOpcode(VoiceOpCodes.SELECT_PROTOCOL, {
protocol: "webrtc",
codecs: Object.values(CodecPayloadType),
data: sdp,
sdp,
rtc_connection_id,
});
};
// createOffer (binding resolves full SDP incl. candidates after gathering)
void webRtcConn
.createOffer()
.then((sdp) => {
this._webRtcWrapper.onLocalDescription?.(sdp);
})
.catch((err: unknown) => {
// PC closed while offer is gathering (stream teardown / reconnect) —
// swallow, the reconnect loop will start a fresh offer.
console.log(
`[goLive:${this.constructor.name}] createOffer rejected: ${err instanceof Error ? err.message : String(err)}`,
);
});
};
reconnect();
return new Promise((resolve) => {
this.once("select_protocol_ack", () => resolve());
});
}
setVideoAttributes(enabled: boolean, attr?: VideoAttribute): void {
if (!this._webRtcParams) throw new Error("WebRTC parameters not set");
const { audioSsrc, videoSsrc, rtxSsrc } = this._webRtcParams;
const payload = !enabled
? {
audio_ssrc: audioSsrc,
video_ssrc: 0,
rtx_ssrc: 0,
streams: [],
}
: (() => {
if (!attr) throw new Error("Need to specify video attributes");
return {
audio_ssrc: audioSsrc,
video_ssrc: videoSsrc,
rtx_ssrc: rtxSsrc,
streams: [
{
type: "video",
rid: "100",
ssrc: videoSsrc,
active: true,
quality: 100,
rtx_ssrc: rtxSsrc,
// hardcode the max bitrate because we don't really know anyway
max_bitrate: 10000 * 1000,
max_framerate: enabled ? attr.fps : 0,
max_resolution: {
type: "fixed",
width: attr.width,
height: attr.height,
},
},
],
};
})();
// CRITICAL: The VIDEO opcode (op 12) is what tells Discord's media server
// to actually forward the video RTP stream on video_ssrc. sendOpcode() is a
// no-op when ws.readyState !== OPEN — and in GoLive the StreamConnection's
// WebSocket can still be in CONNECTING immediately after SELECT_PROTOCOL_ACK
// (the ack listener resolves playStream, but the data channel / ws open
// handshake may lag by a few ms). A dropped op 12 → Discord never activates
// the video SSRC → black/broken video while audio (whose SPEAKING on the
// VoiceConnection already fired) plays fine. Retry until the ws is OPEN
// instead of silently dropping this mandatory signal.
this.sendOpcodeWhenOpen(VoiceOpCodes.VIDEO, payload, "VIDEO");
}
/** Set speaking status */
setSpeaking(speaking: boolean): void {
if (!this._webRtcParams) throw new Error("WebRTC connection not ready");
const payload = {
delay: 0,
speaking: speaking ? 1 : 0,
ssrc: this._webRtcParams.audioSsrc,
};
// Same race as setVideoAttributes: SPEAKING (op 5) must reach Discord. Retry
// until the ws is OPEN rather than dropping it on a transient not-yet-open.
this.sendOpcodeWhenOpen(VoiceOpCodes.SPEAKING, payload, "SPEAKING");
}
/**
* Send an opcode, retrying for a short window if the WebSocket is not yet
* OPEN. Discord's media signaling (VIDEO/op12, SPEAKING/op5) is mandatory —
* a silent no-op (the default sendOpcode behaviour when ws is still
* CONNECTING) breaks GoLive video while leaving audio intact. We wait for
* the open state instead of dropping.
*/
private sendOpcodeWhenOpen(code: number, data: unknown, label: string): void {
const attempt = (triesLeft: number) => {
if (this.ws?.readyState === WebSocket.OPEN) {
this.sendOpcode(code, data);
return;
}
if (triesLeft <= 0) {
console.error(
`[goLive:${this.constructor.name}] ${label} opcode (op=${code}) DROPPED — ws never opened (state=${this.ws?.readyState ?? "null"})`,
);
return;
}
// ws still CONNECTING (or briefly closed during reconnect) — retry.
setTimeout(() => attempt(triesLeft - 1), 50);
};
attempt(40); // up to ~2s
}
}
export type { NativePeerConnection };
services/discord-gateway/src/goLive/BaseMediaStream.ts
-175
View File
@@ -1,175 +0,0 @@
/**
* BaseMediaStream — pacing/sync for GoLive frames. Ported from
* @dank074/discord-video-stream BaseMediaStream.js, minus node-av's
* AVFrame (frames are plain objects here) and debug-level (uses the GMW
* logger instead).
*/
import { Writable } from "node:stream";
import { setTimeout as sleep } from "node:timers/promises";
export interface GoLiveFrame {
data: Buffer | null;
pts: number;
duration: number;
timeBase: { num: number; den: number };
free?: () => void;
}
export class BaseMediaStream extends Writable {
_pts: number | undefined;
_syncTolerance = 20;
_noSleep: boolean;
_startTime: number | undefined;
_startPts: number | undefined;
_sync = true;
_syncStream: BaseMediaStream | undefined;
_type: string;
constructor(type: string, noSleep = false) {
super({ objectMode: true, highWaterMark: 0 });
this._type = type;
this._noSleep = noSleep;
}
get sync(): boolean {
return this._sync;
}
set sync(val: boolean) {
this._sync = val;
}
get syncStream(): BaseMediaStream | undefined {
return this._syncStream;
}
set syncStream(stream: BaseMediaStream | undefined) {
if (stream !== undefined && this === stream.syncStream) {
throw new Error("Cannot sync 2 streams with eachother");
}
this._syncStream = stream;
}
get noSleep(): boolean {
return this._noSleep;
}
set noSleep(val: boolean) {
this._noSleep = val;
if (!val) this.resetTimingCompensation();
}
get pts(): number | undefined {
return this._pts;
}
get syncTolerance(): number {
return this._syncTolerance;
}
set syncTolerance(n: number) {
if (n < 0) return;
this._syncTolerance = n;
}
async _sendFrame(_frame: Buffer, _frametime: number): Promise<void> {
throw new Error("Not implemented");
}
ptsDelta(): number | undefined {
if (this.pts !== undefined && this.syncStream?.pts !== undefined) {
return this.pts - this.syncStream.pts;
}
return undefined;
}
isAhead(): boolean {
const delta = this.ptsDelta();
return (
this.syncStream?.writableEnded === false &&
delta !== undefined &&
delta > this.syncTolerance
);
}
isBehind(): boolean {
const delta = this.ptsDelta();
return (
this.syncStream?.writableEnded === false &&
delta !== undefined &&
delta < -this.syncTolerance
);
}
resetTimingCompensation(): void {
this._startTime = this._startPts = undefined;
}
async _write(
frame: GoLiveFrame,
_encoding: BufferEncoding,
callback: (error?: Error | null) => void,
): Promise<void> {
const { data, pts, duration, timeBase } = frame;
if (!data) {
frame.free?.();
callback();
return;
}
const frametime = (Number(duration) / timeBase.den) * timeBase.num * 1000;
const start_sendFrame = performance.now();
await this._sendFrame(Buffer.from(data), frametime);
const end_sendFrame = performance.now();
this._pts = (Number(pts) / timeBase.den) * timeBase.num * 1000;
this.emit("pts", this._pts);
const sendTime = end_sendFrame - start_sendFrame;
const ratio = sendTime / frametime;
if (ratio > 1) {
// Frame takes longer to send than its frametime — warn once per 100
if (
this._lastWarnedRatio === undefined ||
ratio > this._lastWarnedRatio
) {
this._lastWarnedRatio = ratio;
}
}
this._startTime ??= start_sendFrame;
this._startPts ??= this._pts;
const sleepMs = Math.max(
0,
this._pts -
this._startPts +
frametime -
(end_sendFrame - this._startTime),
);
if (this._noSleep || sleepMs === 0) {
callback(null);
} else if (this.sync && this.isBehind()) {
// Stream is behind — don't sleep for this frame
this.resetTimingCompensation();
callback(null);
} else if (this.sync && this.isAhead()) {
// Stream is ahead — wait until the sync stream catches up
do {
await sleep(frametime);
} while (this.sync && this.isAhead());
this.resetTimingCompensation();
callback(null);
} else {
await sleep(sleepMs);
callback(null);
}
frame.free?.();
}
_lastWarnedRatio: number | undefined;
_destroy(
error: Error | null,
callback: (error?: Error | null) => void,
): void {
super._destroy(error, callback);
this.syncStream = undefined;
}
}
services/discord-gateway/src/goLive/CodecPayloadType.ts
-71
View File
@@ -1,71 +0,0 @@
/** Payload types for Discord GoLive media — ported from @dank074/discord-video-stream. */
export interface CodecPayloadTypeEntry {
name: string;
type: "audio" | "video";
clockRate: number;
priority: number;
payload_type: number;
rtx_payload_type?: number;
encode?: boolean;
decode?: boolean;
}
export const CodecPayloadType: Record<string, CodecPayloadTypeEntry> = {
opus: {
name: "opus",
type: "audio",
clockRate: 48000,
priority: 1000,
payload_type: 120,
},
H264: {
name: "H264",
type: "video",
clockRate: 90000,
priority: 1000,
payload_type: 101,
rtx_payload_type: 102,
encode: true,
decode: true,
},
H265: {
name: "H265",
type: "video",
clockRate: 90000,
priority: 1000,
payload_type: 103,
rtx_payload_type: 104,
encode: true,
decode: true,
},
VP8: {
name: "VP8",
type: "video",
clockRate: 90000,
priority: 1000,
payload_type: 105,
rtx_payload_type: 106,
encode: true,
decode: true,
},
VP9: {
name: "VP9",
type: "video",
clockRate: 90000,
priority: 1000,
payload_type: 107,
rtx_payload_type: 108,
encode: true,
decode: true,
},
AV1: {
name: "AV1",
type: "video",
clockRate: 90000,
priority: 1000,
payload_type: 109,
rtx_payload_type: 110,
encode: true,
decode: true,
},
};
services/discord-gateway/src/goLive/Demuxer.ts
-647
View File
@@ -1,647 +0,0 @@
/**
* Lightweight demuxer — replaces node-av's LibavDemuxer for GoLive.
*
* Spawns ffmpeg to remux input into H264 AnnexB on stdout (video only —
* screen share doesn't need to mux audio into the demuxer; audio goes
* separately). This replaces the 114MB node-av binary with a plain ffmpeg
* spawn.
*
* Each video "frame" emitted is a complete NAL sequence terminated by a
* keyframe boundary (IDR). Audio is not extracted here — for GoLive with
* audio, the NUT mux + full demuxer would be needed; screen share audio is
* handled via a separate ffmpeg instance (see getDirectScreenInput).
*/
import { spawn } from "node:child_process";
import { existsSync, readdirSync } from "node:fs";
import { join } from "node:path";
import { PassThrough } from "node:stream";
import type { GoLiveFrame } from "./BaseMediaStream.js";
/** 4-byte AnnexB start code (00 00 00 01) used when building access units. */
const startCode4 = Buffer.from([0, 0, 0, 1]);
/**
* Resolve ffmpeg/ffprobe binary. Prefers explicit env override, then PATH,
* then a Nix-store ffmpeg-headless (the GMW flake provides it in the service
* profile, but dev shells / tests may not have it on PATH).
*/
function resolveBin(name: "ffmpeg"): string {
const override = process.env.FFMPEG_PATH;
if (override && existsSync(override)) return override;
// Nix store scan: <store>/<hash>-ffmpeg-headless-*/bin/<name>
const store = "/nix/store";
if (existsSync(store)) {
const entries = readdirSync(store);
for (const entry of entries) {
if (!entry.includes("ffmpeg-headless-")) continue;
const candidate = join(store, entry, "bin", name);
if (existsSync(candidate)) return candidate;
}
}
return name; // fall back to PATH
}
const FFMPEG = resolveBin("ffmpeg");
export const AVCodecID = {
AV_CODEC_ID_H264: 27,
AV_CODEC_ID_HEVC: 173,
AV_CODEC_ID_VP8: 139,
AV_CODEC_ID_VP9: 167,
AV_CODEC_ID_AV1: 225,
AV_CODEC_ID_OPUS: 86019,
} as const;
export type AVCodecID = (typeof AVCodecID)[keyof typeof AVCodecID];
export const AV_PKT_FLAG_KEY = 1;
export interface Frame {
data: Buffer | null;
pts: number;
duration: number;
timeBase: { num: number; den: number };
flags: number;
streamIndex: number;
free(): void;
}
export interface DemuxedStream {
codec: number;
codecName: string;
width: number;
height: number;
framerate_num: number;
framerate_den: number;
sample_rate: number;
stream: PassThrough;
}
/**
* Probe a media file for stream info using ffmpeg's stderr (the
* ffmpeg-headless Nix package ships ffmpeg but not ffprobe). Returns
* stream descriptors in the same shape ffprobe -show_streams would.
*/
export async function probeStreams(
url: string,
): Promise<Array<Record<string, unknown>>> {
return new Promise((resolve, reject) => {
const proc = spawn(FFMPEG, [
"-hide_banner",
"-loglevel",
"info",
"-i",
url,
"-f",
"null",
"-",
]);
let stderr = "";
proc.stderr.on("data", (d: Buffer) => (stderr += d.toString()));
proc.on("close", () => {
// Parse "Stream #0:0: Video: h264 (High), yuv420p, 640x360, 30 fps"
const streams: Array<Record<string, unknown>> = [];
const re = /Stream #0:(\d+): (Video|Audio): ([^,]+)/g;
let m: RegExpExecArray | null;
// biome-ignore lint/suspicious/noAssignInExpressions: regex loop idiom
while ((m = re.exec(stderr)) !== null) {
const [full, idx, kind, codecRaw] = m;
void full;
const codecName = codecRaw.split(" ")[0].toLowerCase();
const stream: Record<string, unknown> = {
index: Number(idx),
codec_type: kind.toLowerCase(),
codec_name: codecName,
width: 0,
height: 0,
r_frame_rate: "0/1",
sample_rate: 0,
};
// dimensions: "640x360"
const dim = /(\d{2,5})x(\d{2,5})/.exec(stderr.slice(m.index));
if (dim) {
stream.width = Number(dim[1]);
stream.height = Number(dim[2]);
}
// fps: "30 fps" or "29.97 fps"
const fps = /(\d+(?:\.\d+)?) fps/.exec(stderr.slice(m.index));
if (fps) {
const v = Number(fps[1]);
stream.r_frame_rate = `${Math.round(v * 1000)}/1000`;
}
// sample rate for audio: "48000 Hz"
const sr = /(\d+) Hz/.exec(stderr.slice(m.index));
if (sr) stream.sample_rate = Number(sr[1]);
streams.push(stream);
}
resolve(streams);
});
proc.on("error", (err) => reject(err));
});
}
/**
* Demux input (URL string or readable stream) into video frames on a
* PassThrough. Streams input DIRECTLY into ffmpeg (no spool-to-file — the
* live NUT/H264 source never ends, so spooling deadlocks). ffmpeg emits
* AnnexB H264 on stdout; NAL units are split into frames on the fly.
* Video metadata is parsed from ffmpeg stderr during init.
*/
export async function demux(
input: string | PassThrough,
opts: { format: string; frameRate?: number },
): Promise<{
video: DemuxedStream | undefined;
audio: DemuxedStream | undefined;
close: () => void;
}> {
// objectMode pipes carrying one frame per item. HWM 2 keeps backpressure
// near-instant: at most ~1-2 frames in flight (~66ms @ 30fps) before the
// encoder is throttled, so the viewer sees near-live video instead of a
// multi-second backlog (the old HWM 128 held 128 frames ≈ 4.3s of lag).
// BaseMediaStream below has HWM 0, so the chain is tightly coupled to the
// WebRTC sender's real pace — faithful to @dank074/discord-video-stream.
const vPipe = new PassThrough({ objectMode: true, highWaterMark: 2 });
const aPipe = new PassThrough({ objectMode: true, highWaterMark: 2 });
const isStream = typeof input !== "string";
// NUT/matroska input (prepareStream with includeAudio) carries audio; the
// h264 path is video-only raw AnnexB. Video always goes to stdout (pipe:1);
// audio goes to fd3 (pipe:3) so stderr stays free for metadata parsing.
const containerFormat =
isStream && opts.format !== "h264" ? opts.format : null;
const withAudio = isStream && containerFormat !== null;
const args: string[] = [
"-hide_banner",
// info level: stream init lines ("Stream #0:0: Video: h264...") go to
// stderr and are parsed for dimensions/fps.
"-loglevel",
"info",
// Real-time throttle: read piped input at 1x so the WHOLE upstream chain
// (encoder x264, merge ffmpeg, yt-dlp download) is paced by wall-clock,
// not by network/VOD speed. Without this the encoder bursts ~10x faster
// than real-time and the vPipe queue grows unboundedly — the WebRTC
// sender correctly paces 30fps but always emits the OLDEST buffered
// frame, so video freezes/lags while audio (small, jitter-buffer
// recoverable) stays smooth. Pausing proc.stdout instead (previous fix)
// stalled the same ffmpeg's fd3 audio too → audio stuttered AND the
// already-built backlog never drained. `-re` fixes production rate at
// source; a bounded backlog below still protects against sender stalls.
...(isStream ? ["-re"] : []),
// Input format hint: raw H264 has NO magic header, so ffmpeg's
// auto-detection fails with "Invalid data found when processing input"
// whenever the first bytes arrive late/buffered. Pin the demuxer input
// format for streams (NUT for the audio-capable path).
...(withAudio
? ["-f", containerFormat as string]
: isStream
? ["-f", "h264"]
: []),
"-i",
isStream ? "pipe:0" : input,
"-map",
"0:v:0",
"-c:v",
"copy",
"-f",
"h264",
"pipe:1",
...(withAudio
? ["-map", "0:a:0?", "-c:a", "copy", "-f", "opus", "pipe:3"]
: ["-an"]),
];
const proc = spawn(FFMPEG, args, {
stdio: isStream
? withAudio
? ["pipe", "pipe", "pipe", "pipe"]
: ["pipe", "pipe", "pipe"]
: ["ignore", "pipe", "pipe"],
});
console.log(
`[goLive:Demuxer] spawn ffmpeg pid=${proc.pid} input=${isStream ? "stream" : input} args=${args.join(" ")}`,
);
// Pipe live input straight into ffmpeg stdin — never await stream end.
if (isStream && proc.stdin) {
input.pipe(proc.stdin);
input.on("error", () => proc.stdin?.destroy());
}
// Audio: ffmpeg writes Ogg Opus on fd3 (pipe:3). Parse OGG pages into
// opus packets and emit them as GoLiveFrames (20ms, 48kHz) on aPipe.
if (withAudio && proc.stdio[3]) {
createOggOpusDemux(
proc.stdio[3] as unknown as NodeJS.ReadableStream,
aPipe,
);
console.log("[goLive:Demuxer] audio pipe wired (fd3 → Ogg Opus → aPipe)");
}
// Track which stream kinds we've seen. We must NOT stop parsing on the
// first stream found: ffmpeg can print the video line and audio line in
// separate stderr chunks (input arrives slowly), and the old
// early-return dropped the audio line forever
// → aInfo undefined → no audio RTP → static GoLive tile.
let seenVideo = false;
let seenAudio = false;
// Parse stream metadata from ffmpeg stderr as it arrives (first chunk has
// the init lines). Fall back to H264 defaults if parsing fails.
let vInfo: DemuxedStream = {
codec: AVCodecID.AV_CODEC_ID_H264,
codecName: "h264",
width: 0,
height: 0,
framerate_num: 0,
framerate_den: 1,
sample_rate: 0,
stream: vPipe,
};
let aInfo: DemuxedStream | undefined;
// With audio expected (NUT input), ALWAYS expose an audio stream even if
// ffmpeg's audio init line hasn't arrived in stderr yet. prepareStream
// encodes libopus into the NUT unconditionally (`-map 0:a:0? -c:a libopus`),
// so fd3 WILL carry Ogg Opus — aInfo must not stay undefined just because
// the metadata line raced the resolve. The stderr handler below upgrades
// this default with real sample_rate metadata when the line lands.
if (withAudio) {
aInfo = {
codec: AVCodecID.AV_CODEC_ID_OPUS,
codecName: "opus",
width: 0,
height: 0,
framerate_num: 0,
framerate_den: 0,
sample_rate: 48000,
stream: aPipe,
};
}
let stderrBuf = "";
if (proc.stderr) {
proc.stderr.on("data", (d: Buffer) => {
const text = d.toString();
stderrBuf = (stderrBuf + text).slice(-16384);
// Surface actionable lines: ffmpeg errors + stream init lines
if (/error|invalid|no such|failed|cannot|not found|unable/i.test(text)) {
console.log(
`[goLive:Demuxer] ffmpeg stderr: ${text.trim().split("\n").slice(0, 4).join(" | ")}`,
);
}
const streamRe = /Stream #0:(\d+): (Video|Audio): ([^,]+)/g;
let m: RegExpExecArray | null;
const found: Array<{ kind: string; codecRaw: string }> = [];
// biome-ignore lint/suspicious/noAssignInExpressions: regex loop idiom
while ((m = streamRe.exec(stderrBuf)) !== null) {
found.push({ kind: m[2], codecRaw: m[3] });
}
if (process.env.GMW_DEMUX_DEBUG) {
console.log(
`[goLive:Demuxer] DEBUG stderrBuf=${JSON.stringify(stderrBuf.slice(0, 300))} found=${JSON.stringify(found)}`,
);
}
const v = found.find((s) => s.kind === "Video");
const a = found.find((s) => s.kind === "Audio");
if (v) {
seenVideo = true;
const codecName = v.codecRaw.split(" ")[0].toLowerCase();
const dim = /(\d{2,5})x(\d{2,5})/.exec(stderrBuf);
const fps = /(\d+(?:\.\d+)?) fps/.exec(stderrBuf);
vInfo = {
codec:
AVCodecID[
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
"AV_CODEC_ID_H264"
] ?? AVCodecID.AV_CODEC_ID_H264,
codecName,
width: dim ? Number(dim[1]) : 0,
height: dim ? Number(dim[2]) : 0,
framerate_num: fps ? Math.round(Number(fps[1]) * 1000) : 0,
framerate_den: fps ? 1000 : 1,
sample_rate: 0,
stream: vPipe,
};
}
if (a) {
seenAudio = true;
const codecName = a.codecRaw.split(" ")[0].toLowerCase();
const sr = /(\d+) Hz/.exec(stderrBuf);
aInfo = {
codec:
AVCodecID[
(codecName.toUpperCase() as keyof typeof AVCodecID) ??
"AV_CODEC_ID_OPUS"
] ?? AVCodecID.AV_CODEC_ID_OPUS,
codecName,
width: 0,
height: 0,
framerate_num: 0,
framerate_den: 0,
sample_rate: sr ? Number(sr[1]) : 0,
stream: aPipe,
};
}
// Mark that at least one stream kind was seen. Note: we must NOT
// resolve the metadata wait on the FIRST stream kind alone. With live
// NUT input, ffmpeg can print the video init line in one stderr chunk
// and the audio init line in the NEXT chunk (NUT info-stream packets
// arrive as ffmpeg reads them from the pipe). The old code returned
// immediately on `parsedMeta=true` — the audio line then landed in the
// handler AFTER `return { audio: aInfo }` had already captured
// `undefined` → no audio RTP → static GoLive tile even though the NUT
// carried audio. Wait for BOTH kinds (when audio is expected).
// (parsedMeta removed — we now wait for both via allSeen() below.)
});
}
// Wait (briefly) for ffmpeg to print its stream init lines on stderr so
// vInfo/aInfo carry real metadata. With audio expected, wait for BOTH the
// video and audio init lines (they may arrive in separate stderr chunks on
// live input); the timeout covers slow starts / genuinely audio-less input.
const allSeen = () =>
withAudio ? seenVideo && seenAudio : seenVideo || seenAudio;
await Promise.race([
new Promise<void>((resolve) => {
const check = setInterval(() => {
if (allSeen()) {
clearInterval(check);
resolve();
}
}, 25);
}),
new Promise<void>((resolve) => setTimeout(resolve, 3000)),
]);
// Scan stdout for AnnexB NAL units and group them into ACCESS UNITS
// (one picture). Discord's H264 decoder requires a complete access unit —
// parameter sets + slice — inside a single RTP frame. Emitting each NAL
// as its own frame (SPS/PPS/SEI separate from the slice) makes the decoder
// unable to produce ANY picture: production showed a black GoLive tile
// despite frames flowing (5892B slices + 4B PPS + 33B SPS as separate
// frames, each with a near-zero RTP timestamp delta). We therefore buffer
// NALs and flush one frame per slice, prepending the parameter sets that
// precede it, and timestamp it as ONE frame at the video frame rate.
// EMISSION: faithful to @dank074/discord-video-stream — the demuxer does NOT
// pace. It writes each access unit straight to vPipe (objectMode, HWM 128)
// with a monotonically increasing PTS; BaseMediaStream (ported 1:1 from dank)
// then paces playback via sleep-PTS + A/V sync and applies backpressure when
// the WebRTC sender can't keep up. This is what works upstream; the custom
// setInterval/tail-drop clocks we tried broke IDR delivery → blank tiles.
let videoBuf = Buffer.alloc(0);
let frameCount = 0;
let pendingNals: Buffer[] = [];
let pendingHasSlice = false;
let pendingIsKey = false;
// Raw H264 streams carry no timing info — ffmpeg's h264 demuxer guesses
// 25fps on stderr. Prefer the caller's explicit frameRate (the encode
// setting); it drives both RTP timestamp advance and pacing.
const videoFps =
opts.frameRate ?? (vInfo.framerate_num / vInfo.framerate_den || 30);
// Write one frame to the video pipe with backpressure: if the pipe buffer is
// full, pause ffmpeg's stdout so the encoder self-throttles (instead of
// building an unbounded backlog). Resumed on drain. Faithful to dank's
// `resume &&= vPipe.write(packet)` in LibavDemuxer.
const writeFrame = (frame: {
data: Buffer;
pts: number;
duration: number;
timeBase: { num: number; den: number };
flags: number;
streamIndex: number;
free: () => void;
}): void => {
const ok = vPipe.write(frame);
if (!ok) proc.stdout?.pause();
};
const flushAccessUnit = () => {
if (pendingNals.length === 0) return;
// AnnexB access unit: 00 00 00 01 + NAL for every buffered NAL. The
// packetizer (H264RtpPacketizer, StartSequence separator) needs the
// start codes to find NAL boundaries inside the frame.
const parts: Buffer[] = [];
for (const n of pendingNals) parts.push(startCode4, n);
const au = Buffer.concat(parts);
const isKey = pendingIsKey;
pendingNals = [];
pendingHasSlice = false;
pendingIsKey = false;
// Write directly to the video pipe (with backpressure via writeFrame).
// PTS advances one frame per output frame at videoFps (duration=1 in a
// 1/fps timebase) so BaseMediaStream computes the correct frametime and the
// WebRTC RTP timestamp advances by clockRate/fps per frame (3000 @ 30fps /
// 90kHz). Keyframes carry AV_PKT_FLAG_KEY so the decoder re-establishes a
// reference.
writeFrame({
data: au,
pts: frameCount,
duration: 1,
timeBase: { num: 1, den: videoFps },
flags: isKey ? AV_PKT_FLAG_KEY : 0,
streamIndex: 0,
free: () => {},
});
frameCount++;
if (frameCount === 1 || frameCount % 30 === 0) {
console.log(
`[goLive:Demuxer] frames=${frameCount} last=${au.length}B key=${isKey}`,
);
}
};
if (proc.stdout) {
vPipe.on("drain", () => proc.stdout?.resume());
proc.stdout.on("data", (chunk: Buffer) => {
videoBuf = Buffer.concat([videoBuf, chunk]);
// Find start codes (00 00 01 or 00 00 00 01) and split NALs
let start = 0;
// If buffer starts with zeros, that's the first start code — emit from there
while (start < videoBuf.length) {
let scPos = -1;
for (let i = start + 1; i < videoBuf.length - 2; i++) {
if (
videoBuf[i] === 0 &&
videoBuf[i + 1] === 0 &&
videoBuf[i + 2] === 1
) {
scPos = i + 3;
break;
}
}
if (scPos === -1) break;
// Emit the NAL from `start` to `scPos` (but skip the start code bytes at `start`)
if (start < scPos) {
let nalStart = start;
// Skip start code bytes for the NAL itself (00 00 01)
if (
videoBuf[nalStart] === 0 &&
videoBuf[nalStart + 1] === 0 &&
videoBuf[nalStart + 2] === 1
) {
nalStart += 3;
} else if (
nalStart + 3 < scPos &&
videoBuf[nalStart] === 0 &&
videoBuf[nalStart + 1] === 0 &&
videoBuf[nalStart + 2] === 0 &&
videoBuf[nalStart + 3] === 1
) {
nalStart += 4;
}
const nal = videoBuf.subarray(nalStart, scPos);
// Trim trailing zero bytes (from start code overlap)
let end = nal.length;
while (end > 0 && nal[end - 1] === 0) end--;
if (end > 0) {
const nalTrimmed = nal.subarray(0, end);
const nalType = nalTrimmed[0] & 0x1f;
const isSlice = nalType === 1 || nalType === 5;
if (isSlice) {
// A new slice while one is pending closes the previous
// access unit (x264 emits one slice per frame).
if (pendingHasSlice) flushAccessUnit();
pendingNals.push(Buffer.from(nalTrimmed));
pendingHasSlice = true;
if (nalType === 5) pendingIsKey = true;
} else {
// Parameter-set / SEI / AUD / filler NAL. After a slice these
// belong to the NEXT access unit — flush the completed frame.
if (pendingHasSlice) flushAccessUnit();
pendingNals.push(Buffer.from(nalTrimmed));
}
}
}
// Skip the 00 00 01 at scPos-3 to find next
start = scPos;
// But the next start code needs at least 3 bytes
if (start > videoBuf.length - 3) break;
}
// Keep remaining bytes (potential partial NAL or start code)
if (start > 0 && start < videoBuf.length) {
videoBuf = videoBuf.subarray(start);
} else if (videoBuf.length > 4) {
// No full NAL found, but avoid unbounded growth
// Keep a sliding window
videoBuf = videoBuf.subarray(videoBuf.length - 3);
}
});
// Backpressure (faithful to dank's `resume &&= vPipe.write`): when the
// downstream pipe's buffer is full, stop reading from ffmpeg's stdout so
// the encoder self-throttles instead of building an unbounded backlog.
// Resume on drain.
vPipe.on("drain", () => proc.stdout?.resume());
proc.stdout.on("end", () => {
flushAccessUnit();
vPipe.end();
aPipe.end();
});
}
proc.on("close", () => {
vPipe.end();
aPipe.end();
});
const close = () => {
proc.kill("SIGTERM");
vPipe.end();
aPipe.end();
};
return { video: vInfo, audio: aInfo, close };
}
/**
* Parse an Ogg Opus byte stream (as written by ffmpeg's `-f opus` muxer)
* into individual opus packets and push them onto `out` as GoLiveFrames
* (duration 960 @ 48kHz = 20ms, matching the OpusRtpPacketizer clock).
*
* OGG page structure:
* "OggS" | ver(1) | header_type(1) | granule(8 LE) | serial(4) | seq(4) |
* crc(4) | page_segments(1) | segment_table[n] | payload
* Lacing: a value < 255 ends a packet; 255 continues it (0.5KB chunk).
* The first packet is OpusHead (19B) — skipped, as is OpusTags.
*/
function createOggOpusDemux(
input: NodeJS.ReadableStream,
out: PassThrough,
): void {
let buf = Buffer.alloc(0);
// Packets assembled from lacing; packetParts accumulates across pages
// when a packet spans a page boundary (continued flag / 255 lacing).
let packetParts: Buffer[] = [];
let headerDone = false;
let frameIndex = 0;
const emitPacket = (packet: Buffer) => {
if (!headerDone) {
// First packet = OpusHead ("OpusHead"), second = OpusTags. Skip both.
const magic = packet.toString("latin1", 0, 8);
if (magic === "OpusHead" || magic === "OpusTags") return;
headerDone = true;
}
out.write({
data: packet,
pts: frameIndex * 960,
duration: 960,
timeBase: { num: 1, den: 48000 },
free: () => {},
} satisfies GoLiveFrame);
frameIndex++;
};
const processPages = () => {
while (true) {
// Sync to "OggS"
const sync = buf.indexOf("OggS", 0, "latin1");
if (sync === -1) {
// Keep the tail (partial sync pattern) for the next chunk
buf = buf.length > 3 ? buf.subarray(buf.length - 3) : buf;
return;
}
if (sync > 0) buf = buf.subarray(sync);
if (buf.length < 27) return; // need full page header
const numSeg = buf[26];
if (buf.length < 27 + numSeg) return; // need segment table
let payloadLen = 0;
for (let i = 0; i < numSeg; i++) payloadLen += buf[27 + i];
if (buf.length < 27 + numSeg + payloadLen) return; // need payload
const headerType = buf[5];
// Extract packets from the payload using lacing values
let off = 27 + numSeg;
for (let i = 0; i < numSeg; i++) {
const lace = buf[27 + i];
const part = buf.subarray(off, off + lace);
off += lace;
packetParts.push(Buffer.from(part));
if (lace < 255) {
const packet = Buffer.concat(packetParts);
packetParts = [];
if ((headerType & 0x01) === 0) {
// Not a continuation page → packet starts here
emitPacket(packet);
} else if (headerDone) {
// Continued page — packet body, emit directly
emitPacket(packet);
}
// (header packets on continuation pages are dropped)
}
}
buf = buf.subarray(off);
if (buf.length === 0) return;
}
};
input.on("data", (chunk: Buffer) => {
buf = Buffer.concat([buf, chunk]);
processPages();
});
input.on("end", () => {
out.end();
});
input.on("error", () => {
out.end();
});
}
services/discord-gateway/src/goLive/Encoders.ts
-69
View File
@@ -1,69 +0,0 @@
/**
* Lightweight encoders config — ported from @dank074/discord-video-stream
* encoders/software.js. Only software (libx264) is needed for GoLive.
*/
export interface EncoderSettings {
name: string;
options: string[];
outFilters?: string[];
globalOptions?: string[];
}
export interface EncoderSet {
H264: EncoderSettings;
H265: EncoderSettings;
VP8: EncoderSettings;
VP9: EncoderSettings;
AV1: EncoderSettings;
}
/** Software x264 encoder. Matches @dank074's software() defaults. */
export function software(
opts: {
x264?: { preset?: string; tune?: string };
x265?: { preset?: string; tune?: string };
} = {},
): () => EncoderSet {
const { x264, x265 } = opts;
const { preset: x264Preset = "superfast", tune: x264Tune = "zerolatency" } =
x264 ?? {};
const { preset: x265Preset = "superfast", tune: x265Tune } = x265 ?? {};
return () => ({
H264: {
name: "libx264",
// -profile:v baseline is REQUIRED: the SDP advertises
// profile-level-id=42e01f (constrained baseline) and Discord's
// receiver decodes with that profile. x264's default is High — a
// High-profile bitstream against a baseline SDP negotiation fails to
// decode → black GoLive tile (production bug, fixed 2026-08-12).
// zerolatency matches @dank074 (no lookahead — correct for live).
options: [
"-forced-idr 1",
"-profile:v baseline",
// repeat-headers: SPS/PPS inline BEFORE EVERY IDR, not just the
// first. Required for the NUT container path (NUT stores extradata
// in the header and `-c:v copy` remux loses it → decoder sees
// "non-existing PPS 0 referenced" → no picture) and lets Discord's
// decoder recover after any PLI/keyframe request mid-stream.
"-x264-params",
"repeat-headers=1",
`-tune ${x264Tune}`,
`-preset ${x264Preset}`,
],
},
H265: {
name: "libx265",
options: [
"-forced-idr 1",
...(x265Tune ? [`-tune ${x265Tune}`] : []),
`-preset ${x265Preset}`,
],
},
VP8: { name: "libvpx", options: ["-deadline 20000"] },
VP9: { name: "libvpx-vp9", options: ["-deadline 20000"] },
AV1: { name: "libsvtav1", options: [] },
});
}
export const Encoders = { software };
services/discord-gateway/src/goLive/GatewayOpCodes.ts
-41
View File
@@ -1,41 +0,0 @@
/** Discord gateway opcodes used by Streamer — ported from @dank074/discord-video-stream. */
export enum GatewayOpCodes {
DISPATCH = 0,
HEARTBEAT = 1,
IDENTIFY = 2,
PRESENCE_UPDATE = 3,
VOICE_STATE_UPDATE = 4,
VOICE_SERVER_PING = 5,
RESUME = 6,
RECONNECT = 7,
REQUEST_GUILD_MEMBERS = 8,
INVALID_SESSION = 9,
HELLO = 10,
HEARTBEAT_ACK = 11,
CALL_CONNECT = 13,
GUILD_SUBSCRIPTIONS = 14,
LOBBY_CONNECT = 15,
LOBBY_DISCONNECT = 16,
LOBBY_VOICE_STATES_UPDATE = 17,
STREAM_CREATE = 18,
STREAM_DELETE = 19,
STREAM_WATCH = 20,
STREAM_PING = 21,
STREAM_SET_PAUSED = 22,
REQUEST_GUILD_APPLICATION_COMMANDS = 24,
EMBEDDED_ACTIVITY_LAUNCH = 25,
EMBEDDED_ACTIVITY_CLOSE = 26,
EMBEDDED_ACTIVITY_UPDATE = 27,
REQUEST_FORUM_UNREADS = 28,
REMOTE_COMMAND = 29,
GET_DELETED_ENTITY_IDS_NOT_MATCHING_HASH = 30,
REQUEST_SOUNDBOARD_SOUNDS = 31,
SPEED_TEST_CREATE = 32,
SPEED_TEST_DELETE = 33,
REQUEST_LAST_MESSAGES = 34,
SEARCH_RECENT_MEMBERS = 35,
REQUEST_CHANNEL_STATUSES = 36,
GUILD_SUBSCRIPTIONS_BULK = 37,
GUILD_CHANNELS_RESYNC = 38,
REQUEST_CHANNEL_MEMBER_COUNT = 39,
}
services/discord-gateway/src/goLive/SPSVUIRewriter.ts
-291
View File
@@ -1,291 +0,0 @@
/**
* H264 SPS VUI rewriter — ported from @dank074/discord-video-stream
* SPSVUIRewriter.js. Rewrites the SPS so Discord's receiver applies
* bitstream restrictions (max_num_reorder_frames=0, max_dec_frame_buffering
* bounded) — required for low-latency GoLive decode.
*/
import {
AnnexBBitstreamReader,
AnnexBBitstreamWriter,
} from "./AnnexBBitstreamReaderWriter.js";
export function rewriteSPSVUI(buffer: Uint8Array): Buffer {
const reader = new AnnexBBitstreamReader(buffer.subarray(1));
const writer = new AnnexBBitstreamWriter();
const readBit = (n = 1) => reader.readBits(n);
const writeBit = (v: number, n = 1) => writer.writeBits(v, n);
const readU = (n: number) => reader.readUnsigned(n);
const writeU = (v: number, n: number) => writer.writeUnsigned(v, n);
const readUE = () => reader.readUnsignedExpGolomb();
const writeUE = (v: number) => writer.writeUnsignedExpGolomb(v);
const readSE = () => reader.readSignedExpGolomb();
const writeSE = (v: number) => writer.writeSignedExpGolomb(v);
// Rewrite the NAL header
writeU(buffer[0], 8);
const profile_idc = readU(8);
writeU(profile_idc, 8);
const constraint_flags = readU(8);
writeU(constraint_flags, 8);
const level_idc = readU(8);
writeU(level_idc, 8);
const seq_parameter_set_id = readUE();
writeUE(seq_parameter_set_id);
// If profile in high profiles, additional fields
const highProfiles = new Set([
100, 110, 122, 244, 44, 83, 86, 118, 128, 138, 144,
]);
if (highProfiles.has(profile_idc)) {
const chroma_format_idc = readUE();
writeUE(chroma_format_idc);
if (chroma_format_idc === 3) {
const separate_colour_plane_flag = readBit(1);
writeBit(separate_colour_plane_flag, 1);
}
const bit_depth_luma_minus8 = readUE();
writeUE(bit_depth_luma_minus8);
const bit_depth_chroma_minus8 = readUE();
writeUE(bit_depth_chroma_minus8);
const qpprime_y_zero_transform_bypass_flag = readBit(1);
writeBit(qpprime_y_zero_transform_bypass_flag, 1);
const seq_scaling_matrix_present_flag = readBit(1);
writeBit(seq_scaling_matrix_present_flag, 1);
if (seq_scaling_matrix_present_flag) {
const scalingCount = chroma_format_idc !== 3 ? 8 : 12;
for (let i = 0; i < scalingCount; i++) {
const seq_scaling_list_present_flag = readBit(1);
writeBit(seq_scaling_list_present_flag, 1);
if (seq_scaling_list_present_flag) {
const size = i < 6 ? 16 : 64;
let lastScale = 8;
let nextScale = 8;
for (let j = 0; j < size; j++) {
const delta = readSE();
writeSE(delta);
nextScale = (lastScale + delta + 256) % 256;
if (nextScale !== 0) lastScale = nextScale;
}
}
}
}
}
const log2_max_frame_num_minus4 = readUE();
writeUE(log2_max_frame_num_minus4);
const pic_order_cnt_type = readUE();
writeUE(pic_order_cnt_type);
if (pic_order_cnt_type === 0) {
const log2_max_pic_order_cnt_lsb_minus4 = readUE();
writeUE(log2_max_pic_order_cnt_lsb_minus4);
} else if (pic_order_cnt_type === 1) {
const delta_pic_order_always_zero_flag = readBit(1);
writeBit(delta_pic_order_always_zero_flag, 1);
const offset_for_non_ref_pic = readSE();
writeSE(offset_for_non_ref_pic);
const offset_for_top_to_bottom_field = readSE();
writeSE(offset_for_top_to_bottom_field);
const num_ref_frames_in_pic_order_cnt_cycle = readUE();
writeUE(num_ref_frames_in_pic_order_cnt_cycle);
for (let i = 0; i < num_ref_frames_in_pic_order_cnt_cycle; i++) {
const offset_for_ref_frame = readSE();
writeSE(offset_for_ref_frame);
}
}
const max_num_ref_frames = readUE();
writeUE(max_num_ref_frames);
const gaps_in_frame_num_value_allowed_flag = readBit(1);
writeBit(gaps_in_frame_num_value_allowed_flag, 1);
const pic_width_in_mbs_minus1 = readUE();
writeUE(pic_width_in_mbs_minus1);
const pic_height_in_map_units_minus1 = readUE();
writeUE(pic_height_in_map_units_minus1);
const frame_mbs_only_flag = readBit(1);
writeBit(frame_mbs_only_flag, 1);
if (frame_mbs_only_flag === 0) {
const mb_adaptive_frame_field_flag = readBit(1);
writeBit(mb_adaptive_frame_field_flag, 1);
}
const direct_8x8_inference_flag = readBit(1);
writeBit(direct_8x8_inference_flag, 1);
const frame_cropping_flag = readBit(1);
writeBit(frame_cropping_flag, 1);
if (frame_cropping_flag) {
const frame_crop_left_offset = readUE();
writeUE(frame_crop_left_offset);
const frame_crop_right_offset = readUE();
writeUE(frame_crop_right_offset);
const frame_crop_top_offset = readUE();
writeUE(frame_crop_top_offset);
const frame_crop_bottom_offset = readUE();
writeUE(frame_crop_bottom_offset);
}
// https://webrtc.googlesource.com/src/+/5f2c9278f35e47ff72eb191669d473b7400c9f3e/common_video/h264/sps_vui_rewriter.cc#283
function addBitstreamRestriction() {
// motion_vectors_over_pic_boundaries_flag: u(1) — Default is 1 when not present.
writeBit(1, 1);
// max_bytes_per_pic_denom: ue(v) — Default is 2 when not present.
writeUE(2);
// max_bits_per_mb_denom: ue(v) — Default is 1 when not present.
writeUE(1);
// log2_max_mv_length_horizontal / vertical — both default to 16.
writeUE(16);
writeUE(16);
// IMPORTANT: max_num_reorder_frames must be 0 for low latency.
writeUE(0);
writeUE(max_num_ref_frames);
}
const vui_parameters_present_flag = readBit(1);
writeBit(1, 1);
// If no VUI exists, write one
if (!vui_parameters_present_flag) {
// aspect_ratio_info_present_flag, overscan_info_present_flag. Both u(1).
writeBit(0, 2);
// video_signal_type_present_flag, u(1) — write 0, ignore color space.
writeBit(0, 1);
// chroma_loc_info_present_flag, timing_info_present_flag,
// nal_hrd_parameters_present_flag, vcl_hrd_parameters_present_flag,
// pic_struct_present_flag — all u(1)
writeBit(0, 5);
// bitstream_restriction_flag: u(1)
writeBit(1, 1);
addBitstreamRestriction();
} else {
// VUI parsing and copying
const aspect_ratio_info_present_flag = readBit(1);
writeBit(aspect_ratio_info_present_flag, 1);
if (aspect_ratio_info_present_flag) {
const aspect_ratio_idc = readU(8);
writeU(aspect_ratio_idc, 8);
if (aspect_ratio_idc === 255) {
const sar_width = readU(16);
writeU(sar_width, 16);
const sar_height = readU(16);
writeU(sar_height, 16);
}
}
const overscan_info_present_flag = readBit(1);
writeBit(overscan_info_present_flag, 1);
if (overscan_info_present_flag) {
const overscan_appropriate_flag = readBit(1);
writeBit(overscan_appropriate_flag, 1);
}
// Read the video signal type, but don't copy it
const video_signal_type_present_flag = readBit(1);
writeBit(0, 1);
if (video_signal_type_present_flag) {
readBit(3); // _video_format
readBit(1); // _video_full_range_flag
const colour_description_present_flag = readBit(1);
if (colour_description_present_flag) {
readU(8); // _colour_primaries
readU(8); // _transfer_characteristics
readU(8); // _matrix_coeffs
}
}
const chroma_loc_info_present_flag = readBit(1);
writeBit(chroma_loc_info_present_flag, 1);
if (chroma_loc_info_present_flag) {
const chroma_sample_loc_type_top_field = readUE();
writeUE(chroma_sample_loc_type_top_field);
const chroma_sample_loc_type_bottom_field = readUE();
writeUE(chroma_sample_loc_type_bottom_field);
}
const timing_info_present_flag = readBit(1);
writeBit(timing_info_present_flag, 1);
if (timing_info_present_flag) {
const num_units_in_tick = readU(32);
writeU(num_units_in_tick, 32);
const time_scale = readU(32);
writeU(time_scale, 32);
const fixed_frame_rate_flag = readBit(1);
writeBit(fixed_frame_rate_flag, 1);
}
const nal_hrd_parameters_present_flag = readBit(1);
writeBit(nal_hrd_parameters_present_flag, 1);
if (nal_hrd_parameters_present_flag) {
// hrd_parameters()
const cpb_cnt_minus1 = readUE();
writeUE(cpb_cnt_minus1);
const bit_rate_scale = readBit(4);
writeBit(bit_rate_scale, 4);
const cpb_size_scale = readBit(4);
writeBit(cpb_size_scale, 4);
for (let i = 0; i <= cpb_cnt_minus1; i++) {
const bit_rate_value_minus1 = readUE();
writeUE(bit_rate_value_minus1);
const cpb_size_value_minus1 = readUE();
writeUE(cpb_size_value_minus1);
const cbr_flag = readBit(1);
writeBit(cbr_flag, 1);
}
const initial_cpb_removal_delay_length_minus1 = readBit(5);
writeBit(initial_cpb_removal_delay_length_minus1, 5);
const cpb_removal_delay_length_minus1 = readBit(5);
writeBit(cpb_removal_delay_length_minus1, 5);
const dpb_output_delay_length_minus1 = readBit(5);
writeBit(dpb_output_delay_length_minus1, 5);
const time_offset_length = readBit(5);
writeBit(time_offset_length, 5);
}
const vcl_hrd_parameters_present_flag = readBit(1);
writeBit(vcl_hrd_parameters_present_flag, 1);
if (vcl_hrd_parameters_present_flag) {
// hrd_parameters()
const cpb_cnt_minus1 = readUE();
writeUE(cpb_cnt_minus1);
const bit_rate_scale = readBit(4);
writeBit(bit_rate_scale, 4);
const cpb_size_scale = readBit(4);
writeBit(cpb_size_scale, 4);
for (let i = 0; i <= cpb_cnt_minus1; i++) {
const bit_rate_value_minus1 = readUE();
writeUE(bit_rate_value_minus1);
const cpb_size_value_minus1 = readUE();
writeUE(cpb_size_value_minus1);
const cbr_flag = readBit(1);
writeBit(cbr_flag, 1);
}
const initial_cpb_removal_delay_length_minus1 = readBit(5);
writeBit(initial_cpb_removal_delay_length_minus1, 5);
const cpb_removal_delay_length_minus1 = readBit(5);
writeBit(cpb_removal_delay_length_minus1, 5);
const dpb_output_delay_length_minus1 = readBit(5);
writeBit(dpb_output_delay_length_minus1, 5);
const time_offset_length = readBit(5);
writeBit(time_offset_length, 5);
}
if (nal_hrd_parameters_present_flag || vcl_hrd_parameters_present_flag) {
const low_delay_hrd_flag = readBit(1);
writeBit(low_delay_hrd_flag, 1);
}
const pic_struct_present_flag = readBit(1);
writeBit(pic_struct_present_flag, 1);
const bitstream_restriction_flag = readBit(1);
writeBit(1, 1);
if (!bitstream_restriction_flag) {
addBitstreamRestriction();
} else {
const motion_vectors_over_pic_boundaries_flag = readBit(1);
writeBit(motion_vectors_over_pic_boundaries_flag, 1);
const max_bytes_per_pic_denom = readUE();
writeUE(max_bytes_per_pic_denom);
const max_bits_per_mb_denom = readUE();
writeUE(max_bits_per_mb_denom);
const log2_max_mv_length_horizontal = readUE();
writeUE(log2_max_mv_length_horizontal);
const log2_max_mv_length_vertical = readUE();
writeUE(log2_max_mv_length_vertical);
readUE(); // _num_reorder_frames
writeUE(0);
readUE(); // _max_dec_frame_buffering
writeUE(max_num_ref_frames);
}
}
writeBit(1, 1); // rbsp_stop_one_bit
writer.flush();
return writer.toBuffer();
}
services/discord-gateway/src/goLive/StreamConnection.ts
-45
View File
@@ -1,45 +0,0 @@
/**
* StreamConnection — GoLive stream connection (screen share).
* Ported from @dank074/discord-video-stream StreamConnection.js.
*/
import { BaseMediaConnection } from "./BaseMediaConnection.js";
import { VoiceOpCodes } from "./VoiceOpCodes.js";
export class StreamConnection extends BaseMediaConnection {
_streamKey: string | null = null;
_serverId: string | null = null;
setSpeaking(speaking: boolean): void {
if (!this.webRtcParams) throw new Error("WebRTC connection not ready");
this.sendOpcode(VoiceOpCodes.SPEAKING, {
delay: 0,
speaking: speaking ? 2 : 0,
ssrc: this.webRtcParams.audioSsrc,
});
}
get daveChannelId(): string {
if (this._serverId === null) {
throw new Error("Server ID not set (this shouldn't happen)");
}
const channelId = BigInt(this._serverId) - 1n;
return channelId.toString();
}
get serverId(): string | null {
return this._serverId;
}
set serverId(id: string | null) {
this._serverId = id;
}
get streamKey(): string | null {
return this._streamKey;
}
set streamKey(value: string | null) {
this._streamKey = value;
}
}
services/discord-gateway/src/goLive/Streamer.ts
-351
View File
@@ -1,351 +0,0 @@
/**
* 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 };
services/discord-gateway/src/goLive/VideoStream.ts
-20
View File
@@ -1,20 +0,0 @@
/**
* VideoStream — feeds encoded H264 frames into the WebRTC connection.
* Ported from @dank074/discord-video-stream VideoStream.js.
*/
import { BaseMediaStream } from "./BaseMediaStream.js";
import type { WebRtcConnWrapper } from "./WebRtcWrapper.js";
export class VideoStream extends BaseMediaStream {
_conn: WebRtcConnWrapper;
constructor(conn: WebRtcConnWrapper, noSleep = false) {
super("video", noSleep);
this._conn = conn;
}
async _sendFrame(frame: Buffer, frametime: number): Promise<void> {
this._conn.sendVideoFrame(frame, frametime);
}
}
services/discord-gateway/src/goLive/VoiceConnection.ts
-25
View File
@@ -1,25 +0,0 @@
/**
* VoiceConnection — guild/DM voice channel GoLive connection.
* Ported from @dank074/discord-video-stream VoiceConnection.js.
*/
import { BaseMediaConnection } from "./BaseMediaConnection.js";
import type { StreamConnection } from "./StreamConnection.js";
export class VoiceConnection extends BaseMediaConnection {
streamConnection: StreamConnection | null = null;
get daveChannelId(): string {
return this.channelId;
}
get serverId(): string | null {
// for guild vc it is the guild id, for dm voice it is the channel id
return this.guildId ?? this.channelId;
}
stop(): void {
super.stop();
this.streamConnection?.stop();
}
}
services/discord-gateway/src/goLive/VoiceOpCodes.ts
-38
View File
@@ -1,38 +0,0 @@
/** Discord voice WebSocket opcodes — ported from @dank074/discord-video-stream. */
export enum VoiceOpCodes {
IDENTIFY = 0,
SELECT_PROTOCOL = 1,
READY = 2,
HEARTBEAT = 3,
SELECT_PROTOCOL_ACK = 4,
SPEAKING = 5,
HEARTBEAT_ACK = 6,
RESUME = 7,
HELLO = 8,
RESUMED = 9,
CLIENTS_CONNECT = 11,
VIDEO = 12,
CLIENT_DISCONNECT = 13,
SESSION_UPDATE = 14,
MEDIA_SINK_WANTS = 15,
VOICE_BACKEND_VERSION = 16,
CHANNEL_OPTIONS_UPDATE = 17,
FLAGS = 18,
SPEED_TEST = 19,
PLATFORM = 20,
DAVE_PREPARE_TRANSITION = 21,
DAVE_EXECUTE_TRANSITION = 22,
DAVE_TRANSITION_READY = 23,
DAVE_PREPARE_EPOCH = 24,
MLS_INVALID_COMMIT_WELCOME = 31,
}
/** Binary voice WebSocket opcodes (DAVE / MLS). */
export enum VoiceOpCodesBinary {
MLS_EXTERNAL_SENDER = 25,
MLS_KEY_PACKAGE = 26,
MLS_PROPOSALS = 27,
MLS_COMMIT_WELCOME = 28,
MLS_ANNOUNCE_COMMIT_TRANSITION = 29,
MLS_WELCOME = 30,
}
services/discord-gateway/src/goLive/WebRtcWrapper.ts
-223
View File
@@ -1,223 +0,0 @@
/**
* WebRTC connection wrapper for GoLive — ported from
* @dank074/discord-video-stream WebRtcWrapper.js, with the media stack
* (packetizers, RTCP SR/NACK, pacing) provided by the libdatachannel-min
* binding instead of node-datachannel's JS-exposed media classes.
*/
import {
H264Helpers,
H264NalUnitTypes,
splitNalu,
startCode3,
} from "./AnnexBHelper.js";
import { CodecPayloadType } from "./CodecPayloadType.js";
import type { NativePeerConnection, NativeTrack } from "./native.js";
import { loadNative } from "./native.js";
import { rewriteSPSVUI } from "./SPSVUIRewriter.js";
import { normalizeVideoCodec } from "./utils.js";
export type WebRtcVideoCodec = "H264" | "H265" | "VP8" | "VP9" | "AV1";
export interface WebRtcParams {
address: string;
port: number;
audioSsrc: number;
videoSsrc: number;
rtxSsrc: number;
supportedEncryptionModes: string[];
}
/** Minimal surface of the media connection that WebRtcWrapper drives. */
export interface VideoAttribute {
fps: number;
width: number;
height: number;
}
export interface MediaConnectionLike {
daveReady: boolean;
daveSession: {
encryptOpus(frame: Buffer): Buffer;
encrypt(mediaType: number, codec: number, frame: Buffer): Buffer;
} | null;
webRtcParams: WebRtcParams | null;
setSpeaking(speaking: boolean): void;
setVideoAttributes(enabled: boolean, attr?: VideoAttribute): void;
}
/** Media types used by DAVE encryption (from @dank074). */
export enum DaveMediaType {
AUDIO = 0,
VIDEO = 1,
}
/** DAVE codec ids (from @dank074). */
export enum DaveCodec {
UNKNOWN = 0,
VP8 = 2,
VP9 = 3,
H264 = 4,
H265 = 5,
AV1 = 6,
}
export class WebRtcConnWrapper {
private _mediaConn: MediaConnectionLike;
private _webRtcConn: NativePeerConnection | null = null;
private _audioTrack: NativeTrack | null = null;
private _videoTrack: NativeTrack | null = null;
private _videoCodec: WebRtcVideoCodec | null = null;
private _videoFrameLog = 0;
/** Assigned by BaseMediaConnection to send the gathered SDP to Discord. */
onLocalDescription: ((sdp: string) => void) | null = null;
constructor(mediaConn: MediaConnectionLike) {
this._mediaConn = mediaConn;
}
initWebRtc(): NativePeerConnection {
const native = loadNative();
this._webRtcConn = new native.PeerConnection({
iceServers: ["stun:stun.l.google.com:19302"],
});
// Track mids must match @dank074: "0" audio, "1" video.
this._audioTrack = this._webRtcConn.addTrack("0", "audio");
this._videoTrack = this._webRtcConn.addTrack("1", "video");
return this._webRtcConn;
}
close(): void {
this._webRtcConn?.close();
this._webRtcConn = null;
}
get webRtcConn(): NativePeerConnection | null {
return this._webRtcConn;
}
get ready(): boolean {
return this._webRtcConn?.state() === "connected";
}
get mediaConnection(): MediaConnectionLike {
return this._mediaConn;
}
sendAudioFrame(frame: Buffer, frametime: number): void {
if (!this.ready || !this._audioTrack) return;
const clockRate = CodecPayloadType.opus.clockRate;
if (this.mediaConnection.daveReady && this.mediaConnection.daveSession) {
frame = this.mediaConnection.daveSession.encryptOpus(frame);
}
this._audioTrack.sendFrame(frame);
this._audioTrack.addTimestamp(Math.round((frametime * clockRate) / 1000));
}
sendVideoFrame(frame: Buffer, frametime: number): void {
if (!this.ready || !this._videoTrack) {
if (this._videoFrameLog === 0) {
console.log(
`[goLive:WebRtc] sendVideoFrame DROPPED ready=${this.ready} track=${this._videoTrack !== null}`,
);
this._videoFrameLog++;
}
return;
}
const clockRate = CodecPayloadType[this._videoCodec ?? "H264"].clockRate;
if (this._videoCodec === "H264") {
let spsRewritten = false;
const nalus = splitNalu(frame).map((el) => {
if (H264Helpers.getUnitType(el) === H264NalUnitTypes.SPS) {
spsRewritten = true;
return rewriteSPSVUI(el);
}
return el;
});
if (spsRewritten)
frame = Buffer.concat(nalus.flatMap((el) => [startCode3, el]));
}
if (this.mediaConnection.daveReady && this.mediaConnection.daveSession) {
let daveCodec = DaveCodec.UNKNOWN;
switch (this._videoCodec) {
case "H264":
daveCodec = DaveCodec.H264;
break;
case "H265":
daveCodec = DaveCodec.H265;
break;
case "VP8":
daveCodec = DaveCodec.VP8;
break;
case "VP9":
daveCodec = DaveCodec.VP9;
break;
case "AV1":
daveCodec = DaveCodec.AV1;
break;
default:
break;
}
frame = this.mediaConnection.daveSession.encrypt(
DaveMediaType.VIDEO,
daveCodec,
frame,
);
}
this._videoTrack.sendFrame(frame);
this._videoTrack.addTimestamp(Math.round((frametime * clockRate) / 1000));
this._videoFrameLog++;
if (this._videoFrameLog === 1 || this._videoFrameLog % 30 === 0) {
console.log(
`[goLive:WebRtc] sendVideoFrame #${this._videoFrameLog} bytes=${frame.length} ready=${this.ready}`,
);
}
}
setPacketizer(videoCodec: string): void {
if (!this.mediaConnection.webRtcParams) {
throw new Error("WebRTC connection not ready");
}
const { audioSsrc, videoSsrc } = this.mediaConnection.webRtcParams;
console.log(
`[goLive:WebRtc] setPacketizer(${videoCodec}) audioSsrc=${audioSsrc} videoSsrc=${videoSsrc} rtxSsrc=${this.mediaConnection.webRtcParams.rtxSsrc}`,
);
this._videoCodec = normalizeVideoCodec(videoCodec);
// Audio packetizer: opus 120 @ 48kHz, playout delay ext id 5 (like @dank074)
this._audioTrack?.setPacketizer(
"audio",
audioSsrc,
CodecPayloadType.opus.payload_type,
CodecPayloadType.opus.clockRate,
5,
0,
1,
);
// Video packetizer: H264/H265/AV1 with their payload types
const codecEntry = CodecPayloadType[this._videoCodec];
if (!codecEntry) {
throw new Error(`Packetizer not implemented for ${this._videoCodec}`);
}
const nativeKind =
this._videoCodec === "H264"
? "h264"
: this._videoCodec === "H265"
? "h265"
: this._videoCodec === "AV1"
? "av1"
: (() => {
throw new Error(
`Packetizer not implemented for ${this._videoCodec}`,
);
})();
this._videoTrack?.setPacketizer(
nativeKind,
videoSsrc,
codecEntry.payload_type,
codecEntry.clockRate,
5,
0,
10,
);
}
}
services/discord-gateway/src/goLive/index.ts
-19
View File
@@ -1,19 +0,0 @@
/**
* goLive public API — re-exports the ported @dank074 modules.
* Drop-in replacement for `@dank074/discord-video-stream` in
* screenShareController.ts.
*/
export { AudioStream } from "./AudioStream.js";
export { BaseMediaConnection } from "./BaseMediaConnection.js";
export { BaseMediaStream } from "./BaseMediaStream.js";
export { CodecPayloadType } from "./CodecPayloadType.js";
export { demux } from "./Demuxer.js";
export { Encoders } from "./Encoders.js";
export { playStream, prepareStream } from "./prepareStream.js";
export { StreamConnection } from "./StreamConnection.js";
export { Streamer } from "./Streamer.js";
export { normalizeVideoCodec } from "./utils.js";
export { VideoStream } from "./VideoStream.js";
export { VoiceConnection } from "./VoiceConnection.js";
export { WebRtcConnWrapper } from "./WebRtcWrapper.js";
services/discord-gateway/src/goLive/native.ts
-116
View File
@@ -1,116 +0,0 @@
/**
* Loader + typings for the minimal libdatachannel N-API binding
* (native/libdatachannel-min). The binding exposes ONLY what GoLive needs:
* PeerConnection, DataChannel, Track (raw RTP + media packetizer chain).
*
* The .node file is built by node-gyp against libdatachannel 0.24.0 (built
* from source — nixpkgs 0.24.1 is glibc-incompatible with this host). It is
* NOT shipped via npm; the Nix flake builds it as part of the gateway.
*/
export interface NativeTrack {
/** Send a RAW RTP/RTCP packet (no media handler installed). */
send(buffer: Uint8Array): void;
/** Send an ENCODED frame; the packetizer chain turns it into RTP. */
sendFrame(buffer: Uint8Array): void;
/** Advance the packetizer RTP timestamp by delta (clock-rate units). */
addTimestamp(delta: number): void;
/** Install the media-handler chain (packetizer → RTCP SR → NACK → pacing). */
setPacketizer(
kind: "audio" | "h264" | "h265" | "av1",
ssrc: number,
payloadType: number,
clockRate: number,
playoutDelayId: number,
playoutDelayMin: number,
playoutDelayMax: number,
): void;
isOpen(): boolean;
close(): void;
}
export interface NativePeerConnection {
/** mid must be "0" (audio) or "1" (video) — matches @dank074's track defs. */
addTrack(mid: string, kind: "audio" | "video"): NativeTrack;
/** Resolves with the full SDP (incl. candidates) after gathering completes. */
createOffer(): Promise<string>;
/** Resolves with the auto-generated answer SDP. */
createAnswer(offerSdp: string): Promise<string>;
setRemoteDescription(sdp: string, type: "offer" | "answer"): void;
state(): string;
close(): void;
onStateChange(cb: (state: string) => void): void;
}
export interface NativeBinding {
PeerConnection: new (config: {
iceServers: string[];
}) => NativePeerConnection;
DataChannel: unknown;
Track: unknown;
}
let cached: NativeBinding | null = null;
/** Load the native binding. Throws only if the .node is truly missing —
* callers (screen share) guard with `isNativeAvailable()`. */
export function loadNative(): NativeBinding {
if (cached) return cached;
// Resolve relative to this file: src/goLive/ → native/libdatachannel-min/
const candidates = [
new URL(
"../../native/libdatachannel-min/build/Release/datachannel_min.node",
import.meta.url,
),
new URL(
"../../../native/libdatachannel-min/build/Release/datachannel_min.node",
import.meta.url,
),
];
let lastErr: unknown;
for (const url of candidates) {
try {
// @ts-expect-error — .node modules are not typed; dynamic require via file URL
const mod = process.dlopen ? null : null;
void mod;
const nativePath = url.pathname;
// eslint-disable-next-line @typescript-eslint/no-require-imports
const req = createRequire(import.meta.url);
const binding = req(nativePath) as NativeBinding;
if (typeof binding.PeerConnection === "function") {
cached = binding;
return binding;
}
} catch (e) {
lastErr = e;
}
}
// Fallback: plain relative require (tsx / jest environments)
try {
const req = createRequire(import.meta.url);
const binding = req(
"../../native/libdatachannel-min/build/Release/datachannel_min.node",
) as NativeBinding;
if (typeof binding.PeerConnection === "function") {
cached = binding;
return binding;
}
} catch (e) {
lastErr = e;
}
throw new Error(
`libdatachannel-min native binding not built (${String(lastErr)}). Run: cd native/libdatachannel-min && npx node-gyp rebuild`,
);
}
import { createRequire } from "node:module";
/** True when the native binding is built — screen share stays disabled otherwise. */
export function isNativeAvailable(): boolean {
try {
loadNative();
return true;
} catch {
return false;
}
}
services/discord-gateway/src/goLive/prepareStream.ts
-477
View File
@@ -1,477 +0,0 @@
/**
* prepareStream & playStream — ported from @dank074/discord-video-stream
* newApi.js (Encoders/prepareStream/playStream), but uses `child_process.spawn`
* + ffmpeg CLI args directly instead of fluent-ffmpeg + node-av.
*
* Replaces the @dank074 video pipeline entirely:
* input (URL or Readable) → ffmpeg spawn → H264 AnnexB frames
* → Demuxer stream → VideoStream/AudioStream → WebRtcConnWrapper
*/
import { type ChildProcess, spawn } from "node:child_process";
import { existsSync, readdirSync } from "node:fs";
import { join } from "node:path";
import { PassThrough, type Readable } from "node:stream";
import { AudioStream } from "./AudioStream.js";
import { demux } from "./Demuxer.js";
import { type EncoderSettings, Encoders } from "./Encoders.js";
import { VideoStream } from "./VideoStream.js";
import type { WebRtcConnWrapper } from "./WebRtcWrapper.js";
export interface PrepareStreamResult {
command: ChildProcess;
output: PassThrough;
encoder: () => Record<string, EncoderSettings>;
options: Record<string, unknown>;
videoCodec: string;
width: number;
height: number;
frameRate?: number;
includeAudio: boolean;
/** Container the encoder muxes to: "nut" (audio-capable) or "h264" (raw). */
format: "nut" | "h264";
}
function isFiniteNonZero(n: unknown): n is number {
return typeof n === "number" && !!n && Number.isFinite(n);
}
const DEFAULT_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.36",
Connection: "keep-alive",
};
/** Resolve ffmpeg binary (env override → PATH → Nix store ffmpeg-headless). */
function resolveFfmpeg(): string {
if (process.env.FFMPEG_PATH && existsSync(process.env.FFMPEG_PATH)) {
return process.env.FFMPEG_PATH;
}
const store = "/nix/store";
if (existsSync(store)) {
const entries = readdirSync(store);
for (const entry of entries) {
if (!entry.includes("ffmpeg-headless-")) continue;
const candidate = join(store, entry, "bin", "ffmpeg");
if (existsSync(candidate)) return candidate;
}
}
return "ffmpeg";
}
const FFMPEG_BIN = resolveFfmpeg();
/**
* prepareStream — build an ffmpeg command (as spawn args + PassThrough output)
* that transcodes the input into a pipe we can demux. Mirrors @dank074's
* prepareStream but produces a raw spawn instead of a fluent-ffmpeg command.
*/
export function prepareStream(
input: string | Readable,
options: Record<string, unknown> = {},
): PrepareStreamResult {
const mergedOptions = {
noTranscoding: false,
width: isFiniteNonZero(options.width)
? Math.round(options.width as number)
: -2,
height: isFiniteNonZero(options.height)
? Math.round(options.height as number)
: -2,
frameRate:
isFiniteNonZero(options.frameRate) && (options.frameRate as number) > 0
? options.frameRate
: undefined,
videoCodec: (options.videoCodec as string) ?? "H264",
bitrateVideo:
isFiniteNonZero(options.bitrateVideo) &&
(options.bitrateVideo as number) > 0
? Math.round(options.bitrateVideo as number)
: 5000,
bitrateVideoMax:
isFiniteNonZero(options.bitrateVideoMax) &&
(options.bitrateVideoMax as number) > 0
? Math.round(options.bitrateVideoMax as number)
: 7000,
bitrateAudio:
isFiniteNonZero(options.bitrateAudio) &&
(options.bitrateAudio as number) > 0
? Math.round(options.bitrateAudio as number)
: 128,
includeAudio: options.includeAudio ?? true,
encoder:
(options.encoder as () => Record<string, EncoderSettings>) ??
Encoders.software(),
customHeaders: {
...DEFAULT_HEADERS,
...(options.customHeaders as Record<string, string> | undefined),
},
customInputOptions: (options.customInputOptions as string[]) ?? [],
customFfmpegFlags: (options.customFfmpegFlags as string[]) ?? [],
minimizeLatency: options.minimizeLatency ?? false,
// realtime: throttle ffmpeg's INPUT read to 1x so the encoder tracks
// wall-clock instead of slurping a VOD at network speed. Without this,
// a YouTube screen share downloads the whole clip fast and the encoder
// produces a ~10x frame backlog that the demuxer buffers unboundedly —
// the sender paces at 30fps but always emits the OLDEST buffered frames,
// so the viewer sees frozen/laggy video while audio (tiny, jitter-
// buffer recoverable) stays smooth. That is the "video stuck, voice
// normal" symptom. `-re` caps the pipeline at 1x end-to-end. Default on
// because this prepareStream is only used for screen share (VOD URLs).
realtime: options.realtime ?? true,
};
const output = new PassThrough();
const args: string[] = [
"-hide_banner",
"-loglevel",
"error",
// Real-time throttle: read the input at 1x so the encoder paces with
// wall-clock and does NOT force-duplicate frames to fill -r 30. This
// applies to BOTH URL and live-Pipe (Readable) inputs: a screen-share
// pipe (yt-dlp merge → ffmpeg → stdout) is delivered at NETWORK speed
// (bursts, stalls) and is NOT self-paced — without -re the encoder slurps
// it instantly and, when the merge stalls, x264 -r 30 repeats the last
// held frame ~30x → the viewer sees ~1fps while WebRTC still pushes
// 30fps. -re reads the pipe at the stream's native PTS rate so each
// output frame is a fresh picture. (Previous builds only added -re for
// string URLs, so the screen-share pipe got none — root of the 1fps
// symptom.)
...(mergedOptions.realtime ? ["-re"] : []),
...(typeof input === "string" ? ["-i", input] : ["-i", "pipe:0"]),
...mergedOptions.customInputOptions,
];
if (mergedOptions.minimizeLatency) {
args.push("-fflags", "nobuffer", "-analyzeduration", "0");
}
if (typeof input === "string" && input.startsWith("http")) {
const headerStr = Object.entries(mergedOptions.customHeaders)
.map(([k, v]) => `${k}: ${v}`)
.join("\r\n");
args.push(
"-headers",
headerStr,
"-reconnect",
"1",
"-reconnect_at_eof",
"1",
"-reconnect_streamed",
"1",
"-reconnect_delay_max",
"4294",
);
}
// Video
args.push("-map", "0:v:0");
if (mergedOptions.noTranscoding) {
args.push("-c:v", "copy");
} else {
args.push(`-vf`, `scale=${mergedOptions.width}:${mergedOptions.height}`);
if (mergedOptions.frameRate)
args.push("-r", String(mergedOptions.frameRate));
const enc = mergedOptions.encoder()[mergedOptions.videoCodec];
if (!enc)
throw new Error(
`Encoder settings not specified for ${mergedOptions.videoCodec}`,
);
// Encoder options are declared as single strings like "-forced-idr 1";
// spawn needs each flag and value as separate argv entries.
const encOptions = enc.options.flatMap((opt) =>
opt.split(/\s+/).filter(Boolean),
);
args.push(
"-b:v",
`${mergedOptions.bitrateVideo}k`,
"-maxrate:v",
`${mergedOptions.bitrateVideoMax}k`,
"-bufsize:v",
`${Math.round(mergedOptions.bitrateVideo / 2)}k`,
"-bf",
"0",
"-pix_fmt",
"yuv420p",
"-force_key_frames",
"expr:gte(t,n_forced*1)",
"-c:v",
enc.name,
...encOptions,
...(enc.globalOptions ?? []).flatMap((opt) =>
opt.split(/\s+/).filter(Boolean),
),
);
}
// Audio: transcode to libopus. NUT muxer on stdout carries video (H264
// AnnexB) + audio (Ogg Opus) as ONE stream into the Demuxer, which re-splits
// them via a child ffmpeg -f nut -i pipe:0 -c:v copy -f h264 pipe:1 ... . The
// Demuxer's start-code scan runs on THAT child ffmpeg's stdout (pure H264),
// NOT on NUT — so NAL type 5 (IDR) is parsed correctly. (Outputting raw
// h264+opus on two pipes directly was tried and broke: the audio pipe was
// never attached to the demuxer's input, so audio RTP never flowed.)
if (mergedOptions.includeAudio) {
args.push(
"-map",
"0:a:0?",
"-c:a",
"libopus",
"-b:a",
`${mergedOptions.bitrateAudio}k`,
"-ar",
"48000",
"-ac",
"2",
);
} else {
args.push("-an");
}
// NUT muxer carries video+audio; the raw h264 muxer cannot ("h264 muxer
// does not support any stream of type audio" -> header write fails ->
// empty stdout -> black tile). Audio delivery requires NUT.
const outFormat = mergedOptions.includeAudio ? "nut" : "h264";
args.push("-f", outFormat, "pipe:1");
const isUrl = typeof input === "string";
const proc: ChildProcess = isUrl
? spawn(FFMPEG_BIN, args, { stdio: ["ignore", "pipe", "pipe"] })
: spawn(FFMPEG_BIN, args, { stdio: ["pipe", "pipe", "pipe"] });
if (proc.stdin && !isUrl) {
// Race guard: the merge ffmpeg may have already exited (transient 403
// or stream death) before this function attaches its listeners — the
// input's 'end'/'error' events then fire into the void and the encoder
// stdin NEVER receives EOF, leaving an encoder that waits forever and a
// screen share that shows a black tile with zero frames. Check the
// terminal state eagerly and EOF the encoder immediately.
if (input.readableEnded || input.destroyed) {
proc.stdin.end();
} else {
input.on("data", (chunk: Buffer) => proc.stdin?.write(chunk));
input.on("end", () => proc.stdin?.end());
input.on("error", () => proc.stdin?.destroy());
}
}
// Safety: proc may error before playStream attaches a demux listener on
// `output`. A no-op listener here prevents an unhandled 'error' event
// on the PassThrough from crashing the gateway on ffmpeg spawn failure.
output.on("error", () => {});
proc.stdout?.pipe(output);
proc.stderr?.on("data", () => {
/* swallow ffmpeg stderr */
});
proc.on("error", (err) => {
// spawn failed (e.g. ffmpeg missing). If someone is consuming output
// (demux attaches an 'error' listener) propagate; otherwise just end.
if (output.listenerCount("error") > 0) {
output.destroy(err);
} else {
output.end();
}
});
proc.on("close", () => {
output.end();
});
return {
command: proc,
output,
encoder: mergedOptions.encoder,
options: mergedOptions,
videoCodec: mergedOptions.videoCodec,
width: mergedOptions.width,
height: mergedOptions.height,
frameRate: mergedOptions.frameRate,
includeAudio: !!mergedOptions.includeAudio,
format: outFormat,
};
}
export interface PlayStreamOptions {
type?: "go-live" | "video";
format?: string;
width?: number | ((v: unknown) => number);
height?: number | ((v: unknown) => number);
frameRate?: number | ((v: unknown) => number);
readrateInitialBurst?: number;
streamPreview?: boolean;
}
/**
* playStream — demux the prepareStream output and pipe frames into the
* WebRTC connection's video/audio streams. Resolves when the video stream
* ends (natural EOF or the ffmpeg command is killed via cleanup/stop).
*/
export async function playStream(
prepared: PrepareStreamResult,
streamer: { createStream: () => Promise<WebRtcConnWrapper> },
options: PlayStreamOptions = {},
): Promise<void> {
const conn = await streamer.createStream();
console.log("[goLive:playStream] createStream resolved");
const {
video,
audio,
close: demuxClose,
} = await demux(prepared.output, {
format: options.format ?? prepared.format ?? "nut",
frameRate:
typeof options.frameRate === "number" ? options.frameRate : undefined,
});
console.log(
`[goLive:playStream] demux done codec=${video?.codecName ?? "?"} ${video?.width ?? 0}x${video?.height ?? 0} fps=${video ? video.framerate_num / video.framerate_den || 30 : 30} audio=${audio?.codecName ?? "none"}`,
);
if (!video) throw new Error("No video stream in media");
conn.setPacketizer(video.codecName);
conn.mediaConnection.setSpeaking(true);
console.log(
`[goLive:playStream] setPacketizer(${video.codecName}) + setSpeaking done`,
);
const w =
typeof options.width === "function"
? options.width(video)
: (options.width ?? video.width);
const h =
typeof options.height === "function"
? options.height(video)
: (options.height ?? video.height);
const fr =
typeof options.frameRate === "function"
? options.frameRate(video)
: (options.frameRate ??
(video.framerate_num / video.framerate_den || 30));
conn.mediaConnection.setVideoAttributes(true, {
width: Math.round(w),
height: Math.round(h),
fps: Math.round(fr),
});
const vStream = new VideoStream(conn);
video.stream.pipe(vStream);
// Audio: Discord's GoLive pipeline expects RTP on the audio SSRC too —
// a video-only stream (zero audio packets) shows a static tile/thumbnail
// instead of live video. Pipe opus frames from the demuxer (silence is
// injected at the encoder when the source has no audio track).
let aStream: AudioStream | undefined;
if (audio) {
aStream = new AudioStream(conn);
audio.stream.pipe(aStream);
console.log(
`[goLive:playStream] audio stream attached (${audio.codecName})`,
);
// NOTE: NO syncStream wiring here. Upstream dank sets
// `vStream.syncStream = aStream` because node-av provides real PTS from the
// NUT container, so ptsDelta() is meaningful (both streams in media time).
// Our raw-h264 demuxer synthesizes PTS per-stream from frame indexes in
// DIFFERENT timebases (video 1/fps, audio 1/48000). If audio starts late
// (ffmpeg audio init, Ogg header), ptsDelta stays positive forever →
// isAhead() → video sleeps in a loop → video freezes after ~1s. Per-stream
// sleep-PTS pacing alone keeps both at 1000ms/s, which is correct without
// a shared clock. (If real PTS is ever added, re-enable syncStream.)
}
const cleanup = () => {
try {
prepared.command.kill("SIGTERM");
} catch {
/* already dead */
}
demuxClose();
try {
conn.mediaConnection.setSpeaking(false);
conn.mediaConnection.setVideoAttributes(false);
} catch {
/* connection already torn down */
}
};
// First-frame watchdog: if the encoder never delivers a single frame
// (dead merge input, empty stream, codec mismatch), fail fast instead of
// "playing" a black tile forever. The demuxer resolves with fallback
// metadata even when no frame ever arrives, so this timeout is the only
// place that detects "started but nothing flowing".
let firstFrameTimer: NodeJS.Timeout | null = null;
let gotFirstFrame = false;
const firstFrame = new Promise<void>((resolve, reject) => {
firstFrameTimer = setTimeout(() => {
if (!gotFirstFrame) {
cleanup();
reject(
new Error(
"No video frames within 10s of stream start — input stream failed",
),
);
}
}, 10000);
video.stream.once("data", () => {
gotFirstFrame = true;
if (firstFrameTimer) clearTimeout(firstFrameTimer);
resolve();
});
});
return new Promise<void>((resolve, reject) => {
let settled = false;
const settle = (fn: () => void) => () => {
if (settled) return;
settled = true;
if (firstFrameTimer) clearTimeout(firstFrameTimer);
fn();
};
vStream.once("finish", () => {
settle(() => {
cleanup();
if (!gotFirstFrame) {
reject(new Error("Screen video stream ended without any frame"));
} else {
resolve();
}
})();
});
vStream.once("error", () => {
settle(() => {
cleanup();
if (!gotFirstFrame) {
reject(new Error("Screen video stream errored before first frame"));
} else {
resolve();
}
})();
});
// The stream may end without ever producing a frame (input was
// silently dead) — surface that instead of resolving "successfully".
video.stream.once("end", () => {
settle(() => {
cleanup();
if (!gotFirstFrame) {
reject(new Error("Screen video stream ended before any frame"));
} else {
resolve();
}
})();
});
// Watchdog timeout: no frame arrived within 10s — fail fast instead of
// "playing" a black tile forever. cleanup() kills the encoder so the
// vStream finish/error handlers above still fire, but the settled guard
// ensures this rejection wins.
firstFrame.catch((err) => {
settle(() => {
cleanup();
reject(err);
})();
});
});
}
export { Encoders };
services/discord-gateway/src/goLive/utils.ts
-82
View File
@@ -1,82 +0,0 @@
/** GoLive helpers — ported from @dank074/discord-video-stream/utils.js. */
export function normalizeVideoCodec(
codec: string,
): "H264" | "H265" | "VP8" | "VP9" | "AV1" {
if (/H\.?264|AVC/i.test(codec)) return "H264";
if (/H\.?265|HEVC/i.test(codec)) return "H265";
if (/VP(8|9)/i.test(codec)) return codec.toUpperCase() as "VP8" | "VP9";
if (/AV1/i.test(codec)) return "AV1";
throw new Error(`Unknown codec: ${codec}`);
}
/**
* The available video streams are sent by the client on connection to the
* voice gateway using OpCode Identify (0); the server replies with the ssrc
* and rtxssrc for each available stream using OpCode Ready (2). RID
* distinguishes simulcast streams of the same video source — we only send one
* quality stream, so a single entry is hardcoded.
*/
export const STREAMS_SIMULCAST = [{ type: "screen", rid: "100", quality: 100 }];
export const max_int16bit = 2 ** 16;
export const max_int32bit = 2 ** 32;
export function isFiniteNonZero(n: unknown): n is number {
return typeof n === "number" && !!n && Number.isFinite(n);
}
export interface ParsedStreamKey {
type: "guild" | "call";
channelId: string;
guildId: string | null;
userId: string;
}
export function parseStreamKey(streamKey: string): ParsedStreamKey {
const streamKeyArray = streamKey.split(":");
const type = streamKeyArray.shift();
if (type !== "guild" && type !== "call") {
throw new Error(`Invalid stream key type: ${type}`);
}
if (
(type === "guild" && streamKeyArray.length < 3) ||
(type === "call" && streamKey.length < 2)
) {
throw new Error(`Invalid stream key: ${streamKey}`);
}
let guildId: string | null = null;
if (type === "guild") {
guildId = streamKeyArray.shift() ?? null;
}
const channelId = streamKeyArray.shift();
const userId = streamKeyArray.shift();
if (!channelId || !userId) {
throw new Error(`Invalid stream key: ${streamKey}`);
}
return { type, channelId, guildId, userId };
}
export function generateStreamKey(
type: "guild" | "call",
guildId: string | null,
channelId: string,
userId: string,
): string {
return `${type}${type === "guild" ? `:${guildId}` : ""}:${channelId}:${userId}`;
}
export interface VoiceChannelLike {
type: string;
id: string;
guildId?: string | null;
}
export function isVoiceChannel(channel: VoiceChannelLike): boolean {
return (
channel.type === "DM" ||
channel.type === "GROUP_DM" ||
channel.type === "GUILD_STAGE_VOICE" ||
channel.type === "GUILD_VOICE"
);
}
services/discord-gateway/src/modules/command-handler/commandHandler.ts
+1 -6
View File
@@ -83,12 +83,7 @@ export class CommandHandler {
// Create domain-specific handlers with their dependencies
this.voiceHandler = new VoiceHandler(client, voiceController);
this.mediaHandler = new MediaHandler(client, () =>
voiceController.getStatus(),
);
// Give media handler access to disconnect/reconnect voice around screen
// share (GoLive needs its own WebRTC connection).
this.mediaHandler.setVoiceController(() => voiceController);
this.mediaHandler = new MediaHandler();
this.guildHandler = new GuildHandler(client);
this.moderationHandler = new ModerationHandler(client);
services/discord-gateway/src/modules/command-handler/media.handler.ts
+6 -115
View File
@@ -1,6 +1,5 @@
import { randomUUID } from "node:crypto";
import { StreamType } from "@discordjs/voice";
import type { Client } from "discord.js-selfbot-v13";
import type { CommandMessage, CommandReply } from "../../shared/index.js";
import { createChildLogger } from "../../shared/logger/index.js";
import {
@@ -13,10 +12,6 @@ import type {
MediaQueueItem,
} from "../voice-recording/mediaTypes.js";
import { discordPlayer } from "../voice-recording/player.js";
import {
ScreenShareController,
type ScreenShareVoiceStatus,
} from "../voice-recording/screenShareController.js";
import { setMediaStatusKey } from "./mediaStatusSink.js";
// ---------------------------------------------------------------------------
@@ -84,17 +79,8 @@ function buildStatusPayload(): MediaStatusPayload {
export class MediaHandler {
private logger = createChildLogger("media-handler");
private screenController: ScreenShareController | null = null;
private screenPlayback: { stop(): void } | null = null;
constructor(
private readonly client: Client | null = null,
private readonly getVoiceStatus: () => ScreenShareVoiceStatus = () => ({
connected: false,
activeGuildId: null,
activeChannelId: null,
}),
) {
constructor() {
// Register auto-advance on natural track end. advanceQueue mutates the
// module-level currentTrackItem/queue, so we must re-publish the status
// key afterward: otherwise the backend's Redis `media:status` cache (and
@@ -108,31 +94,11 @@ export class MediaHandler {
});
}
/**
* Give MediaHandler access to the VoiceController so screen-share can
* disconnect/reconnect the @discordjs audio connection around a GoLive
* stream (Discord allows only one voice session per user).
*/
private voiceControllerAccessor:
| (() => {
disconnectGuild(guildId: string): Promise<void>;
connect(guildId: string, channelId: string): Promise<unknown>;
getStatus(): {
activeGuildId: string | null;
activeChannelId: string | null;
};
})
| null = null;
setVoiceController(accessor: typeof this.voiceControllerAccessor): void {
this.voiceControllerAccessor = accessor;
}
/**
* Persist the latest media state to Redis so the backend/frontend see queue
* advances that happen outside a command (natural track end, screen-share
* done). CommandHandler owns the Redis status-key writes for command-triggered
* changes; this covers the side-effect-only path.
* advances that happen outside a command (natural track end). CommandHandler
* owns the Redis status-key writes for command-triggered changes; this
* covers the side-effect-only path.
*/
private publishStatus(): void {
try {
@@ -152,7 +118,6 @@ export class MediaHandler {
async handleMediaQueue(cmd: CommandMessage): Promise<CommandReply<unknown>> {
// Accept both `url` (canonical) and `source` (legacy FE) for resilience.
const url = String(cmd.payload.url ?? cmd.payload.source ?? "").trim();
const mode: MediaMode = cmd.payload.mode === "screen" ? "screen" : "music";
const requestedBy = String(cmd.payload.requestedBy ?? "unknown");
if (!url) {
@@ -175,74 +140,6 @@ export class MediaHandler {
};
}
// Screen share (GoLive) path — bypasses the audio queue entirely.
if (mode === "screen") {
try {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway client not initialized",
};
}
if (!this.screenController) {
this.screenController = new ScreenShareController(
this.client,
this.getVoiceStatus,
// releaseVoice — disconnect the @discordjs/voice connection so the
// dank074 Streamer can take over (Discord: one voice session/user).
async (status) => {
const vc = this.voiceControllerAccessor?.();
const guildId = status.activeGuildId ?? null;
if (vc && guildId) {
await vc.disconnectGuild(guildId);
}
},
// restoreVoice — reconnect the @discordjs audio connection after
// the stream ends so mic/listen keep working.
async (status) => {
const vc = this.voiceControllerAccessor?.();
if (vc && status.activeGuildId && status.activeChannelId) {
await vc.connect(status.activeGuildId, status.activeChannelId);
}
},
);
}
const playback = await this.screenController.start(url);
this.screenPlayback = playback;
currentTrackItem = {
id: randomUUID(),
source: url,
title: url,
kind: "url",
mode: "screen",
requestedBy,
addedAt: Date.now(),
status: "playing",
};
playback.done
.catch((err) => {
this.logger.error(
{ error: err instanceof Error ? err.message : String(err) },
"Screen playback promise rejected",
);
})
.finally(() => {
this.screenPlayback = null;
if (currentTrackItem?.mode === "screen") {
currentTrackItem = null;
}
});
this.logger.info({ url }, "Screen share started");
return { id: cmd.id, success: true, data: buildStatusPayload() };
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
this.logger.error({ error: message }, "Screen share failed to start");
return { id: cmd.id, success: false, data: null, error: message };
}
}
// Lightweight metadata fetch for display — the full resolve happens in playNext
let title: string = url;
let duration: number | undefined;
@@ -264,7 +161,7 @@ export class MediaHandler {
source: url,
title,
kind: "url" as const,
mode,
mode: "music",
requestedBy,
addedAt: Date.now(),
status: "queued",
@@ -308,12 +205,6 @@ export class MediaHandler {
}
async handleMediaStop(cmd: CommandMessage): Promise<CommandReply<unknown>> {
// Stop screen share if active — playback.done.finally clears the item.
this.screenPlayback?.stop();
this.screenPlayback = null;
if (currentTrackItem?.mode === "screen") {
currentTrackItem = null;
}
discordPlayer.stop("music");
currentTrackItem = null;
mediaQueue.length = 0; // Clear entire queue
@@ -394,7 +285,7 @@ export class MediaHandler {
// Music playback: transcode once to high-quality OggOpus (48kHz stereo,
// 192kbps) with volume baked into the encode. This avoids the double
// lossy encode that inlineVolume would cause and gives Discord the
// cleanest possible stream. Screen share bypasses this entirely.
// cleanest possible stream.
const transcoded = transcodeToHighQualityOgg(
resolution.stream,
discordPlayer.getMusicVolume(),
services/discord-gateway/src/modules/voice-recording/mediaSource.ts
+3 -158
View File
@@ -1,14 +1,5 @@
import { type ChildProcess, spawn } from "node:child_process";
import {
chmodSync,
existsSync,
mkdtempSync,
readdirSync,
readFileSync,
rmSync,
statSync,
writeFileSync,
} from "node:fs";
import { chmodSync, existsSync, readFileSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { PassThrough, type Readable } from "node:stream";
@@ -232,7 +223,7 @@ function buildNotInstalledError(): Error {
* in account's cookies. The path is configurable via GMW_YT_COOKIES_PATH
* (default: the BWS-provided file the deploy writes to /etc/.../ytcookies.txt).
* If the file doesn't exist we pass nothing and fall back to anon (YouTube
* may 403 — screen share will fail gracefully, not crash).
* may 403 — playback will fail gracefully, not crash).
*/
var _cachedCookiePath: string | null = null;
function buildCookieArgs(): string[] {
@@ -267,7 +258,7 @@ function buildCookieArgs(): string[] {
// Never hand the ORIGINAL system file to yt-dlp: recent yt-dlp rewrites
// the cookie file on close (`--cookies` implies write-back). The system
// file is owned by another user (root/deploy) and the service user
// cannot write it → PermissionError → yt-dlp exits 1 → screen share
// cannot write it → PermissionError → yt-dlp exits 1 → playback
// fails for every attempt. Copy to a per-run temp file (like the env
// branch above) so write-back lands somewhere we own; if the original
// is not readable we fall back to anonymous (YouTube may 403 → the
@@ -305,44 +296,6 @@ function buildCookieArgs(): string[] {
return [];
}
/** Invidious instances for anon YouTube fetch (fallback when cookies 403). */
export const INVIDIOUS_INSTANCES = [
"yewtu.be",
"yewtu.nanomorph.dev",
"invidious.snopyta.org",
"invidious.kavin.rocks",
];
/** True if url is a YouTube watch URL (youtu.be / youtube.com/watch). */
export function isYoutubeWatchUrl(url: string): boolean {
try {
const u = new URL(url);
return (
u.hostname === "youtu.be" ||
(u.hostname === "www.youtube.com" && u.pathname === "/watch") ||
(u.hostname === "youtube.com" && u.pathname === "/watch")
);
} catch {
return false;
}
}
/** Rewrite a YouTube watch URL to an invidious instance (anon, no bot-check). */
export function toInvidiousUrl(url: string, instance: string): string {
try {
const u = new URL(url);
if (u.hostname === "youtu.be") {
const id = u.pathname.slice(1);
return `https://${instance}/watch?v=${id}`;
}
const id = u.searchParams.get("v");
if (id) return `https://${instance}/watch?v=${id}`;
return url;
} catch {
return url;
}
}
// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------
@@ -505,114 +458,6 @@ export function resolveMediaUrl(
});
}
/**
* Download the merged video+audio media for screen share to a TEMP FILE
* first, then return the file path.
*
* Screen-share playback does NOT stream from yt-dlp stdout anymore: the
* merge feed is delivered at network speed (bursts + stalls), and a live
* pipe defeats ffmpeg's `-re` throttle (the input PTS timeline is
* unreliable), so the encoder force-duplicates held frames → ~1fps video.
* Downloading the FULL clip to a file first gives the encoder a clean,
* monotonic-PTS input, where `-re` (applied in prepareStream for string
* inputs) pacing is proven reliable.
*
* @returns absolute path of the completed media file (caller should delete
* it via cleanup after playback ends).
*/
export function downloadScreenInput(url: string): Promise<string> {
return new Promise<string>((resolve, reject) => {
// Temp dir per run (world-writable like /tmp) so parallel/retry runs
// never collide on merge fragments.
const tmpDir = mkdtempSync(join(tmpdir(), "gmw-ytdlp-"));
chmodSync(tmpDir, 0o1777);
const cookieArgs = buildCookieArgs();
const args = [
"-f",
"bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best",
// File output (NOT `-o -`): yt-dlp merges DASH fragments into a real
// container with clean timestamps, which is what -re needs to pace.
"-o",
join(tmpDir, "media.%(ext)s"),
"--no-playlist",
"--no-warnings",
"--no-progress",
...cookieArgs,
url,
];
logger.info({ url }, "Downloading full media for screen share input");
const proc = spawn("yt-dlp", args, {
stdio: ["ignore", "pipe", "pipe"],
});
activeProcesses.add(proc);
let stderrBuf = "";
const MAX_STDERR = 4096;
proc.stderr?.on("data", (chunk: Buffer) => {
if (stderrBuf.length < MAX_STDERR) {
stderrBuf += chunk.toString("utf8");
}
});
let settled = false;
const failOnce = (message: string) => {
if (settled) return;
settled = true;
activeProcesses.delete(proc);
rmSync(tmpDir, { recursive: true, force: true });
reject(new Error(message));
};
proc.on("error", (err: NodeJS.ErrnoException) => {
failOnce(`yt-dlp failed to start: ${err.message}`);
});
let procFinished = false;
proc.on("close", (code) => {
procFinished = true;
activeProcesses.delete(proc);
if (code !== 0) {
const detail = stderrBuf.trim() ? `: ${stderrBuf.trim()}` : "";
failOnce(`yt-dlp download failed (exit ${code})${detail}`);
return;
}
// Find the media file yt-dlp wrote (skip .part / .ytdl temp state).
let mediaPath: string | null = null;
for (const entry of readdirSync(tmpDir)) {
if (entry.endsWith(".part") || entry.endsWith(".ytdl")) continue;
mediaPath = join(tmpDir, entry);
break;
}
if (mediaPath && existsSync(mediaPath) && statSync(mediaPath).size > 0) {
settled = true;
resolve(mediaPath);
} else {
failOnce("yt-dlp finished but produced no media file");
}
});
// Safety net: a stalled download must not hang the gateway forever.
const timer = setTimeout(
() => {
try {
proc.kill("SIGTERM");
} catch {
/* already dead */
}
if (!procFinished) {
failOnce("yt-dlp download timed out (10 min)");
}
},
10 * 60 * 1000,
);
proc.once("close", () => clearTimeout(timer));
});
}
/**
* Extract metadata (title, duration, thumbnail) from a media URL
* without downloading the audio stream.
services/discord-gateway/src/modules/voice-recording/mediaTypes.ts
+2 -18
View File
@@ -1,7 +1,7 @@
import type { Readable } from "node:stream";
import type { StreamType } from "@discordjs/voice";
export type MediaMode = "music" | "screen";
export type MediaMode = "music";
export type MediaSourceKind =
| "url"
| "local"
@@ -51,23 +51,7 @@ export interface MusicPlayer {
play(source: ResolvedMediaSource): MusicPlayback;
}
export interface ScreenSharePlayback {
done: Promise<void>;
stop(): void;
}
export interface ScreenShareVoiceStatus {
connected: boolean;
activeGuildId: string | null;
activeChannelId: string | null;
}
export interface ScreenShareController {
isActive(): boolean;
start(source: string): Promise<ScreenSharePlayback>;
}
export type DiscordPlayerOwner = "none" | "browser-bridge" | "music" | "screen";
export type DiscordPlayerOwner = "none" | "browser-bridge" | "music";
export interface DiscordPlayOptions {
inputType?: StreamType;
services/discord-gateway/src/modules/voice-recording/screenShareController.ts
-291
View File
@@ -1,291 +0,0 @@
import { rmSync } from "node:fs";
import { dirname } from "node:path";
import type { Client } from "discord.js-selfbot-v13";
import { createChildLogger } from "@/shared/logger/index";
import {
Encoders,
normalizeVideoCodec,
playStream,
prepareStream,
Streamer,
} from "../../goLive/index.js";
import {
downloadScreenInput,
INVIDIOUS_INSTANCES,
isYoutubeWatchUrl,
toInvidiousUrl,
} from "./mediaSource.js";
import type { ScreenSharePlayback } from "./mediaTypes.js";
import { discordPlayer } from "./player.js";
const logger = createChildLogger("screen-share");
export interface ScreenShareVoiceStatus {
connected: boolean;
activeGuildId: string | null;
activeChannelId: string | null;
}
/**
* Discord Go Live (screenshare) via @dank074/discord-video-stream.
*
* Pipeline:
* URL (YouTube, dll.) → yt-dlp direct video URL → ffmpeg (H264 720p30) →
* playStream({ type: "go-live" }) → Discord voice channel as Go Live.
*
* Restored from the pre-microservices implementation (commit d50ce86,
* src/media/screenShareController.ts) — the interface survived in
* mediaTypes.ts but the implementation was lost during the split.
*/
export class ScreenShareController {
private logger = createChildLogger("screen-share");
private streamer: Streamer | null = null;
private active: ScreenSharePlayback | null = null;
constructor(
private readonly client: Client,
private readonly getVoiceStatus: () => ScreenShareVoiceStatus,
/** Disconnect the @discordjs/voice connection so the Streamer can take
* over the voice channel (Discord allows only ONE voice session per user
* — two connections collide and the Streamer never gets VOICE_SERVER_UPDATE). */
private readonly releaseVoice: (
status: ScreenShareVoiceStatus,
) => void | Promise<void>,
/** Reconnect the @discordjs/voice connection after the stream ends. */
private readonly restoreVoice: (
status: ScreenShareVoiceStatus,
) => void | Promise<void>,
) {}
isActive(): boolean {
return this.active !== null;
}
/**
* Resolve the screen-share input with retry (downloads the FULL media to a
* temp file; returns the file path).
*
* Transient YouTube 403s kill a download BEFORE completion; we validate
* the finished file and retry with a FRESH yt-dlp run (signed DASH URLs
* expire quickly). Downloading to a file (instead of streaming the merge
* pipe) gives the encoder a monotonic-PTS input, so ffmpeg `-re` pacing
* in prepareStream actually works (it does NOT on live pipes — the root
* of the ~1fps video).
*/
private async resolveInputWithRetry(source: string): Promise<string> {
const MAX_ATTEMPTS = 3;
let lastError: Error | null = null;
// YouTube may 403 even with account cookies (IP-bound session / bot check
// on VPS IP). When the source is a YouTube URL and cookies fail, fall back
// to anon Invidious mirror instances — no auth needed.
const isYt = isYoutubeWatchUrl(source);
let invidiousIdx = 0;
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
// On a 403 against YouTube, try the next Invidious instance for this attempt.
if (
isYt &&
lastError &&
/403|bot|Sign in|not a bot|access denied|permission|EACCES|cookie/i.test(
lastError.message,
) &&
invidiousIdx < INVIDIOUS_INSTANCES.length
) {
const inst = INVIDIOUS_INSTANCES[invidiousIdx];
this.logger.warn(
{ attempt, instance: inst, error: lastError.message },
"YouTube blocked (403); falling back to Invidious mirror",
);
source = toInvidiousUrl(source, inst);
invidiousIdx++;
}
try {
const mediaPath = await downloadScreenInput(source);
this.logger.info(
{ mediaPath, attempt },
"Screen input downloaded to file",
);
return mediaPath;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
this.logger.warn(
{
attempt,
maxAttempts: MAX_ATTEMPTS,
error: lastError.message,
},
"Screen input download failed; retrying with fresh yt-dlp",
);
if (attempt < MAX_ATTEMPTS) {
await new Promise((r) => setTimeout(r, 1500 * attempt));
}
}
}
throw (
lastError ??
new Error("Screen input resolution failed after multiple attempts")
);
}
async start(source: string): Promise<ScreenSharePlayback> {
const status = this.getVoiceStatus();
if (!status.connected || !status.activeGuildId || !status.activeChannelId) {
throw new Error("Connect to a voice channel before sharing screen");
}
if (this.active || discordPlayer.getOwner() !== "none") {
throw new Error("Another media mode is active");
}
try {
const input = await this.resolveInputWithRetry(source);
if (!this.streamer) {
this.streamer = new Streamer(this.client);
}
const guild = this.client.guilds.cache.get(status.activeGuildId);
const channel = guild?.channels.cache.get(status.activeChannelId);
if (
!channel ||
(channel.type !== "GUILD_VOICE" && channel.type !== "GUILD_STAGE_VOICE")
) {
throw new Error(
`Voice channel ${status.activeChannelId} not found for screen share`,
);
}
// Free the @discordjs/voice connection BEFORE the Streamer joins, so
// the user has only one voice session (Discord requirement).
await this.releaseVoice(status);
await Promise.race([
this.streamer.joinVoiceChannel(channel),
new Promise<never>((_, reject) =>
setTimeout(
() =>
reject(
new Error("Timed out joining voice channel for screen share"),
),
15000,
),
),
]);
const prepared = prepareStream(input, {
encoder: Encoders.software({ x264: { preset: "superfast" } }),
width: 1280,
height: 720,
frameRate: 30,
bitrateVideo: 2500,
bitrateVideoMax: 4000,
// GoLive with audio: the encoder muxes to NUT (video h264 + opus
// audio) so the audio SSRC carries RTP too. Discord's GoLive
// pipeline expects audio — a video-only stream shows a static
// tile/thumbnail instead of live video. When the source has no
// audio track, the encoder's `-map 0:a:0?` yields no audio stream
// and the demuxer simply reports none (video still flows).
includeAudio: true,
videoCodec: normalizeVideoCodec("H264"),
});
const { command } = prepared;
// The downloaded temp media lives in a per-run tmpdir; remove it once
// playback is done (natural end, failure, or user stop). The tmpdir is
// the parent of the media file, so deleting it removes the file too.
const cleanupTempMedia = () => {
try {
if (typeof input === "string") {
rmSync(dirname(input), { recursive: true, force: true });
}
} catch {
/* best-effort — tmp dirs are world-writable, leak is bounded */
}
};
let stopped = false;
// Restore the @discordjs/voice connection after the stream ends (both
// natural end and failure), so the user can keep using audio/mic.
const restoreAfter = () => {
if (!stopped) {
stopped = true;
try {
command.kill("SIGTERM");
} catch {
/* already dead */
}
}
cleanupTempMedia();
try {
this.streamer?.voiceConnection?.stop();
} catch {
/* already gone */
}
if (this.restoreVoice) {
// Best-effort restore after a short delay. Discord often needs the
// Streamer's session fully torn down before @discordjs/voice can
// re-join; if that races, the reconnect times out — the FE shows
// disconnected and the user just clicks Connect again. This is an
// accepted UX tradeoff for GoLive (single voice session per user).
setTimeout(() => {
Promise.resolve(this.restoreVoice(status)).catch((err) => {
this.logger.warn(
{ error: err instanceof Error ? err.message : String(err) },
"Failed to restore voice connection after screen share (user can reconnect manually)",
);
});
}, 5000);
}
};
const done = playStream(prepared, this.streamer, {
type: "go-live",
width: 1280,
height: 720,
frameRate: 30,
})
.catch((err: unknown) => {
// Never let a stream failure become an unhandledRejection — that
// crashed the whole gateway. Log + surface via the done promise.
const message = err instanceof Error ? err.message : String(err);
this.logger.error(
{ error: message, source },
"Screen stream failed during playback",
);
})
.finally(() => {
restoreAfter();
this.active = null;
});
this.active = {
done,
stop: () => {
if (stopped) return;
stopped = true;
try {
command.kill("SIGTERM");
} catch {
/* already dead */
}
cleanupTempMedia();
// Leave the voice channel the Streamer joined (its own connection).
try {
this.streamer?.voiceConnection?.stop();
} catch {
/* already gone */
}
this.active = null;
},
};
logger.info({ source }, "Screen share started");
return this.active;
} catch (error) {
this.active = null;
const message = error instanceof Error ? error.message : String(error);
logger.error({ error: message, source }, "Screen stream failed");
throw error;
}
}
}