54c3ce7d1b
Resolve the Discord user ID to a server nickname / global name (cached) and display that in the transcript channel + console logs.
354 lines
13 KiB
TypeScript
354 lines
13 KiB
TypeScript
/**
|
|
* Discord voice I/O.
|
|
*
|
|
* - Joins the caller's voice channel.
|
|
* - Receives each speaker's Opus stream, decodes to PCM, and on end-of-speech
|
|
* forwards the utterance (as a WAV) to the brain bridge.
|
|
* - Plays the brain's spoken reply back into the channel.
|
|
*
|
|
* No AI logic here — capture in, audio out. The brain lives in bridge/.
|
|
*/
|
|
import { Readable } from "node:stream";
|
|
import {
|
|
joinVoiceChannel,
|
|
createAudioPlayer,
|
|
createAudioResource,
|
|
EndBehaviorType,
|
|
StreamType,
|
|
AudioPlayerStatus,
|
|
VoiceConnection,
|
|
VoiceConnectionStatus,
|
|
entersState,
|
|
type AudioPlayer,
|
|
} from "@discordjs/voice";
|
|
import prism from "prism-media";
|
|
import type { VoiceBasedChannel } from "discord.js";
|
|
import { converseStream } from "./bridge.ts";
|
|
import { config } from "./config.ts";
|
|
|
|
const DISCORD_RATE = 48000;
|
|
const DISCORD_CHANNELS = 2;
|
|
|
|
/** Build a minimal PCM16 mono WAV around raw little-endian samples. */
|
|
function pcm16MonoToWav(pcm: Buffer, sampleRate: number): Buffer {
|
|
const header = Buffer.alloc(44);
|
|
const dataLen = pcm.length;
|
|
header.write("RIFF", 0);
|
|
header.writeUInt32LE(36 + dataLen, 4);
|
|
header.write("WAVE", 8);
|
|
header.write("fmt ", 12);
|
|
header.writeUInt32LE(16, 16);
|
|
header.writeUInt16LE(1, 20); // PCM
|
|
header.writeUInt16LE(1, 22); // mono
|
|
header.writeUInt32LE(sampleRate, 24);
|
|
header.writeUInt32LE(sampleRate * 2, 28); // byte rate (mono * 2 bytes)
|
|
header.writeUInt16LE(2, 32); // block align
|
|
header.writeUInt16LE(16, 34); // bits per sample
|
|
header.write("data", 36);
|
|
header.writeUInt32LE(dataLen, 40);
|
|
return Buffer.concat([header, pcm]);
|
|
}
|
|
|
|
/** Downmix interleaved stereo PCM16 to mono PCM16. */
|
|
function stereoToMono(stereo: Buffer): Buffer {
|
|
const samples = stereo.length / 4; // 2 ch * 2 bytes
|
|
const mono = Buffer.alloc(samples * 2);
|
|
for (let i = 0; i < samples; i++) {
|
|
const l = stereo.readInt16LE(i * 4);
|
|
const r = stereo.readInt16LE(i * 4 + 2);
|
|
mono.writeInt16LE((l + r) >> 1, i * 2);
|
|
}
|
|
return mono;
|
|
}
|
|
|
|
export class VoiceSession {
|
|
readonly guildId: string;
|
|
private connection: VoiceConnection;
|
|
private player: AudioPlayer;
|
|
private listening = new Set<string>();
|
|
/** Set once the session is torn down (user left / leave command). In-flight
|
|
* captures check this so we don't run STT/reply or post a transcript for
|
|
* audio that arrived after the user already left the channel. */
|
|
private destroyed = false;
|
|
/** Opus subscriptions currently capturing, so leave() can end them
|
|
* immediately instead of waiting out the silence timeout. */
|
|
private activeStreams = new Set<{ destroy: () => void }>();
|
|
/** Pending reply clips. Played one at a time so concurrent speakers don't
|
|
* cut each other's replies off. */
|
|
private playQueue: Buffer[] = [];
|
|
/** Optional callback to surface EVERY heard utterance (and its outcome) to a
|
|
* text channel — including ones dropped before/at STT — so misses are
|
|
* diagnosable. `note` says why (e.g. "음성 아님(VAD 차단)", "너무 짧음", "ok"). */
|
|
onTurn?: (info: {
|
|
user: string;
|
|
/** Resolved display name (server nickname / global name) for the speaker,
|
|
* so logs show a human name instead of the raw Discord user ID. */
|
|
userName?: string;
|
|
transcript: string;
|
|
reply: string;
|
|
note?: string;
|
|
/** Wall-clock epoch-ms markers for each pipeline stage, so the transcript
|
|
* channel can show what took long. Listening is measured here (capture
|
|
* start -> end of speech); LLM/TTS come from the brain bridge. STT shows
|
|
* up as the gap between listenEndMs and llmStartMs. */
|
|
listenStartMs?: number;
|
|
listenEndMs?: number;
|
|
llmStartMs?: number;
|
|
llmEndMs?: number;
|
|
ttsStartMs?: number;
|
|
ttsEndMs?: number;
|
|
}) => void;
|
|
/** Live screen-share state, sent with each turn so the brain routes search
|
|
* (Chrome while broadcasting, Gemini when off). */
|
|
getBroadcasting?: () => boolean;
|
|
/** Apply a broadcast directive the brain requested (start/stop the stream). */
|
|
onBroadcastAction?: (action: "start" | "stop") => void | Promise<void>;
|
|
|
|
/** The selfbot client behind this voice connection + the negotiated voice
|
|
* session_id, so the broadcast (Go-Live) can ride the SAME session — one
|
|
* account hears/speaks AND broadcasts. */
|
|
private readonly client: any;
|
|
private readonly channelId: string;
|
|
private sessionId?: string;
|
|
|
|
constructor(channel: VoiceBasedChannel) {
|
|
this.guildId = channel.guild.id;
|
|
this.channelId = channel.id;
|
|
this.client = (channel as any).client;
|
|
// Wrap the gateway adapter to capture our own voice session_id (needed to
|
|
// create the Go-Live stream on this same session).
|
|
const realCreator = channel.guild.voiceAdapterCreator;
|
|
const adapterCreator: typeof realCreator = (methods) =>
|
|
realCreator({
|
|
...methods,
|
|
onVoiceStateUpdate: (d: any) => {
|
|
if (d?.session_id) this.sessionId = d.session_id;
|
|
return methods.onVoiceStateUpdate(d);
|
|
},
|
|
});
|
|
this.connection = joinVoiceChannel({
|
|
channelId: channel.id,
|
|
guildId: channel.guild.id,
|
|
adapterCreator,
|
|
selfDeaf: false, // we need to hear users
|
|
selfMute: false,
|
|
});
|
|
this.player = createAudioPlayer();
|
|
this.connection.subscribe(this.player);
|
|
// Drain the queue when the current clip finishes.
|
|
this.player.on(AudioPlayerStatus.Idle, () => this.pump());
|
|
this.attachReceiver();
|
|
}
|
|
|
|
/** The shared session for single-account Go-Live, or null if session_id isn't
|
|
* captured yet / the client is unavailable. */
|
|
getSharedSession(): {
|
|
client: unknown;
|
|
guildId: string;
|
|
channelId: string;
|
|
sessionId: string;
|
|
botId: string;
|
|
} | null {
|
|
const botId = this.client?.user?.id;
|
|
if (!this.sessionId || !this.client || !botId) return null;
|
|
return { client: this.client, guildId: this.guildId, channelId: this.channelId, sessionId: this.sessionId, botId };
|
|
}
|
|
|
|
async ready(): Promise<void> {
|
|
await entersState(this.connection, VoiceConnectionStatus.Ready, 20_000);
|
|
}
|
|
|
|
private attachReceiver() {
|
|
const receiver = this.connection.receiver;
|
|
receiver.speaking.on("start", (userId: string) => {
|
|
if (this.listening.has(userId)) return;
|
|
this.listening.add(userId);
|
|
this.captureUtterance(userId).finally(() => this.listening.delete(userId));
|
|
});
|
|
}
|
|
|
|
/** Resolve a speaker's Discord user ID to a human display name (server
|
|
* nickname, else global name / username), cached so we don't refetch every
|
|
* utterance. Falls back to the ID if lookup fails. */
|
|
private nameCache = new Map<string, string>();
|
|
private async displayName(userId: string): Promise<string> {
|
|
const cached = this.nameCache.get(userId);
|
|
if (cached) return cached;
|
|
let name = userId;
|
|
try {
|
|
const guild: any = this.client?.guilds?.cache?.get(this.guildId);
|
|
let member: any = guild?.members?.cache?.get(userId);
|
|
if (!member && guild?.members?.fetch) member = await guild.members.fetch(userId).catch(() => null);
|
|
if (member) {
|
|
name = member.displayName || member.nickname || member.user?.globalName || member.user?.username || userId;
|
|
} else {
|
|
const u: any = this.client?.users?.cache?.get(userId) || (await this.client?.users?.fetch?.(userId).catch(() => null));
|
|
name = u?.globalName || u?.username || userId;
|
|
}
|
|
} catch {
|
|
/* fall back to id */
|
|
}
|
|
this.nameCache.set(userId, name);
|
|
return name;
|
|
}
|
|
|
|
private async captureUtterance(userId: string): Promise<void> {
|
|
// Don't start a new capture once we're tearing down (user left).
|
|
if (this.destroyed) return;
|
|
// "듣기 시작": the moment we begin capturing this speaker's utterance.
|
|
const listenStartMs = Date.now();
|
|
const opusStream = this.connection.receiver.subscribe(userId, {
|
|
end: { behavior: EndBehaviorType.AfterSilence, duration: config.silenceMs },
|
|
});
|
|
this.activeStreams.add(opusStream);
|
|
const decoder = new prism.opus.Decoder({
|
|
frameSize: 960,
|
|
channels: DISCORD_CHANNELS,
|
|
rate: DISCORD_RATE,
|
|
});
|
|
const chunks: Buffer[] = [];
|
|
const pcmStream = opusStream.pipe(decoder);
|
|
pcmStream.on("data", (c: Buffer) => chunks.push(c));
|
|
|
|
await new Promise<void>((resolve) => pcmStream.once("end", () => resolve()));
|
|
this.activeStreams.delete(opusStream);
|
|
// If the user left while we were capturing, drop this utterance entirely —
|
|
// don't run STT/reply or report a (usually empty/VAD-blocked) turn for
|
|
// audio that trailed in after they left.
|
|
if (this.destroyed) return;
|
|
// "듣기 종료": end of speech (silence detected). Anything after this is
|
|
// STT + reply + TTS on the brain side.
|
|
const listenEndMs = Date.now();
|
|
|
|
if (!chunks.length) return;
|
|
const mono = stereoToMono(Buffer.concat(chunks));
|
|
// Ignore blips shorter than ~300ms (likely noise / key clicks) — but still
|
|
// report them so the transcript channel shows every captured utterance.
|
|
if (mono.length < DISCORD_RATE * 0.3 * 2) {
|
|
this.onTurn?.({
|
|
user: userId,
|
|
userName: await this.displayName(userId),
|
|
transcript: "",
|
|
reply: "",
|
|
note: "너무 짧음(<300ms)",
|
|
listenStartMs,
|
|
listenEndMs,
|
|
});
|
|
return;
|
|
}
|
|
const wav = pcm16MonoToWav(mono, DISCORD_RATE);
|
|
|
|
try {
|
|
// Streaming turn: the brain sends transcript/reply first, then one audio
|
|
// clip per sentence as it is synthesised. We enqueue each clip on arrival
|
|
// so the first sentence starts playing while the rest are still spoken.
|
|
// The transcript-channel report is sent once the stream ends so it can
|
|
// include TTS timing (synthesis runs after the meta line). Audio still
|
|
// plays as it arrives — only the diagnostic text post waits.
|
|
let metaSeen: {
|
|
transcript: string;
|
|
reply: string;
|
|
note?: string;
|
|
llm_start_ms?: number;
|
|
llm_end_ms?: number;
|
|
} | undefined;
|
|
let endSeen: { tts_start_ms?: number; tts_end_ms?: number } | undefined;
|
|
await converseStream(wav, this.getBroadcasting?.(), {
|
|
onMeta: async (m) => {
|
|
metaSeen = m;
|
|
// Apply any broadcast directive the brain requested (e.g. user said
|
|
// "방송 켜줘 / 꺼줘") before the reply audio plays. The meta line
|
|
// always precedes the audio clips, so awaiting here preserves order.
|
|
if (m.broadcast_action && this.onBroadcastAction) {
|
|
try {
|
|
await this.onBroadcastAction(m.broadcast_action);
|
|
} catch (e) {
|
|
console.error("[voice] broadcast action failed:", e);
|
|
}
|
|
}
|
|
},
|
|
onAudio: (clip) => this.play(clip),
|
|
onEnd: (end) => {
|
|
endSeen = end;
|
|
},
|
|
});
|
|
// Report EVERY turn (even empty/VAD-dropped) so the transcript channel
|
|
// explains why a turn did or didn't answer, with full stage timing.
|
|
this.onTurn?.({
|
|
user: userId,
|
|
userName: await this.displayName(userId),
|
|
transcript: metaSeen?.transcript ?? "",
|
|
reply: metaSeen?.reply ?? "",
|
|
note: metaSeen?.note,
|
|
listenStartMs,
|
|
listenEndMs,
|
|
llmStartMs: metaSeen?.llm_start_ms,
|
|
llmEndMs: metaSeen?.llm_end_ms,
|
|
ttsStartMs: endSeen?.tts_start_ms,
|
|
ttsEndMs: endSeen?.tts_end_ms,
|
|
});
|
|
} catch (err) {
|
|
console.error("[voice] converse failed:", err);
|
|
}
|
|
}
|
|
|
|
/** Queue a WAV buffer for playback (FIFO, one clip at a time). */
|
|
play(wav: Buffer) {
|
|
this.playQueue.push(wav);
|
|
this.pump();
|
|
}
|
|
|
|
private pump() {
|
|
if (this.player.state.status !== AudioPlayerStatus.Idle) return;
|
|
const next = this.playQueue.shift();
|
|
if (!next) return;
|
|
const resource = createAudioResource(Readable.from(next), {
|
|
inputType: StreamType.Arbitrary,
|
|
});
|
|
this.player.play(resource);
|
|
}
|
|
|
|
destroy() {
|
|
this.destroyed = true;
|
|
// End any in-flight captures now so their pcmStream resolves immediately
|
|
// and the post-capture `destroyed` check drops them (no trailing
|
|
// post-leave VAD turns).
|
|
for (const s of this.activeStreams) {
|
|
try {
|
|
s.destroy();
|
|
} catch {
|
|
/* already ended */
|
|
}
|
|
}
|
|
this.activeStreams.clear();
|
|
try {
|
|
this.connection.destroy();
|
|
} catch {
|
|
/* already gone */
|
|
}
|
|
}
|
|
}
|
|
|
|
/** One session per guild. */
|
|
const sessions = new Map<string, VoiceSession>();
|
|
|
|
export async function joinChannel(channel: VoiceBasedChannel): Promise<VoiceSession> {
|
|
sessions.get(channel.guild.id)?.destroy();
|
|
const session = new VoiceSession(channel);
|
|
sessions.set(channel.guild.id, session);
|
|
await session.ready();
|
|
return session;
|
|
}
|
|
|
|
export function leaveGuild(guildId: string): boolean {
|
|
const s = sessions.get(guildId);
|
|
if (!s) return false;
|
|
s.destroy();
|
|
sessions.delete(guildId);
|
|
return true;
|
|
}
|
|
|
|
export function getSession(guildId: string): VoiceSession | undefined {
|
|
return sessions.get(guildId);
|
|
}
|