watch_sceen_ai/dave/bot.mjs

// Official Discord BOT voice joiner (replaces the selfbot join.mjs).
//
// Why this exists: the project migrated off the user-token selfbot path onto an
// official bot application. For the voice loop (STT input) this is the fully
// supported, ToS-safe path — @discordjs/voice lets a bot JOIN a voice channel,
// pass the DAVE/MLS E2EE handshake (via @snazzah/davey, handled internally), and
// RECEIVE per-user Opus audio. Only screenshare VIDEO receive still requires a
// selfbot, and video ("눈") is deferred, so nothing here needs a user token.
//
// This replicates M1 (join target channel, stay, detect speakers) and, for free,
// gives partial M2: it decodes each speaker's Opus to PCM and counts frames — the
// exact stream the STT stage will consume.
//
// Usage:
//   node bot.mjs                 # join and stay until killed
//   RUN_MS=15000 node bot.mjs    # join, hold 15s, then leave (for verification)
//
// Requires: the bot must be INVITED to the target guild with the "Connect" and
// "Speak" voice permissions. See dave/README or run `node bot.mjs --invite`.

import fs from 'node:fs';
import { Readable } from 'node:stream';
import {
  Client,
  GatewayIntentBits,
} from 'discord.js';
import {
  joinVoiceChannel,
  getVoiceConnection,
  entersState,
  VoiceConnectionStatus,
  EndBehaviorType,
  createAudioPlayer,
  createAudioResource,
  StreamType,
  NoSubscriberBehavior,
} from '@discordjs/voice';
import prism from 'prism-media';

// ---------- config ----------
function loadEnvFile() {
  try {
    return Object.fromEntries(
      fs.readFileSync(new URL('../.env', import.meta.url), 'utf8')
        .split('\n').filter(l => l && !l.startsWith('#') && l.includes('='))
        .map(l => { const i = l.indexOf('='); return [l.slice(0, i).trim(), l.slice(i + 1).trim()]; })
    );
  } catch { return {}; }
}
const env = loadEnvFile();
const TOKEN = process.env.DISCORD_BOT_TOKEN || env.DISCORD_BOT_TOKEN;
const GUILD_ID = process.env.GUILD_ID || env.GUILD_ID || '1352269198297923648';
const CHANNEL_ID = process.env.CHANNEL_ID || env.CHANNEL_ID || '1352269198914621465';
const RUN_MS = process.env.RUN_MS != null ? Number(process.env.RUN_MS) : 0; // 0 = stay forever
// Python STT+TTS voice-turn endpoint (run `python -m wsai --voice-server`).
const VOICE_ENDPOINT = process.env.WSAI_VOICE_ENDPOINT || env.WSAI_VOICE_ENDPOINT
  || 'http://127.0.0.1:8787/api/voice-turn';
// Dashboard control plane (state push + command poll), same host as voice-turn.
const API_BASE = VOICE_ENDPOINT.replace(/\/api\/voice-turn\/?$/, '');
const REPORT_ENDPOINT = API_BASE + '/api/bot/report';
const REPORT_INTERVAL_MS = Number(process.env.WSAI_REPORT_INTERVAL_MS || 2500);
// Ignore utterances shorter than this many PCM bytes (48kHz*2ch*2B = 192000 B/s),
// so key clicks / brief noise don't trigger a turn. ~0.35s.
const MIN_UTTERANCE_BYTES = Number(process.env.WSAI_MIN_UTTERANCE_BYTES || 67000);

const t0 = Date.now();
const log = (...a) => console.log(`[+${String(Date.now() - t0).padStart(6)}ms]`, ...a);

// Throttle bursty repeated logs. DAVE (E2EE) group transitions — someone joins
// or leaves the voice channel — briefly deliver undecryptable packets, so the
// same "recv stream error" can fire many times in a second. Log the first
// occurrence of a given message immediately, then collapse repeats within a
// window into one summary line instead of flooding the log.
const _throttle = new Map(); // key -> { count, timer }
function logThrottled(key, msg, windowMs = 10_000) {
  const e = _throttle.get(key);
  if (e) { e.count++; return; }
  log(msg);
  const timer = setTimeout(() => {
    const cur = _throttle.get(key);
    _throttle.delete(key);
    if (cur && cur.count > 0) log(`${msg} (+${cur.count} more in ${Math.round(windowMs / 1000)}s)`);
  }, windowMs);
  if (typeof timer.unref === 'function') timer.unref();
  _throttle.set(key, { count: 0, timer });
}

