feat: enhance media handling and audio processing logic

README.md

@@ -8,7 +8,8 @@ Stack utama: Node.js, pnpm, TypeScript, `discord.js-selfbot-v13`, `@discordjs/vo
 
 - Node.js versi modern yang kompatibel dengan TypeScript dan Vite.
 - pnpm 10.x. Repo ini dipin ke `pnpm@10.25.0`.
-- FFmpeg tersedia di `PATH` untuk proses muxing audio.
+- FFmpeg tersedia di `PATH` untuk proses muxing audio dan playback media.
+- `yt-dlp` tersedia di `PATH` untuk resolve audio YouTube, search result YouTube, dan Spotify track.
 - Native audio dependencies dapat dibuild di mesin lokal (`@discordjs/opus`, `better-sqlite3`, `sodium-native`).
 
 Install FFmpeg:
@@ -21,6 +22,14 @@ sudo apt install ffmpeg
 sudo pacman -S ffmpeg
 ```
 
+Install `yt-dlp`:
+
+```bash
+pnpm run install:yt-dlp
+```
+
+Script installer akan memakai package manager yang tersedia (`pacman`, `apt-get`, `dnf`, `brew`) atau fallback ke `pipx`/`pip`.
+
 ## Setup
 
 ```bash
@@ -76,6 +85,9 @@ pnpm run test
 
 # Build frontend + TypeScript
 pnpm run build
+
+# Install external yt-dlp CLI for YouTube/search/Spotify track playback
+pnpm run install:yt-dlp
 ```
 
 ## Database
@@ -104,7 +116,8 @@ pnpm run db:studio
 - Attachment capture dan upload ke endpoint Picser.
 - SQLite/PostgreSQL via Drizzle ORM.
 - REST API dan WebSocket untuk dashboard.
-- Dashboard React untuk pesan, gambar, voice, dan moderation review.
+- Dashboard React untuk pesan, gambar, voice, media playback, dan moderation review.
+- Media playback dari direct URL, file lokal, YouTube URL, search terms, dan Spotify track URL.
 - Metrics Prometheus di endpoint server.
 - Retry dengan backoff untuk operasi eksternal.
 - AI moderation analysis opsional via konfigurasi `AI_*`.

package.json

@@ -18,7 +18,8 @@
     "db:generate": "drizzle-kit generate",
     "db:migrate": "drizzle-kit migrate",
     "db:migrate:programmatic": "tsx src/database/migrate.ts",
-    "db:studio": "drizzle-kit studio"
+    "db:studio": "drizzle-kit studio",
+    "install:yt-dlp": "sh scripts/install-yt-dlp.sh"
   },
   "dependencies": {
     "@dank074/discord-video-stream": "workspace:*",

public/index.html

@@ -69,6 +69,7 @@
       isListening: false,
       localStreaming: false,
       localListening: false,
+      mediaAutoListening: false,
       audioContextTransmit: null,
       audioContextListen: null,
       processor: null,
@@ -86,7 +87,7 @@
 
     async function apiRequest(url, options = {}) { const response = await fetch(url, { headers: { 'Content-Type': 'application/json', ...(options.headers || {}) }, ...options }); if (!response.ok) { const error = await response.json().catch(() => ({ message: response.statusText })); throw new Error(error.message || response.statusText); } return response.json(); }
     function showError(message) { el.errorBox.textContent = message; el.errorBox.style.display = 'block'; setTimeout(() => { el.errorBox.style.display = 'none'; }, 4500); }
-    function postUIState(patch) { return apiRequest('/api/ui-state', { method: 'POST', body: JSON.stringify(patch) }); }
+    async function postUIState(patch) { const next = await apiRequest('/api/ui-state', { method: 'POST', body: JSON.stringify(patch) }); await applyServerState(next); return next; }
     function renderOptions(select, items, placeholder) { select.replaceChildren(); const first = document.createElement('option'); first.value = ''; first.textContent = placeholder; select.appendChild(first); for (const item of items) { const option = document.createElement('option'); option.value = item.id; option.textContent = item.name; select.appendChild(option); } }
     function appendOptions(select, items) { const existing = new Set([...select.options].map((option) => option.value)); for (const item of items) { if (existing.has(item.id)) continue; const option = document.createElement('option'); option.value = item.id; option.textContent = item.name; select.appendChild(option); } }
     function appendEmpty(parent, message) { const empty = document.createElement('div'); empty.className = 'empty'; empty.textContent = message; parent.appendChild(empty); }
@@ -101,7 +102,7 @@
     async function disconnectVoice() { await apiRequest('/api/disconnect', { method: 'POST' }); await refreshStatus(); }
 
     function connectWebSocket() { const protocol = location.protocol === 'https:' ? 'wss:' : 'ws:'; state.socket = new WebSocket(`${protocol}//${location.host}/ws`); state.socket.binaryType = 'arraybuffer'; state.socket.onopen = () => { el.wsDot.classList.add('on'); el.wsStatusText.textContent = 'Connected'; }; state.socket.onclose = () => { el.wsDot.classList.remove('on'); el.wsStatusText.textContent = 'Reconnecting'; setTimeout(connectWebSocket, 2500); }; state.socket.onerror = () => { el.wsDot.classList.remove('on'); el.wsDot.classList.add('warn'); el.wsStatusText.textContent = 'Socket error'; }; state.socket.onmessage = (event) => { if (event.data instanceof ArrayBuffer) { handleIncomingPCM(event.data); return; } try { handleJsonEvent(event.data); } catch {} }; }
-    function handleJsonEvent(raw) { const message = JSON.parse(raw); if (message.type === 'ui_state') return applyServerState(message.state); if (message.type === 'user_state') return renderUsers(message.users || []); if (message.type === 'message_created') { state.text.unshift(message.data); renderText(); } if (message.type === 'message_updated') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, { edited_content: message.data.edited_content, edited_at: message.data.edited_at, type: 'edited' }); renderText(); } if (message.type === 'message_deleted') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, { deleted_at: message.data.deleted_at, type: 'deleted' }); renderText(); } if (message.type === 'attachment_uploaded') fetchText(); if (message.type === 'message_analyzed') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, message.data); renderText(); } if (message.type === 'media_state') { state.media = message.state; renderMedia(); } }
+    function handleJsonEvent(raw) { const message = JSON.parse(raw); if (message.type === 'ui_state') return applyServerState(message.state); if (message.type === 'user_state') return renderUsers(message.users || []); if (message.type === 'message_created') { state.text.unshift(message.data); renderText(); } if (message.type === 'message_updated') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, { edited_content: message.data.edited_content, edited_at: message.data.edited_at, type: 'edited' }); renderText(); } if (message.type === 'message_deleted') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, { deleted_at: message.data.deleted_at, type: 'deleted' }); renderText(); } if (message.type === 'attachment_uploaded') fetchText(); if (message.type === 'message_analyzed') { const item = state.text.find((entry) => entry.id === message.data.id); if (item) Object.assign(item, message.data); renderText(); } if (message.type === 'media_state') { state.media = message.state; reconcileDynamicAudio().catch((error) => showError(error.message)); renderMedia(); } }
 
     async function applyServerState(next) {
       if (!next || state.applyingServerState) return;
@@ -134,14 +135,14 @@
       if (textChanged || textGuildChanged || state.activeTab === 'text') {
         fetchText().catch((error) => showError(error.message));
       }
-      await reconcileListenState();
+      await reconcileDynamicAudio();
-      await reconcileStreamingState();
       state.applyingServerState = false;
     }
 
     function applyActiveTab(tab) { document.querySelectorAll('.tab-btn').forEach((item) => item.classList.toggle('active', item.dataset.tab === tab)); document.querySelectorAll('.tab-content').forEach((item) => item.classList.toggle('active', item.id === tab)); el.activeTabLabel.textContent = tab === 'text' ? 'Text' : 'Voice'; }
