diff --git a/services/discord-gateway/src/modules/reaction-tracking/index.ts b/services/discord-gateway/src/modules/reaction-tracking/index.ts new file mode 100644 index 0000000..326e7f0 --- /dev/null +++ b/services/discord-gateway/src/modules/reaction-tracking/index.ts @@ -0,0 +1 @@ +export { registerReactionCapture } from "./reactionCapture.js"; diff --git a/services/discord-gateway/src/modules/reaction-tracking/reactionCapture.ts b/services/discord-gateway/src/modules/reaction-tracking/reactionCapture.ts new file mode 100644 index 0000000..6546f31 --- /dev/null +++ b/services/discord-gateway/src/modules/reaction-tracking/reactionCapture.ts @@ -0,0 +1,170 @@ +import { createChildLogger } from "@bete/shared/logger"; +import type { Client, MessageReaction, User } from "discord.js-selfbot-v13"; +import { getDatabase } from "../../shared/database/drizzle.js"; +import { reactionsTable } from "../../shared/database/schema.js"; +import { config } from "../../shared/config/config.js"; +import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js"; + +const logger = createChildLogger("reaction-tracking"); + +// ─── Helpers ───────────────────────────────────────────────────────────── + +function isMonitoredGuild(guildId: string | null | undefined): boolean { + if (!guildId) return false; + const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; + if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; + return guildIds.includes(guildId); +} + +function getEmojiIdentifier(reaction: MessageReaction): { + emoji: string; + emojiId: string | null; + animated: boolean; +} { + const emoji = reaction.emoji; + if (emoji.id) { + return { + emoji: emoji.name ?? emoji.id, + emojiId: emoji.id, + animated: Boolean((emoji as any).animated), + }; + } + return { + emoji: emoji.name ?? "unknown", + emojiId: null, + animated: false, + }; +} + +// ─── Event Handlers ────────────────────────────────────────────────────── + +async function handleReactionAdd( + reaction: MessageReaction, + user: User, +): Promise { + const guildId = reaction.message.guildId; + if (!isMonitoredGuild(guildId)) return; + if (user.bot) return; + + const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); + const now = Date.now(); + const id = `${reaction.message.id}-${emojiId ?? emoji}-${user.id}`; + + try { + const db = getDatabase(); + await (db as any).insert(reactionsTable).values({ + id, + message_id: reaction.message.id, + channel_id: reaction.message.channelId, + guild_id: guildId, + user_id: user.id, + username: user.username, + emoji, + emoji_id: emojiId, + animated, + reaction_type: "add", + created_at: now, + }).onConflictDoNothing(); + + logger.debug( + { messageId: reaction.message.id, emoji, userId: user.id }, + "Reaction recorded", + ); + } catch (error) { + logger.error( + { messageId: reaction.message.id, error: String(error) }, + "Failed to record reaction", + ); + } +} + +async function handleReactionRemove( + reaction: MessageReaction, + user: User, +): Promise { + const guildId = reaction.message.guildId; + if (!isMonitoredGuild(guildId)) return; + if (user.bot) return; + + const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); + const now = Date.now(); + const id = `${reaction.message.id}-${emojiId ?? emoji}-${user.id}`; + + try { + const db = getDatabase(); + await (db as any).insert(reactionsTable).values({ + id, + message_id: reaction.message.id, + channel_id: reaction.message.channelId, + guild_id: guildId, + user_id: user.id, + username: user.username, + emoji, + emoji_id: emojiId, + animated, + reaction_type: "remove", + created_at: now, + }).onConflictDoNothing(); + + logger.debug( + { messageId: reaction.message.id, emoji, userId: user.id }, + "Reaction removal recorded", + ); + } catch (error) { + logger.error( + { messageId: reaction.message.id, error: String(error) }, + "Failed to record reaction removal", + ); + } +} + +// ─── Registration ──────────────────────────────────────────────────────── + +export function registerReactionCapture( + client: Client, + eventBroadcaster: EventBroadcaster, +): void { + logger.info("Registering reaction capture"); + + client.on("messageReactionAdd", async (reaction, user) => { + await handleReactionAdd(reaction, user); + + const guildId = reaction.message.guildId; + if (!isMonitoredGuild(guildId)) return; + + const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); + + eventBroadcaster.reactionAdded({ + message_id: reaction.message.id, + channel_id: reaction.message.channelId, + guild_id: guildId, + user_id: user.id, + username: user.username, + emoji, + emoji_id: emojiId, + animated, + created_at: Date.now(), + }).catch(() => {}); + }); + + client.on("messageReactionRemove", async (reaction, user) => { + await handleReactionRemove(reaction, user); + + const guildId = reaction.message.guildId; + if (!isMonitoredGuild(guildId)) return; + + const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); + + eventBroadcaster.reactionRemoved({ + message_id: reaction.message.id, + channel_id: reaction.message.channelId, + guild_id: guildId, + user_id: user.id, + username: user.username, + emoji, + emoji_id: emojiId, + animated, + created_at: Date.now(), + }).catch(() => {}); + }); +} diff --git a/services/discord-gateway/src/modules/voice-recording/muxer.ts b/services/discord-gateway/src/modules/voice-recording/muxer.ts new file mode 100644 index 0000000..8a0d6e3 --- /dev/null +++ b/services/discord-gateway/src/modules/voice-recording/muxer.ts @@ -0,0 +1,194 @@ +import { createChildLogger } from "@bete/shared/logger"; +import { eq } from "drizzle-orm"; +import type { NodePgDatabase } from "drizzle-orm/node-postgres"; +import { getDatabase } from "../../shared/database/drizzle.js"; +import { muxerJobsTable } from "../../shared/database/schema.js"; +import { config } from "../../shared/config/config.js"; +import { buildMuxFfmpegArgs, runFfmpeg } from "./ffmpegProcess.js"; + +const logger = createChildLogger("muxer"); + +// ─── Types ─────────────────────────────────────────────────────────────── + +export interface MuxerJobData { + inputs: string[]; + output: string; + guildId: string; + channelId: string; + sessionId: string; +} + +// ─── State ─────────────────────────────────────────────────────────────── + +let pollTimer: ReturnType | null = null; + +// ─── Public API ────────────────────────────────────────────────────────── + +/** + * Enqueue a muxer job that merges multiple OGG files into one. + */ +export async function enqueueMuxerJob(data: MuxerJobData): Promise { + const db = getDatabase() as unknown as NodePgDatabase; + const id = `${data.sessionId}-${Date.now()}`; + + try { + await db.insert(muxerJobsTable).values({ + id, + data: JSON.stringify(data), + status: "pending", + attempts: 0, + maxAttempts: 3, + createdAt: Date.now(), + updatedAt: Date.now(), + }); + logger.info({ id, sessionId: data.sessionId }, "Muxer job enqueued"); + } catch (error) { + logger.error( + { id, error: error instanceof Error ? error.message : String(error) }, + "Failed to enqueue muxer job", + ); + } +} + +/** + * Start the background worker that polls for pending muxer jobs. + */ +export function startMuxerWorker(): void { + if (pollTimer) return; + logger.info("Starting muxer worker (interval: 10s)"); + + pollTimer = setInterval(() => { + processNextJobs().catch((err: unknown) => { + logger.error( + { error: String(err) }, + "Muxer worker tick failed", + ); + }); + }, 10_000); +} + +/** + * Stop the background worker. + */ +export function stopMuxerWorker(): void { + if (!pollTimer) return; + clearInterval(pollTimer); + pollTimer = null; + logger.info("Muxer worker stopped"); +} + +// ─── Internal ──────────────────────────────────────────────────────────── + +async function processNextJobs(): Promise { + const db = getDatabase() as unknown as NodePgDatabase; + + try { + const jobs = await db + .select() + .from(muxerJobsTable) + .where(eq(muxerJobsTable.status, "pending")) + .limit(5); + + if (jobs.length === 0) return; + + const results = await Promise.allSettled( + jobs.map((job) => processJob(db, job)), + ); + + for (let i = 0; i < results.length; i++) { + const result = results[i]; + if (result.status === "rejected") { + logger.error( + { jobId: jobs[i].id, error: result.reason }, + "Muxer job failed", + ); + } + } + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to fetch pending muxer jobs", + ); + } +} + +async function processJob( + db: NodePgDatabase, + job: typeof muxerJobsTable.$inferSelect, +): Promise { + // Mark as processing + await db + .update(muxerJobsTable) + .set({ status: "processing", updatedAt: Date.now() }) + .where(eq(muxerJobsTable.id, job.id)); + + try { + const data = JSON.parse(job.data) as MuxerJobData; + + if (!data.inputs || data.inputs.length < 2) { + throw new Error(`Muxer job ${job.id} needs at least 2 inputs, got ${data.inputs?.length ?? 0}`); + } + + logger.info( + { jobId: job.id, inputs: data.inputs.length, output: data.output }, + "Processing muxer job", + ); + + // Build filter: concat all inputs with a crossfade or simple concat + // We use amix for mixing multiple audio streams (not sequential concat) + const inputLabels = data.inputs.map((_, i) => `[${i}:a]`); + const filterComplex = `${inputLabels.join("")}amix=inputs=${data.inputs.length}:duration=first:dropout_transition=2[out]`; + + const args = buildMuxFfmpegArgs({ + inputs: data.inputs, + filter: filterComplex, + output: data.output, + codec: "libopus", + audioFrequency: 48000, + audioChannels: 2, + }); + + await runFfmpeg(args); + + // Mark as completed + await db + .update(muxerJobsTable) + .set({ status: "completed", updatedAt: Date.now() }) + .where(eq(muxerJobsTable.id, job.id)); + + logger.info( + { jobId: job.id, output: data.output }, + "Muxer job completed", + ); + } catch (error) { + const errMsg = error instanceof Error ? error.message : String(error); + const newAttempts = job.attempts + 1; + + if (newAttempts >= job.maxAttempts) { + await db + .update(muxerJobsTable) + .set({ + status: "failed", + error: errMsg, + attempts: newAttempts, + updatedAt: Date.now(), + }) + .where(eq(muxerJobsTable.id, job.id)); + logger.error({ jobId: job.id, error: errMsg }, "Muxer job failed permanently"); + } else { + await db + .update(muxerJobsTable) + .set({ + status: "pending", + error: errMsg, + attempts: newAttempts, + updatedAt: Date.now(), + }) + .where(eq(muxerJobsTable.id, job.id)); + logger.warn( + { jobId: job.id, error: errMsg, attempt: newAttempts }, + "Muxer job will be retried", + ); + } + } +}