// PCM s16le -> WAV container (so the Python side can ffmpeg-decode it).
function wavHeader(dataLen, sampleRate = 48000, channels = 2, bits = 16) {
  const blockAlign = channels * bits / 8;
  const b = Buffer.alloc(44);
  b.write('RIFF', 0); b.writeUInt32LE(36 + dataLen, 4); b.write('WAVE', 8);
  b.write('fmt ', 12); b.writeUInt32LE(16, 16); b.writeUInt16LE(1, 20);
  b.writeUInt16LE(channels, 22); b.writeUInt32LE(sampleRate, 24);
  b.writeUInt32LE(sampleRate * blockAlign, 28); b.writeUInt16LE(blockAlign, 32);
  b.writeUInt16LE(bits, 34); b.write('data', 36); b.writeUInt32LE(dataLen, 40);
  return b;
}

// Speak audio into the voice channel. Feed the reply wav bytes through ffmpeg
// (StreamType.Arbitrary) so @discordjs/voice re-encodes to Opus.
let voicePlayer = null;
function playReply(wavBytes) {
  if (!voicePlayer || !wavBytes || wavBytes.length === 0) return;
  const resource = createAudioResource(Readable.from(wavBytes), { inputType: StreamType.Arbitrary });
  voicePlayer.play(resource);
}

// One utterance: PCM -> WAV -> POST to Python -> play the reply back.
async function handleUtterance(userId, pcm) {
  if (pcm.length < MIN_UTTERANCE_BYTES) {
    log(`utterance too short user=${userId} bytes=${pcm.length} — skip`);
    return;
  }
  const wav = Buffer.concat([wavHeader(pcm.length), pcm]);
  // Resolve who spoke (Discord display name) so the dashboard log can show it.
  let speaker = userId;
  try {
    const g = currentGuildId && client.guilds.cache.get(currentGuildId);
    const m = g && (g.members.cache.get(userId) || await g.members.fetch(userId).catch(() => null));
    if (m) speaker = m.displayName || m.user.username;
  } catch {}
  let resp;
  try {
    resp = await fetch(VOICE_ENDPOINT, {
      method: 'POST',
      headers: { 'Content-Type': 'audio/wav', 'X-User-Name': encodeURIComponent(speaker) },
      body: wav,
    });
  } catch (e) {
    log(`voice-turn POST failed (is \`python -m wsai --voice-server\` running?): ${e.message}`);
    return;
  }
  if (!resp.ok) { log(`voice-turn HTTP ${resp.status}`); return; }
  const heard = decodeURIComponent(resp.headers.get('X-Heard') || '');
  const reply = decodeURIComponent(resp.headers.get('X-Reply') || '');
  const buf = Buffer.from(await resp.arrayBuffer());
  log(`heard="${heard}" reply="${reply}" replyWav=${buf.length}B`);
  playReply(buf);
}

// ---------- invite URL helper ----------
// A bot cannot add itself to a server; a server admin must click an OAuth2 invite
// URL once. Print it so the user can authorise the bot with voice permissions.
if (process.argv.includes('--invite')) {
  const APP_ID = process.env.DISCORD_APP_ID || env.DISCORD_APP_ID || '';
  // permissions: Connect(1<<20) | Speak(1<<21) | UseVAD(1<<25) | ViewChannel(1<<10)
  const perms = (1n << 20n) | (1n << 21n) | (1n << 25n) | (1n << 10n);
  if (!APP_ID) { console.error('set DISCORD_APP_ID (application/client id) in .env to build the invite URL'); process.exit(2); }
  console.log(`https://discord.com/oauth2/authorize?client_id=${APP_ID}&scope=bot&permissions=${perms}`);
  process.exit(0);
}