-    async function reconcileListenState() { if (state.isListening && !state.localListening) { try { await startListeningLocal(); } catch (error) { showError(`Speaker error: ${error.message}`); await postUIState({ isListening: false }); } } else if (!state.isListening && state.localListening) { stopListeningLocal(); } }
+    async function reconcileDynamicAudio() { await reconcileStreamingState(); await reconcileListenState(); }
-    async function reconcileStreamingState() { if (state.isStreaming && !state.localStreaming) { try { await startStreamingLocal(); } catch (error) { showError(`Microphone error: ${error.message}`); await postUIState({ isStreaming: false }); } } else if (!state.isStreaming && state.localStreaming) { stopStreamingLocal(); } }
+    async function reconcileListenState() { const shouldListen = state.isListening || !!state.media.current; if (shouldListen && !state.localListening) { try { await startListeningLocal(!!state.media.current && !state.isListening); } catch (error) { showError(`Speaker error: ${error.message}`); state.isListening = false; state.mediaAutoListening = false; stopListeningLocal(); apiRequest('/api/ui-state', { method: 'POST', body: JSON.stringify({ isListening: false }) }).catch((postError) => showError(postError.message)); } } else if (!shouldListen && state.localListening) { stopListeningLocal(); } else if (state.localListening) { renderListenStatus(); } }
+    async function reconcileStreamingState() { if (state.media.current && state.isStreaming) { state.isStreaming = false; apiRequest('/api/ui-state', { method: 'POST', body: JSON.stringify({ isStreaming: false }) }).catch((postError) => showError(postError.message)); } if (state.isStreaming && !state.localStreaming) { try { await startStreamingLocal(); } catch (error) { showError(`Microphone error: ${error.message}`); state.isStreaming = false; stopStreamingLocal(); apiRequest('/api/ui-state', { method: 'POST', body: JSON.stringify({ isStreaming: false }) }).catch((postError) => showError(postError.message)); } } else if (!state.isStreaming && state.localStreaming) { stopStreamingLocal(); } }
 
     function renderUsers(users) { el.userList.replaceChildren(); if (users.length === 0) return appendEmpty(el.userList, 'No active speakers'); for (const user of users) { const row = document.createElement('div'); row.className = `user-item${user.speaking ? ' speaking' : ''}`; const img = document.createElement('img'); img.src = user.avatar || ''; img.alt = ''; const name = document.createElement('span'); name.textContent = user.username; row.append(img, name); el.userList.appendChild(row); } }
     async function fetchText() { if (!state.selectedTextChannel) return renderText(); const result = await apiRequest(`/api/messages?channel=${encodeURIComponent(state.selectedTextChannel)}&type=text&limit=80`); state.text = result.data || []; renderText(); }
