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

feat: enhance media handling and audio processing logic

Browse Source
This commit is contained in:
MythEclipse
2026-05-15 22:23:29 +07:00
parent 6ac4a5c11a
commit e32e092596
10 changed files with 63 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/moderation/aiAnalyzer.ts
+1 -29
View File
@@ -110,11 +110,6 @@ async function processBatch(
}
conversationErrorCooldown.delete(conversationKey);
logger.info(
{ conversationKey, count: messages.length },
"Batch analysis complete",
);
} catch (error) {
lastError = error instanceof Error ? error.message : String(error);
conversationErrorCooldown.set(
@@ -173,20 +168,12 @@ async function runAnalysisInWorker(
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;
}
@@ -202,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;
}
@@ -230,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
@@ -260,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);
@@ -281,12 +262,7 @@ export function getAnalysisQueueStatus(): AnalysisQueueStatus {
* Starts the pending AI analysis recovery worker
*/
export function startPendingAIAnalysisWorker(): void {
if (!config.AI_ANALYSIS_ENABLED) {
logger.info("AI analysis disabled");
return;
}
logger.info("AI analysis worker started");
if (!config.AI_ANALYSIS_ENABLED) return;
setInterval(async () => {
try {
@@ -310,10 +286,6 @@ export function startPendingAIAnalysisWorker(): void {
continue;
}
logger.debug(
{ conversationKey: key },
"Recovering pending conversation",
);
scheduleConversationAnalysis(key);
}
} catch (error) {
src/moderation/attachmentUploader.ts
+1 -12
View File
@@ -82,12 +82,7 @@ export async function uploadAttachmentToPicser(
},
);
const parsed = parseUploadResponse(response);
logger.info(
{ filename, url: parsed.url },
"Attachment uploaded successfully",
);
return parsed;
return parseUploadResponse(response);
} 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
-14
View File
@@ -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
-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 {
@@ -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
+1 -27
View File
@@ -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;
logger.debug(
{ messageId: message.id, channelId: message.channel_id, inserted },
inserted ? "Message inserted for capture" : "Message already captured",
);
return inserted;
return rows.length > 0;
} 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
+7 -2
View File
@@ -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() {
this.player.unpause();
public unpause(): boolean {
return this.player.unpause();
}
public stop() {
src/routes/analysisRoutes.ts
-5
View File
@@ -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
+1 -9
View File
@@ -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) => {
logger.info(
{ guildId, channelId, messagesSync: count },
"Backlog sync complete",
);
})
.then(() => {})
.catch((error) => {
logger.warn(
{
src/webserver.ts
+40 -27
View File
@@ -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({
rate: RATE,
channels: CHANNELS,
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);
let opusEncoder: prism.opus.Encoder;
let bridgePlayerPaused = true;
const SILENCE_FRAME = Buffer.alloc(BYTES_PER_FRAME, 0);
// Prime OGG headers before player starts reading
opusEncoder.write(Buffer.alloc(BYTES_PER_FRAME, 0));
discordPlayer.playStream(oggBitstream);
discordPlayer.pause();
function startBrowserAudioBridge(): void {
opusEncoder = new prism.opus.Encoder({
rate: RATE,
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) {
discordPlayer.unpause();
playerPaused = false;
wsLogger.info("Transmitting — Discord indicator ON");
ensureBrowserAudioBridge();
if (bridgePlayerPaused) {
const unpaused = discordPlayer.unpause();
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 {