if (!TOKEN) { console.error('no DISCORD_BOT_TOKEN in env or ../.env'); process.exit(2); }

// ---------- client ----------
const client = new Client({
  intents: [GatewayIntentBits.Guilds, GatewayIntentBits.GuildVoiceStates],
});

const perUser = new Map(); // userId -> { opusPackets, pcmFrames }

// --- control-plane state (dashboard drives which channel we're in) --------- #
let currentGuildId = null, currentChannelId = null, currentChannelName = null;
const speakingSet = new Set();   // userIds currently speaking (for participant list)
const activeSubs = new Set();    // userIds with an in-flight receive subscription

// Attach the bot's audio player (so it can speak) to a fresh connection.
function setupPlayer(connection) {
  voicePlayer = createAudioPlayer({ behaviors: { noSubscriber: NoSubscriberBehavior.Play } });
  voicePlayer.on('error', (e) => log(`player error: ${e.message}`));
  connection.subscribe(voicePlayer);
}

// Attach the receive path: capture each utterance and run the voice turn.
function setupReceiver(connection) {
  const receiver = connection.receiver;
  receiver.speaking.on('start', (userId) => {
    speakingSet.add(userId);
    if (userId === client.user.id || activeSubs.has(userId)) return;
    activeSubs.add(userId);
    if (!perUser.has(userId)) perUser.set(userId, { opusPackets: 0, pcmFrames: 0 });
    const opusStream = receiver.subscribe(userId, {
      end: { behavior: EndBehaviorType.AfterSilence, duration: 800 },
    });
    const decoder = new prism.opus.Decoder({ rate: 48000, channels: 2, frameSize: 960 });
    const chunks = [];
    opusStream.on('data', () => { perUser.get(userId).opusPackets++; });
    opusStream.on('error', (e) => { logThrottled(`recv:${e.message}`, `recv stream error user=${userId}: ${e.message}`); activeSubs.delete(userId); });
    opusStream.pipe(decoder);
    decoder.on('data', (d) => { chunks.push(d); perUser.get(userId).pcmFrames++; });
    decoder.on('error', (e) => log(`decode error user=${userId}: ${e.message}`));
    decoder.on('end', () => {
      activeSubs.delete(userId);
      const pcm = Buffer.concat(chunks);
      log(`utterance end user=${userId} pcm=${pcm.length}B — running voice turn`);
      handleUtterance(userId, pcm).catch((e) => log(`voice turn error: ${e.message}`));
    });
  });
  receiver.speaking.on('end', (userId) => speakingSet.delete(userId));
}

// Join (or switch to) a voice channel on command from the dashboard.
async function joinChannel(guildId, channelId) {
  const guild = await client.guilds.fetch(guildId).catch(() => null);
  const channel = guild && await guild.channels.fetch(channelId).catch(() => null);
  if (!channel || !channel.isVoiceBased()) { log(`join failed: ${guildId}/${channelId} not a voice channel`); return; }
  if (currentGuildId && currentGuildId !== guildId) leaveChannel();
  const connection = joinVoiceChannel({
    channelId, guildId, adapterCreator: guild.voiceAdapterCreator,
    selfDeaf: false, selfMute: false,
  });
  connection.on('error', (e) => log(`voice connection error: ${e.message}`));
  try {
    await entersState(connection, VoiceConnectionStatus.Ready, 40_000);
  } catch (e) { log(`join not Ready in 40s: ${e.message}`); return; }
  currentGuildId = guildId; currentChannelId = channelId; currentChannelName = channel.name;
  speakingSet.clear(); activeSubs.clear();
  setupPlayer(connection);
  setupReceiver(connection);
  connection.on(VoiceConnectionStatus.Disconnected, () => {
    log('voice: disconnected — attempting to resume…');
    Promise.race([
      entersState(connection, VoiceConnectionStatus.Signalling, 5_000),
      entersState(connection, VoiceConnectionStatus.Connecting, 5_000),
    ]).catch(() => { log('voice: could not resume'); leaveChannel(); });
  });
  log(`✅ joined voice: "${guild.name}" / "${channel.name}"`);
}