@@ -152,8 +153,9 @@
     function handleIncomingPCM(data) { if (!state.localListening || !state.audioContextListen) return; const headerView = new DataView(data, 0, 4); const userIdHash = headerView.getInt32(0, true); const audioData = data.slice(4); const int16Array = new Int16Array(audioData); const float32Array = new Float32Array(int16Array.length); for (let i = 0; i < int16Array.length; i++) float32Array[i] = int16Array[i] / 32768; const audioBuffer = state.audioContextListen.createBuffer(CHANNELS, float32Array.length / CHANNELS, SAMPLE_RATE); const nowBuffering = audioBuffer.getChannelData(0); for (let i = 0; i < audioBuffer.length; i++) nowBuffering[i] = float32Array[i]; const source = state.audioContextListen.createBufferSource(); source.buffer = audioBuffer; source.connect(state.audioContextListen.destination); const currentTime = state.audioContextListen.currentTime; let userNextStartTime = state.userTimelines.get(userIdHash) || 0; if (userNextStartTime < currentTime) userNextStartTime = currentTime + 0.05; source.start(userNextStartTime); userNextStartTime += audioBuffer.duration; state.userTimelines.set(userIdHash, userNextStartTime); }
     async function startStreamingLocal() { const stream = await navigator.mediaDevices.getUserMedia({ audio: true }); state.localStreaming = true; el.toggleBtn.textContent = 'Stop Transmitting'; state.audioContextTransmit = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: SAMPLE_RATE }); const source = state.audioContextTransmit.createMediaStreamSource(stream); const analyser = state.audioContextTransmit.createAnalyser(); analyser.fftSize = 64; source.connect(analyser); const dataArray = new Uint8Array(analyser.frequencyBinCount); state.processor = state.audioContextTransmit.createScriptProcessor(4096, 1, 1); source.connect(state.processor); state.processor.connect(state.audioContextTransmit.destination); state.processor.onaudioprocess = (event) => { if (!state.localStreaming || state.socket.readyState !== WebSocket.OPEN) return; const inputData = event.inputBuffer.getChannelData(0); const pcmData = new Int16Array(inputData.length); for (let i = 0; i < inputData.length; i++) pcmData[i] = Math.max(-1, Math.min(1, inputData[i])) * 32767; state.socket.send(pcmData.buffer); analyser.getByteFrequencyData(dataArray); bars.forEach((bar, index) => { const percent = (dataArray[index] / 255) * 100; bar.style.height = `${Math.max(2, percent)}%`; }); }; }
     function stopStreamingLocal() { state.localStreaming = false; if (state.processor) state.processor.disconnect(); if (state.audioContextTransmit) state.audioContextTransmit.close(); state.processor = null; state.audioContextTransmit = null; el.toggleBtn.textContent = 'Start Transmitting'; bars.forEach((bar) => { bar.style.height = '2px'; }); }
-    async function startListeningLocal() { if (!state.audioContextListen) state.audioContextListen = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: SAMPLE_RATE }); await state.audioContextListen.resume(); state.localListening = true; el.listenBtn.textContent = 'Stop Listening'; el.listenStatus.textContent = 'Listening Live...'; }
+    async function startListeningLocal(auto = false) { if (!state.audioContextListen) state.audioContextListen = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: SAMPLE_RATE }); await state.audioContextListen.resume(); state.localListening = true; state.mediaAutoListening = auto; renderListenStatus(); }
-    function stopListeningLocal() { state.audioContextListen?.suspend(); state.userTimelines.clear(); state.localListening = false; el.listenBtn.textContent = 'Join Listen Channel'; el.listenStatus.textContent = 'Speaker Off'; }
+    function stopListeningLocal() { state.audioContextListen?.suspend(); state.userTimelines.clear(); state.localListening = false; state.mediaAutoListening = false; renderListenStatus(); }
+    function renderListenStatus() { el.listenBtn.textContent = state.isListening ? 'Stop Listening' : 'Join Listen Channel'; el.listenStatus.textContent = state.localListening ? (state.media.current && state.mediaAutoListening ? 'Media Monitor On' : 'Listening Live...') : 'Speaker Off'; }
     function updateVisualizer(level) { bars.forEach((bar, index) => { const wave = Math.sin(index * 0.55 + Date.now() / 140) * 0.35 + 0.65; bar.style.height = `${Math.max(3, level * 190 * wave)}px`; }); }
 
     document.querySelectorAll('.tab-btn').forEach((button) => { button.addEventListener('click', () => postUIState({ activeTab: button.dataset.tab }).catch((error) => showError(error.message))); });
@@ -167,9 +169,9 @@
     el.channelFilter.addEventListener('change', () => { const selectedTextChannel = el.channelFilter.value; const url = new URL(location.href); if (selectedTextChannel) url.searchParams.set('channel', selectedTextChannel); else url.searchParams.delete('channel'); if (el.textGuildSelect.value) url.searchParams.set('guild', el.textGuildSelect.value); history.replaceState({}, '', url); postUIState({ selectedTextChannel }).catch((error) => showError(error.message)); });
 
     async function fetchMediaStatus() { state.media = await apiRequest('/api/media/status'); renderMedia(); }
