Compare commits

...
2 Commits
14 changed files with 261 additions and 211 deletions
+15 -2
View File
@@ -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. - Node.js versi modern yang kompatibel dengan TypeScript dan Vite.
- pnpm 10.x. Repo ini dipin ke `pnpm@10.25.0`. - 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`). - Native audio dependencies dapat dibuild di mesin lokal (`@discordjs/opus`, `better-sqlite3`, `sodium-native`).
Install FFmpeg: Install FFmpeg:
@@ -21,6 +22,14 @@ sudo apt install ffmpeg
sudo pacman -S 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 ## Setup
```bash ```bash
@@ -76,6 +85,9 @@ pnpm run test
# Build frontend + TypeScript # Build frontend + TypeScript
pnpm run build pnpm run build
# Install external yt-dlp CLI for YouTube/search/Spotify track playback
pnpm run install:yt-dlp
``` ```
## Database ## Database
@@ -104,7 +116,8 @@ pnpm run db:studio
- Attachment capture dan upload ke endpoint Picser. - Attachment capture dan upload ke endpoint Picser.
- SQLite/PostgreSQL via Drizzle ORM. - SQLite/PostgreSQL via Drizzle ORM.
- REST API dan WebSocket untuk dashboard. - 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. - Metrics Prometheus di endpoint server.
- Retry dengan backoff untuk operasi eksternal. - Retry dengan backoff untuk operasi eksternal.
- AI moderation analysis opsional via konfigurasi `AI_*`. - AI moderation analysis opsional via konfigurasi `AI_*`.
+2 -1
View File
@@ -18,7 +18,8 @@
"db:generate": "drizzle-kit generate", "db:generate": "drizzle-kit generate",
"db:migrate": "drizzle-kit migrate", "db:migrate": "drizzle-kit migrate",
"db:migrate:programmatic": "tsx src/database/migrate.ts", "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": { "dependencies": {
"@dank074/discord-video-stream": "workspace:*", "@dank074/discord-video-stream": "workspace:*",
+13 -11
View File
File diff suppressed because one or more lines are too long
+34
View File
@@ -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
+94
View File
@@ -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));
});
+53 -88
View File
@@ -1,9 +1,7 @@
import { Worker } from "node:worker_threads";
import { config } from "../config"; import { config } from "../config";
import { createChildLogger } from "../logger"; import { createChildLogger } from "../logger";
import { buildConversationPromptMessages } from "./conversationContext";
import { runModerationAnalysis } from "./llmModerationClient";
import { import {
getConversationContextBefore,
getMessageById, getMessageById,
getPendingConversationKeys, getPendingConversationKeys,
getPendingMessagesByConversation, getPendingMessagesByConversation,
@@ -38,9 +36,15 @@ const MAX_ACTIVE_REQUESTS = 1;
const DEBOUNCE_MS = 1500; const DEBOUNCE_MS = 1500;
const RECOVERY_INTERVAL_MS = 15000; const RECOVERY_INTERVAL_MS = 15000;
const ERROR_COOLDOWN_MS = 30000; const ERROR_COOLDOWN_MS = 30000;
const MAX_CONTEXT_TOKENS = 8000;
const MAX_BATCH_SIZE = 25; 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) * Gets the conversation key for a message (thread_id or channel_id)
*/ */
@@ -86,68 +90,37 @@ async function processBatch(
activeRequests++; activeRequests++;
conversationProcessing.add(conversationKey); conversationProcessing.add(conversationKey);
try { 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); 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) { } catch (error) {
lastError = error instanceof Error ? error.message : String(error); lastError = error instanceof Error ? error.message : String(error);
conversationErrorCooldown.set(
conversationKey,
Date.now() + ERROR_COOLDOWN_MS,
);
logger.error( logger.error(
{ conversationKey, error: lastError }, { conversationKey, error: lastError },
"Batch analysis failed", "Analysis worker failed",
); );
// Mark all messages in batch as error
for (const msg of messages) { for (const msg of messages) {
const row = await updateMessageAIAnalysis(msg.id, { const row = await updateMessageAIAnalysis(msg.id, {
status: "error", status: "error",
@@ -158,42 +131,49 @@ async function processBatch(
analyzedAt: Date.now(), analyzedAt: Date.now(),
error: lastError, 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 { } finally {
activeRequests--; activeRequests--;
conversationProcessing.delete(conversationKey); 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 * Debounced analysis trigger for a conversation
*/ */
function scheduleConversationAnalysis(conversationKey: string): void { function scheduleConversationAnalysis(conversationKey: string): void {
// Skip if already processing // Skip if already processing
if (conversationProcessing.has(conversationKey)) { if (conversationProcessing.has(conversationKey)) {
logger.debug(
{ conversationKey },
"Conversation already processing, skipping schedule",
);
return; return;
} }
// Skip if in error cooldown // Skip if in error cooldown
const cooldownUntil = conversationErrorCooldown.get(conversationKey); const cooldownUntil = conversationErrorCooldown.get(conversationKey);
if (cooldownUntil && Date.now() < cooldownUntil) { if (cooldownUntil && Date.now() < cooldownUntil) {
logger.debug(
{ conversationKey, cooldownMs: cooldownUntil - Date.now() },
"Conversation in error cooldown, skipping schedule",
);
return; return;
} }
@@ -209,10 +189,6 @@ function scheduleConversationAnalysis(conversationKey: string): void {
// If activeRequests >= MAX_ACTIVE_REQUESTS, requeue instead of waiting // If activeRequests >= MAX_ACTIVE_REQUESTS, requeue instead of waiting
if (activeRequests >= MAX_ACTIVE_REQUESTS) { if (activeRequests >= MAX_ACTIVE_REQUESTS) {
logger.debug(
{ conversationKey, activeRequests },
"Max active requests reached, requeuing conversation",
);
scheduleConversationAnalysis(conversationKey); scheduleConversationAnalysis(conversationKey);
return; return;
} }
@@ -237,7 +213,6 @@ function scheduleConversationAnalysis(conversationKey: string): void {
export async function queueMessageAnalysis(messageId: string): Promise<void> { export async function queueMessageAnalysis(messageId: string): Promise<void> {
if (!config.AI_ANALYSIS_ENABLED) return; if (!config.AI_ANALYSIS_ENABLED) return;
logger.debug({ messageId }, "Queueing message for analysis");
try { try {
// Look up the message to get its conversation key // 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 { export function queueConversationAnalysis(conversationKey: string): void {
if (!config.AI_ANALYSIS_ENABLED) return; if (!config.AI_ANALYSIS_ENABLED) return;
logger.debug({ conversationKey }, "Queueing conversation for analysis");
// Schedule debounced analysis // Schedule debounced analysis
scheduleConversationAnalysis(conversationKey); scheduleConversationAnalysis(conversationKey);
@@ -288,12 +262,7 @@ export function getAnalysisQueueStatus(): AnalysisQueueStatus {
* Starts the pending AI analysis recovery worker * Starts the pending AI analysis recovery worker
*/ */
export function startPendingAIAnalysisWorker(): void { 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 () => { setInterval(async () => {
try { try {
@@ -317,10 +286,6 @@ export function startPendingAIAnalysisWorker(): void {
continue; continue;
} }
logger.debug(
{ conversationKey: key },
"Recovering pending conversation",
);
scheduleConversationAnalysis(key); scheduleConversationAnalysis(key);
} }
} catch (error) { } catch (error) {
+1 -12
View File
@@ -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) { } catch (error) {
logger.error( logger.error(
{ {
@@ -127,8 +122,6 @@ export async function processAttachmentUpload(
filename: string, filename: string,
): Promise<void> { ): Promise<void> {
try { try {
logger.info({ attachmentId, filename }, "Starting attachment upload");
const buffer = await downloadDiscordAttachment(discordUrl); const buffer = await downloadDiscordAttachment(discordUrl);
const sizeMb = buffer.length / (1024 * 1024); const sizeMb = buffer.length / (1024 * 1024);
@@ -141,10 +134,6 @@ export async function processAttachmentUpload(
const result = await uploadAttachmentToPicser(buffer, filename); const result = await uploadAttachmentToPicser(buffer, filename);
await updateAttachmentAsUploaded(attachmentId, result.url, Date.now()); await updateAttachmentAsUploaded(attachmentId, result.url, Date.now());
logger.info(
{ attachmentId, uploadedUrl: result.url },
"Attachment upload completed",
);
} catch (error) { } catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error); const errorMsg = error instanceof Error ? error.message : String(error);
await updateAttachmentAsFailedUpload(attachmentId, errorMsg); await updateAttachmentAsFailedUpload(attachmentId, errorMsg);
-14
View File
@@ -73,11 +73,6 @@ export async function syncBacklogMessages(client: Client): Promise<void> {
await syncSelectedChannelBacklog(client, guild.id, config.TEXT_CHANNEL_ID); await syncSelectedChannelBacklog(client, guild.id, config.TEXT_CHANNEL_ID);
return; return;
} }
logger.info(
{ guildId: guild.id },
"Backlog sync ready (will sync on-demand per selected channel)",
);
} }
export async function syncSelectedChannelBacklog( export async function syncSelectedChannelBacklog(
@@ -102,17 +97,8 @@ export async function syncSelectedChannelBacklog(
} }
const cutoffTime = Date.now() - config.BACKLOG_SYNC_HOURS * 60 * 60 * 1000; 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 { try {
const count = await syncChannelMessages(channel, cutoffTime); const count = await syncChannelMessages(channel, cutoffTime);
logger.info(
{ channelId, count },
"Backlog sync completed for selected channel",
);
return count; return count;
} catch (error) { } catch (error) {
logger.warn( logger.warn(
-13
View File
@@ -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 { export function registerMessageCapture(client: Client): void {
@@ -206,8 +197,6 @@ export function registerMessageCapture(client: Client): void {
deleted_at: deletedAt, deleted_at: deletedAt,
}); });
} }
logger.info({ messageId: message.id }, "Message deletion captured");
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -218,6 +207,4 @@ export function registerMessageCapture(client: Client): void {
); );
} }
}); });
logger.info("Message capture handlers registered");
} }
+1 -27
View File
@@ -61,11 +61,6 @@ export async function insertMessage(message: MessageRecord): Promise<void> {
try { try {
const database = db(); const database = db();
await database.insert(messagesTable).values(message).onConflictDoNothing(); await database.insert(messagesTable).values(message).onConflictDoNothing();
logger.debug(
{ messageId: message.id, channelId: message.channel_id },
"Message inserted",
);
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -94,12 +89,7 @@ export async function upsertMessageForCapture(
.onConflictDoNothing() .onConflictDoNothing()
.returning({ id: messagesTable.id }); .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) { } catch (error) {
logger.error( logger.error(
{ {
@@ -134,8 +124,6 @@ export async function updateMessageAsEdited(
ai_error: null, ai_error: null,
}) })
.where(eq(messagesTable.id, messageId)); .where(eq(messagesTable.id, messageId));
logger.debug({ messageId }, "Message marked as edited");
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -161,8 +149,6 @@ export async function updateMessageAsDeleted(
type: "deleted", type: "deleted",
}) })
.where(eq(messagesTable.id, messageId)); .where(eq(messagesTable.id, messageId));
logger.debug({ messageId }, "Message marked as deleted");
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -217,11 +203,6 @@ export async function insertAttachment(
.insert(attachmentsTable) .insert(attachmentsTable)
.values(attachment) .values(attachment)
.onConflictDoNothing(); .onConflictDoNothing();
logger.debug(
{ attachmentId: attachment.id, messageId: attachment.message_id },
"Attachment inserted",
);
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -282,11 +263,6 @@ export async function updateAttachmentAsUploaded(
uploaded_at: uploadedAt, uploaded_at: uploadedAt,
}) })
.where(eq(attachmentsTable.id, attachmentId)); .where(eq(attachmentsTable.id, attachmentId));
logger.debug(
{ attachmentId, uploadedUrl },
"Attachment marked as uploaded",
);
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
@@ -312,8 +288,6 @@ export async function updateAttachmentAsFailedUpload(
upload_error: error, upload_error: error,
}) })
.where(eq(attachmentsTable.id, attachmentId)); .where(eq(attachmentsTable.id, attachmentId));
logger.debug({ attachmentId, error }, "Attachment marked as failed upload");
} catch (error) { } catch (error) {
logger.error( logger.error(
{ {
+7 -2
View File
@@ -41,14 +41,19 @@ export class DiscordPlayer {
}); });
this.player.play(resource); this.player.play(resource);
this.connection?.subscribe(this.player);
}
public getStatus(): AudioPlayerStatus {
return this.player.state.status;
} }
public pause() { public pause() {
this.player.pause(true); this.player.pause(true);
} }
public unpause() { public unpause(): boolean {
this.player.unpause(); return this.player.unpause();
} }
public stop() { public stop() {
-5
View File
@@ -1,15 +1,12 @@
import type { Router } from "express"; import type { Router } from "express";
import express from "express"; import express from "express";
import { AppError } from "../errors"; import { AppError } from "../errors";
import { createChildLogger } from "../logger";
import { import {
getAnalysisQueueStatus, getAnalysisQueueStatus,
queueMessageAnalysis, queueMessageAnalysis,
} from "../moderation/aiAnalyzer"; } from "../moderation/aiAnalyzer";
import { getMessageById } from "../moderation/messageStore"; import { getMessageById } from "../moderation/messageStore";
const logger = createChildLogger("analysis-routes");
export function createAnalysisRoutes(): Router { export function createAnalysisRoutes(): Router {
const router = express.Router(); const router = express.Router();
@@ -41,8 +38,6 @@ export function createAnalysisRoutes(): Router {
// Queue for analysis // Queue for analysis
await queueMessageAnalysis(id); await queueMessageAnalysis(id);
logger.info({ messageId: id }, "Message queued for re-analysis");
res.json({ res.json({
success: true, success: true,
messageId: id, messageId: id,
+1 -9
View File
@@ -45,7 +45,6 @@ export function createSyncRoutes(client: Client): Router {
} }
if (shouldSkipRecentBacklogSync(guildId, channelId)) { if (shouldSkipRecentBacklogSync(guildId, channelId)) {
logger.debug({ guildId, channelId }, "Skipping recent backlog sync");
res.json({ res.json({
success: true, success: true,
channelId, channelId,
@@ -56,15 +55,8 @@ export function createSyncRoutes(client: Client): Router {
return; return;
} }
logger.info({ guildId, channelId }, "Queueing backlog sync");
syncSelectedChannelBacklog(client, guildId, channelId) syncSelectedChannelBacklog(client, guildId, channelId)
.then((count) => { .then(() => {})
logger.info(
{ guildId, channelId, messagesSync: count },
"Backlog sync complete",
);
})
.catch((error) => { .catch((error) => {
logger.warn( logger.warn(
{ {
+40 -27
View File
@@ -4,6 +4,7 @@ import path from "node:path";
import type { Client } from "discord.js-selfbot-v13"; import type { Client } from "discord.js-selfbot-v13";
import express from "express"; import express from "express";
import helmet from "helmet"; import helmet from "helmet";
import { AudioPlayerStatus } from "@discordjs/voice";
import * as prism from "prism-media"; import * as prism from "prism-media";
import { WebSocketServer } from "ws"; import { WebSocketServer } from "ws";
import { AppError } from "./errors"; 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 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 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 pcmBuffer = Buffer.alloc(0);
let lastBrowserAudioTime = 0; let lastBrowserAudioTime = 0;
let playerPaused = true;
const SILENCE_FRAME = Buffer.alloc(BYTES_PER_FRAME, 0);
// Log level every 2 seconds // Log level every 2 seconds
let dbAccum = 0, let dbAccum = 0,
@@ -339,18 +351,19 @@ export async function startWebserver(
dbAccum += rmsDb(frame); dbAccum += rmsDb(frame);
dbCount++; 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) { } else if (msSinceAudio < SILENCE_TAIL_MS && msSinceAudio > 0) {
// Buffer drained but audio was recent — pad silence to avoid OGG gap // Buffer drained but audio was recent — pad silence to avoid OGG gap
frame = SILENCE_FRAME; 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 // No audio for a while — pause Discord indicator
discordPlayer.pause(); discordPlayer.pause();
playerPaused = true; bridgePlayerPaused = true;
wsLogger.info("Stopped — Discord indicator OFF"); wsLogger.info("Stopped — Discord indicator OFF");
return; return;
} else { } else {