function leaveChannel() {
  try { getVoiceConnection(currentGuildId)?.destroy(); } catch {}
  currentGuildId = currentChannelId = currentChannelName = null;
  speakingSet.clear(); activeSubs.clear();
}

// Build the state snapshot the dashboard shows (identity, joinable servers +
// voice channels, current channel, and who is in it).
function buildState() {
  const guilds = [...client.guilds.cache.values()].map((g) => ({
    id: g.id, name: g.name,
    voiceChannels: [...g.channels.cache.values()]
      .filter((c) => c.isVoiceBased())
      .map((c) => ({ id: c.id, name: c.name })),
  }));
  let members = [];
  if (currentGuildId && currentChannelId) {
    const ch = client.guilds.cache.get(currentGuildId)?.channels.cache.get(currentChannelId);
    if (ch && ch.members) {
      members = [...ch.members.values()].map((m) => ({
        id: m.id, name: m.displayName, speaking: speakingSet.has(m.id),
      }));
    }
  }
  return {
    identity: { id: client.user.id, username: client.user.username, tag: client.user.tag },
    guilds,
    current: { guildId: currentGuildId, channelId: currentChannelId, channelName: currentChannelName },
    members,
  };
}

async function handleCommand(cmd) {
  if (cmd.type === 'join') await joinChannel(cmd.guildId, cmd.channelId);
  else if (cmd.type === 'leave') { leaveChannel(); log('left voice on command'); }
}

// Push state to the dashboard and apply any commands it hands back.
async function reportLoop() {
  let j;
  try {
    const r = await fetch(REPORT_ENDPOINT, {
      method: 'POST', headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(buildState()),
    });
    j = await r.json();
  } catch { return; }  // dashboard down: keep running, retry next tick
  for (const cmd of (j?.commands || [])) {
    try { await handleCommand(cmd); } catch (e) { log(`command ${cmd?.type} failed: ${e.message}`); }
  }
}

let leaving = false;
function leaveAndExit(code = 0) {
  if (leaving) return;
  leaving = true;
  try { getVoiceConnection(currentGuildId || GUILD_ID)?.destroy(); } catch {}
  const summary = [...perUser.entries()].map(([u, s]) => `${u}:opus=${s.opusPackets},pcm=${s.pcmFrames}`);
  log(`leaving. speakers heard: ${summary.length ? summary.join(' ') : '(none)'}`);
  try { client.destroy(); } catch {}
  setTimeout(() => process.exit(code), 300);
}
process.on('SIGINT', () => { log('SIGINT'); leaveAndExit(0); });
process.on('SIGTERM', () => { log('SIGTERM'); leaveAndExit(0); });

// Hard time-box ceiling, armed at startup regardless of handshake state.
if (RUN_MS > 0) setTimeout(() => { log(`RUN_MS=${RUN_MS} hard ceiling elapsed — leaving`); leaveAndExit(0); }, RUN_MS);

client.once('clientReady', async () => {
  log(`logged in as ${client.user.tag} (${client.user.id})`);
  log(`voice endpoint: ${VOICE_ENDPOINT} · control: ${REPORT_ENDPOINT}`);

  // Backward-compat: if a default guild/channel is configured, auto-join it.
  // Otherwise idle and wait for the dashboard to pick a channel.
  if (GUILD_ID && CHANNEL_ID) {
    await joinChannel(GUILD_ID, CHANNEL_ID).catch((e) => log(`initial join failed: ${e.message}`));
  } else {
    log('no default channel — waiting for the dashboard to select a server/voice channel…');
  }

  // Report state + poll commands forever. This is what powers the dashboard's
  // bot info, server/voice-channel pickers, participant list, and join/leave.
  reportLoop();
  const reportTimer = setInterval(reportLoop, REPORT_INTERVAL_MS);
  if (typeof reportTimer.unref === 'function') reportTimer.unref();
});

client.on('error', (e) => log('client error', e.message));
client.login(TOKEN).catch((e) => { log(`FATAL: login failed — ${e.message}`); process.exit(1); });