-    async function queueMedia() { const source = el.mediaSourceInput.value.trim(); if (!source) return showError('Enter a music URL or local file path'); state.media = await apiRequest('/api/media/queue', { method: 'POST', body: JSON.stringify({ source }) }); el.mediaSourceInput.value = ''; renderMedia(); }
+    async function queueMedia() { const source = el.mediaSourceInput.value.trim(); if (!source) return showError('Enter a music URL or local file path'); if (state.isStreaming || state.localStreaming) await postUIState({ isStreaming: false }); state.media = await apiRequest('/api/media/queue', { method: 'POST', body: JSON.stringify({ source }) }); el.mediaSourceInput.value = ''; await reconcileDynamicAudio(); renderMedia(); }
-    async function skipMedia() { state.media = await apiRequest('/api/media/skip', { method: 'POST' }); renderMedia(); }
+    async function skipMedia() { state.media = await apiRequest('/api/media/skip', { method: 'POST' }); await reconcileDynamicAudio(); renderMedia(); }
-    async function stopMedia() { state.media = await apiRequest('/api/media/stop', { method: 'POST' }); renderMedia(); }
+    async function stopMedia() { state.media = await apiRequest('/api/media/stop', { method: 'POST' }); await reconcileDynamicAudio(); renderMedia(); }
     function renderMedia() { el.mediaQueueList.replaceChildren(); const current = state.media.current; el.mediaStatus.textContent = current ? `Playing ${current.title}` : 'Idle'; if (current) { const item = document.createElement('div'); item.className = 'event-card'; item.textContent = `Now: ${current.title}`; el.mediaQueueList.appendChild(item); } for (const queued of state.media.queue || []) { const item = document.createElement('div'); item.className = 'event-card'; item.textContent = queued.title; el.mediaQueueList.appendChild(item); } if (!current && (!state.media.queue || state.media.queue.length === 0)) appendEmpty(el.mediaQueueList, 'No media queued'); }
 
     el.queueMediaBtn.addEventListener('click', () => queueMedia().catch((error) => showError(error.message)));

scripts/install-yt-dlp.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env sh
+set -eu
+
+if command -v yt-dlp >/dev/null 2>&1; then
+  echo "yt-dlp already installed: $(command -v yt-dlp)"
+  yt-dlp --version
+  exit 0
+fi
+
+if command -v pacman >/dev/null 2>&1; then
+  sudo pacman -S --needed yt-dlp
+elif command -v apt-get >/dev/null 2>&1; then
+  sudo apt-get update
+  sudo apt-get install -y yt-dlp
+elif command -v dnf >/dev/null 2>&1; then
+  sudo dnf install -y yt-dlp
+elif command -v brew >/dev/null 2>&1; then
+  brew install yt-dlp
+elif command -v pipx >/dev/null 2>&1; then
+  pipx install yt-dlp
+elif command -v python3 >/dev/null 2>&1; then
+  python3 -m pip install --user --upgrade yt-dlp
+else
+  echo "Could not find pacman, apt-get, dnf, brew, pipx, or python3 to install yt-dlp." >&2
+  exit 1
+fi
+
+if ! command -v yt-dlp >/dev/null 2>&1; then
+  echo "yt-dlp installed but is not on PATH. Restart your shell or add the installer bin directory to PATH." >&2
+  exit 1
+fi
+
+echo "yt-dlp installed: $(command -v yt-dlp)"
+yt-dlp --version

src/moderation/aiAnalysisWorker.ts

@@ -0,0 +1,94 @@
+import { parentPort } from "node:worker_threads";
+import { buildConversationPromptMessages } from "./conversationContext";
+import { runModerationAnalysis } from "./llmModerationClient";
+import {
+  getConversationContextBefore,
+  updateMessageAIAnalysis,
+} from "./messageStore";
+import type { MessageRecord } from "./types";
+
+const MAX_CONTEXT_TOKENS = 8000;
+
+interface AnalysisWorkerRequest {
+  conversationKey: string;
+  messages: MessageRecord[];
+}
+
+type AnalysisWorkerResponse =
+  | {
+      ok: true;
+      conversationKey: string;
+      rows: MessageRecord[];
+    }
+  | {
+      ok: false;
+      conversationKey: string;
+      rows: MessageRecord[];
+      error: string;
+    };
+
+async function processAnalysisRequest({
+  conversationKey,
+  messages,
+}: AnalysisWorkerRequest): Promise<AnalysisWorkerResponse> {
+  try {
+    const firstMessage = messages[0];
+    if (!firstMessage) return { ok: true, conversationKey, rows: [] };
+
+    const contextBefore = await getConversationContextBefore({
+      channelId: firstMessage.channel_id,
+      threadId: firstMessage.thread_id,
+      beforeCreatedAt: firstMessage.created_at,
+      limit: 20,
+    });
+
+    const promptMessages = buildConversationPromptMessages({
+      contextBefore,
+      targets: messages,
+      maxTokens: MAX_CONTEXT_TOKENS,
+    });
+
+    const result = await runModerationAnalysis({
+      targets: messages,
+      contextText: promptMessages.join("\n"),
+    });
+
+    const rows: MessageRecord[] = [];
+    for (const analysisResult of result.results) {
+      const row = await updateMessageAIAnalysis(analysisResult.messageId, {
+        status: analysisResult.status,
+        flags: JSON.stringify(analysisResult.flags),
+        score: analysisResult.score,
+        raw: JSON.stringify(result.raw),
+        analysis: analysisResult.analysis,
+        analyzedAt: Date.now(),
+        error: null,
+      });
+      if (row) rows.push(row);
+    }
+
+    return { ok: true, conversationKey, rows };
+  } catch (error) {
+    const errorMessage = error instanceof Error ? error.message : String(error);
+    const rows: MessageRecord[] = [];
+
+    for (const msg of messages) {
+      const row = await updateMessageAIAnalysis(msg.id, {
+        status: "error",
+        flags: null,
+        score: null,
+        raw: null,
+        analysis: null,
+        analyzedAt: Date.now(),
+        error: errorMessage,
+      });
+      if (row) rows.push(row);
+    }
+
+    return { ok: false, conversationKey, rows, error: errorMessage };
+  }
+}
+
+parentPort?.on("message", async (request: AnalysisWorkerRequest) => {
+  parentPort?.postMessage(await processAnalysisRequest(request));
+});

src/moderation/aiAnalyzer.ts

@@ -1,9 +1,7 @@
+import { Worker } from "node:worker_threads";
 import { config } from "../config";
 import { createChildLogger } from "../logger";
-import { buildConversationPromptMessages } from "./conversationContext";
-import { runModerationAnalysis } from "./llmModerationClient";
 import {
-  getConversationContextBefore,
   getMessageById,
   getPendingConversationKeys,
   getPendingMessagesByConversation,
@@ -38,9 +36,15 @@ const MAX_ACTIVE_REQUESTS = 1;
 const DEBOUNCE_MS = 1500;
 const RECOVERY_INTERVAL_MS = 15000;
 const ERROR_COOLDOWN_MS = 30000;
-const MAX_CONTEXT_TOKENS = 8000;
 const MAX_BATCH_SIZE = 25;
 
+interface AnalysisWorkerResponse {
+  ok: boolean;
+  conversationKey: string;
+  rows: MessageRecord[];
+  error?: string;
+}
+
 /**
  * Gets the conversation key for a message (thread_id or channel_id)
  */
@@ -86,68 +90,37 @@ async function processBatch(
   activeRequests++;
   conversationProcessing.add(conversationKey);
   try {
-    // Get context before the first message
+    const result = await runAnalysisInWorker(conversationKey, messages);
-    const firstMessage = messages[0];
-    const contextBefore = await getConversationContextBefore({
-      channelId: firstMessage.channel_id,
-      threadId: firstMessage.thread_id,
-      beforeCreatedAt: firstMessage.created_at,
-      limit: 20,
-    });
 
-    // Build prompt with context
+    for (const row of result.rows) {
-    const promptMessages = buildConversationPromptMessages({
-      contextBefore,
-      targets: messages,
-      maxTokens: MAX_CONTEXT_TOKENS,
-    });
-
-    const contextText = promptMessages.join("\n");
-
-    // Run moderation analysis
-    const result = await runModerationAnalysis({
-      targets: messages,
-      contextText,
-    });
-
-    // Store results
-    const analyzedRows: MessageRecord[] = [];
-    for (const analysisResult of result.results) {
-      const row = await updateMessageAIAnalysis(analysisResult.messageId, {
-        status: analysisResult.status,
-        flags: JSON.stringify(analysisResult.flags),
-        score: analysisResult.score,
-        raw: JSON.stringify(result.raw),
-        analysis: analysisResult.analysis,
-        analyzedAt: Date.now(),
-        error: null,
-      });
-      if (row) {
-        analyzedRows.push(row);
-      }
-    }
-
-    // Broadcast analyzed messages
-    for (const row of analyzedRows) {
       getModerationBroadcaster()?.messageAnalyzed(row);
     }
 
-    // Clear error cooldown on success
+    if (!result.ok) {
-    conversationErrorCooldown.delete(conversationKey);
+      lastError = result.error ?? "Analysis worker failed";
+      conversationErrorCooldown.set(
+        conversationKey,
+        Date.now() + ERROR_COOLDOWN_MS,
+      );
+      logger.error(
+        { conversationKey, error: lastError },
+        "Batch analysis failed",
+      );
+      return;
+    }
 
-    logger.info(
+    conversationErrorCooldown.delete(conversationKey);
-      { conversationKey, count: messages.length },
-      "Batch analysis complete",
-    );
   } catch (error) {
     lastError = error instanceof Error ? error.message : String(error);
-
+    conversationErrorCooldown.set(
+      conversationKey,
+      Date.now() + ERROR_COOLDOWN_MS,
+    );
     logger.error(
       { conversationKey, error: lastError },
-      "Batch analysis failed",
+      "Analysis worker failed",
     );
 
-    // Mark all messages in batch as error
     for (const msg of messages) {
       const row = await updateMessageAIAnalysis(msg.id, {
         status: "error",
@@ -158,42 +131,49 @@ async function processBatch(
         analyzedAt: Date.now(),
         error: lastError,
       });
-      if (row) {
+      if (row) getModerationBroadcaster()?.messageAnalyzed(row);
-        getModerationBroadcaster()?.messageAnalyzed(row);
-      }
     }
-
-    // Set error cooldown for this conversation
-    conversationErrorCooldown.set(
-      conversationKey,
-      Date.now() + ERROR_COOLDOWN_MS,
-    );
   } finally {
     activeRequests--;
     conversationProcessing.delete(conversationKey);
   }
 }
 
+async function runAnalysisInWorker(
+  conversationKey: string,
+  messages: MessageRecord[],
+): Promise<AnalysisWorkerResponse> {
+  return new Promise((resolve, reject) => {
+    const worker = new Worker(new URL("./aiAnalysisWorker.ts", import.meta.url));
+
+    worker.once("message", (response: AnalysisWorkerResponse) => {
+      worker.terminate().catch((error) => {
+        logger.warn({ error }, "Failed to terminate analysis worker");
+      });
+      resolve(response);
+    });
+    worker.once("error", reject);
+    worker.once("exit", (code) => {
+      if (code !== 0) {
+        reject(new Error(`Analysis worker exited with code ${code}`));
+      }
+    });
+    worker.postMessage({ conversationKey, messages });
+  });
+}
+
 /**
  * Debounced analysis trigger for a conversation
  */
 function scheduleConversationAnalysis(conversationKey: string): void {
   // Skip if already processing
   if (conversationProcessing.has(conversationKey)) {
-    logger.debug(
-      { conversationKey },
-      "Conversation already processing, skipping schedule",
-    );
     return;
   }
 
   // Skip if in error cooldown
   const cooldownUntil = conversationErrorCooldown.get(conversationKey);
   if (cooldownUntil && Date.now() < cooldownUntil) {
-    logger.debug(
-      { conversationKey, cooldownMs: cooldownUntil - Date.now() },
-      "Conversation in error cooldown, skipping schedule",
-    );
     return;
   }
 
@@ -209,10 +189,6 @@ function scheduleConversationAnalysis(conversationKey: string): void {
 
     // If activeRequests >= MAX_ACTIVE_REQUESTS, requeue instead of waiting
     if (activeRequests >= MAX_ACTIVE_REQUESTS) {
-      logger.debug(
-        { conversationKey, activeRequests },
-        "Max active requests reached, requeuing conversation",
-      );
       scheduleConversationAnalysis(conversationKey);
       return;
     }
@@ -237,7 +213,6 @@ function scheduleConversationAnalysis(conversationKey: string): void {
 export async function queueMessageAnalysis(messageId: string): Promise<void> {
   if (!config.AI_ANALYSIS_ENABLED) return;
 
-  logger.debug({ messageId }, "Queueing message for analysis");
 
   try {
     // Look up the message to get its conversation key
@@ -267,7 +242,6 @@ export async function queueMessageAnalysis(messageId: string): Promise<void> {
 export function queueConversationAnalysis(conversationKey: string): void {
   if (!config.AI_ANALYSIS_ENABLED) return;
 
-  logger.debug({ conversationKey }, "Queueing conversation for analysis");
 
   // Schedule debounced analysis
   scheduleConversationAnalysis(conversationKey);
@@ -288,12 +262,7 @@ export function getAnalysisQueueStatus(): AnalysisQueueStatus {
  * Starts the pending AI analysis recovery worker
  */
 export function startPendingAIAnalysisWorker(): void {
-  if (!config.AI_ANALYSIS_ENABLED) {
+  if (!config.AI_ANALYSIS_ENABLED) return;
-    logger.info("AI analysis disabled");
-    return;
-  }
-
-  logger.info("AI analysis worker started");
 
   setInterval(async () => {
     try {
@@ -317,10 +286,6 @@ export function startPendingAIAnalysisWorker(): void {
           continue;
         }
 
-        logger.debug(
-          { conversationKey: key },
-          "Recovering pending conversation",
-        );
         scheduleConversationAnalysis(key);
       }
     } catch (error) {

src/moderation/attachmentUploader.ts

@@ -82,12 +82,7 @@ export async function uploadAttachmentToPicser(
       },
     );
 
-    const parsed = parseUploadResponse(response);
+    return parseUploadResponse(response);
-    logger.info(
-      { filename, url: parsed.url },
-      "Attachment uploaded successfully",
-    );
-    return parsed;
   } catch (error) {
     logger.error(
       {
@@ -127,8 +122,6 @@ export async function processAttachmentUpload(
   filename: string,
 ): Promise<void> {
   try {
-    logger.info({ attachmentId, filename }, "Starting attachment upload");
-
     const buffer = await downloadDiscordAttachment(discordUrl);
 
     const sizeMb = buffer.length / (1024 * 1024);
@@ -141,10 +134,6 @@ export async function processAttachmentUpload(
     const result = await uploadAttachmentToPicser(buffer, filename);
 
     await updateAttachmentAsUploaded(attachmentId, result.url, Date.now());
-    logger.info(
-      { attachmentId, uploadedUrl: result.url },
-      "Attachment upload completed",
-    );
   } catch (error) {
     const errorMsg = error instanceof Error ? error.message : String(error);
     await updateAttachmentAsFailedUpload(attachmentId, errorMsg);

src/moderation/backlogSync.ts

@@ -73,11 +73,6 @@ export async function syncBacklogMessages(client: Client): Promise<void> {
     await syncSelectedChannelBacklog(client, guild.id, config.TEXT_CHANNEL_ID);
     return;
   }
-
-  logger.info(
-    { guildId: guild.id },
-    "Backlog sync ready (will sync on-demand per selected channel)",
-  );
 }
 
 export async function syncSelectedChannelBacklog(
@@ -102,17 +97,8 @@ export async function syncSelectedChannelBacklog(
   }
 
   const cutoffTime = Date.now() - config.BACKLOG_SYNC_HOURS * 60 * 60 * 1000;
-  logger.info(
-    { guildId, channelId, hours: config.BACKLOG_SYNC_HOURS },
-    "Starting backlog sync for selected channel",
-  );
-
   try {
     const count = await syncChannelMessages(channel, cutoffTime);
-    logger.info(
-      { channelId, count },
-      "Backlog sync completed for selected channel",
-    );
     return count;
   } catch (error) {
     logger.warn(

src/moderation/messageCapture.ts

@@ -123,15 +123,6 @@ export async function captureMessage(
       }
     }
   }
-
-  logger.info(
-    {
-      messageId: message.id,
-      channelId: message.channelId,
-      attachmentCount: message.attachments.size,
-    },
-    "Message captured",
-  );
 }
 
 export function registerMessageCapture(client: Client): void {
@@ -206,8 +197,6 @@ export function registerMessageCapture(client: Client): void {
           deleted_at: deletedAt,
         });
       }
-
-      logger.info({ messageId: message.id }, "Message deletion captured");
     } catch (error) {
       logger.error(
         {
@@ -218,6 +207,4 @@ export function registerMessageCapture(client: Client): void {
       );
     }
   });
-
-  logger.info("Message capture handlers registered");
 }

src/moderation/messageStore.ts

@@ -61,11 +61,6 @@ export async function insertMessage(message: MessageRecord): Promise<void> {
   try {
     const database = db();
     await database.insert(messagesTable).values(message).onConflictDoNothing();
-
-    logger.debug(
-      { messageId: message.id, channelId: message.channel_id },
-      "Message inserted",
-    );
   } catch (error) {
     logger.error(
       {
@@ -94,12 +89,7 @@ export async function upsertMessageForCapture(
       .onConflictDoNothing()
       .returning({ id: messagesTable.id });
 
-    const inserted = rows.length > 0;
+    return rows.length > 0;
-    logger.debug(
-      { messageId: message.id, channelId: message.channel_id, inserted },
-      inserted ? "Message inserted for capture" : "Message already captured",
-    );
-    return inserted;
   } catch (error) {
     logger.error(
       {
@@ -134,8 +124,6 @@ export async function updateMessageAsEdited(
         ai_error: null,
       })
       .where(eq(messagesTable.id, messageId));
-
-    logger.debug({ messageId }, "Message marked as edited");
   } catch (error) {
     logger.error(
       {
@@ -161,8 +149,6 @@ export async function updateMessageAsDeleted(
         type: "deleted",
       })
       .where(eq(messagesTable.id, messageId));
-
-    logger.debug({ messageId }, "Message marked as deleted");
   } catch (error) {
     logger.error(
       {
@@ -217,11 +203,6 @@ export async function insertAttachment(
       .insert(attachmentsTable)
       .values(attachment)
       .onConflictDoNothing();
-
-    logger.debug(
-      { attachmentId: attachment.id, messageId: attachment.message_id },
-      "Attachment inserted",
-    );
   } catch (error) {
     logger.error(
       {
@@ -282,11 +263,6 @@ export async function updateAttachmentAsUploaded(
         uploaded_at: uploadedAt,
       })
       .where(eq(attachmentsTable.id, attachmentId));
-
-    logger.debug(
-      { attachmentId, uploadedUrl },
-      "Attachment marked as uploaded",
-    );
   } catch (error) {
     logger.error(
       {
@@ -312,8 +288,6 @@ export async function updateAttachmentAsFailedUpload(
         upload_error: error,
       })
       .where(eq(attachmentsTable.id, attachmentId));
-
-    logger.debug({ attachmentId, error }, "Attachment marked as failed upload");
   } catch (error) {
     logger.error(
       {

src/player.ts

@@ -41,14 +41,19 @@ export class DiscordPlayer {
     });
 
     this.player.play(resource);
+    this.connection?.subscribe(this.player);
+  }
+
+  public getStatus(): AudioPlayerStatus {
+    return this.player.state.status;
   }
 
   public pause() {
     this.player.pause(true);
   }
 
-  public unpause() {
+  public unpause(): boolean {
-    this.player.unpause();
+    return this.player.unpause();
   }
 
   public stop() {

src/routes/analysisRoutes.ts

@@ -1,15 +1,12 @@
 import type { Router } from "express";
 import express from "express";
 import { AppError } from "../errors";
-import { createChildLogger } from "../logger";
 import {
   getAnalysisQueueStatus,
   queueMessageAnalysis,
 } from "../moderation/aiAnalyzer";
 import { getMessageById } from "../moderation/messageStore";
 
-const logger = createChildLogger("analysis-routes");
-
 export function createAnalysisRoutes(): Router {
   const router = express.Router();
 
@@ -41,8 +38,6 @@ export function createAnalysisRoutes(): Router {
       // Queue for analysis
       await queueMessageAnalysis(id);
 
-      logger.info({ messageId: id }, "Message queued for re-analysis");
-
       res.json({
         success: true,
         messageId: id,

src/routes/syncRoutes.ts

@@ -45,7 +45,6 @@ export function createSyncRoutes(client: Client): Router {
       }
 
       if (shouldSkipRecentBacklogSync(guildId, channelId)) {
-        logger.debug({ guildId, channelId }, "Skipping recent backlog sync");
         res.json({
           success: true,
           channelId,
@@ -56,15 +55,8 @@ export function createSyncRoutes(client: Client): Router {
         return;
       }
 
-      logger.info({ guildId, channelId }, "Queueing backlog sync");
-
       syncSelectedChannelBacklog(client, guildId, channelId)
-        .then((count) => {
+        .then(() => {})
-          logger.info(
-            { guildId, channelId, messagesSync: count },
-            "Backlog sync complete",
-          );
-        })
         .catch((error) => {
           logger.warn(
             {

src/webserver.ts

@@ -4,6 +4,7 @@ import path from "node:path";
 import type { Client } from "discord.js-selfbot-v13";
 import express from "express";
 import helmet from "helmet";
+import { AudioPlayerStatus } from "@discordjs/voice";
 import * as prism from "prism-media";
 import { WebSocketServer } from "ws";
 import { AppError } from "./errors";
@@ -286,31 +287,42 @@ export async function startWebserver(
   const SILENCE_TAIL_MS = 300; // continue sending silence for 300ms after browser stops
   const MAX_BUF_BYTES = BYTES_PER_FRAME * 50; // cap at 1 second to avoid runaway buffer
 
-  const opusEncoder = new prism.opus.Encoder({
+  let opusEncoder: prism.opus.Encoder;
-    rate: RATE,
+  let bridgePlayerPaused = true;
-    channels: CHANNELS,
+  const SILENCE_FRAME = Buffer.alloc(BYTES_PER_FRAME, 0);
-    frameSize: FRAME_SIZE,
-  });
-  const oggBitstream = new prism.opus.OggLogicalBitstream({
-    opusHead: new prism.opus.OpusHead({
-      channelCount: CHANNELS,
-      sampleRate: RATE,
-    }),
-    pageSizeControl: { maxPackets: 1 }, // 1 packet per page = 20ms latency
-    crc: true,
-  });
-  opusEncoder.on("error", () => {});
-  opusEncoder.pipe(oggBitstream);
 
-  // Prime OGG headers before player starts reading
+  function startBrowserAudioBridge(): void {
-  opusEncoder.write(Buffer.alloc(BYTES_PER_FRAME, 0));
+    opusEncoder = new prism.opus.Encoder({
-  discordPlayer.playStream(oggBitstream);
+      rate: RATE,
-  discordPlayer.pause();
+      channels: CHANNELS,
+      frameSize: FRAME_SIZE,
+    });
+    const oggBitstream = new prism.opus.OggLogicalBitstream({
+      opusHead: new prism.opus.OpusHead({
+        channelCount: CHANNELS,
+        sampleRate: RATE,
+      }),
+      pageSizeControl: { maxPackets: 1 },
+      crc: true,
+    });
+    opusEncoder.on("error", () => {});
+    opusEncoder.pipe(oggBitstream);
+    opusEncoder.write(Buffer.alloc(BYTES_PER_FRAME, 0));
+    discordPlayer.playStream(oggBitstream);
+    discordPlayer.pause();
+    bridgePlayerPaused = true;
+  }
+
+  function ensureBrowserAudioBridge(): void {
+    if (discordPlayer.getStatus() === AudioPlayerStatus.Idle) {
+      startBrowserAudioBridge();
+    }
+  }
+
+  startBrowserAudioBridge();
 
   let pcmBuffer = Buffer.alloc(0);
   let lastBrowserAudioTime = 0;
-  let playerPaused = true;
-  const SILENCE_FRAME = Buffer.alloc(BYTES_PER_FRAME, 0);
 
   // Log level every 2 seconds
   let dbAccum = 0,
@@ -339,18 +351,19 @@ export async function startWebserver(
       dbAccum += rmsDb(frame);
       dbCount++;
 
-      if (playerPaused) {
+      ensureBrowserAudioBridge();
-        discordPlayer.unpause();
+      if (bridgePlayerPaused) {
-        playerPaused = false;
+        const unpaused = discordPlayer.unpause();
-        wsLogger.info("Transmitting — Discord indicator ON");
+        bridgePlayerPaused = false;
+        wsLogger.info({ unpaused }, "Transmitting — Discord indicator ON");
       }
     } else if (msSinceAudio < SILENCE_TAIL_MS && msSinceAudio > 0) {
       // Buffer drained but audio was recent — pad silence to avoid OGG gap
       frame = SILENCE_FRAME;
-    } else if (!playerPaused && msSinceAudio >= SILENCE_TAIL_MS) {
+    } else if (!bridgePlayerPaused && msSinceAudio >= SILENCE_TAIL_MS) {
       // No audio for a while — pause Discord indicator
       discordPlayer.pause();
-      playerPaused = true;
+      bridgePlayerPaused = true;
       wsLogger.info("Stopped — Discord indicator OFF");
       return;
     } else {