chore(auto): task completed - unknown

This commit is contained in:
MythEclipse
2026-06-13 13:35:11 +07:00
parent d569ffa2db
commit 2e14b5e4ed
22 changed files with 516 additions and 172 deletions
@@ -0,0 +1,30 @@
-- Migration 0010: Add message_reactions and message_edits tables
-- Generated from schema: pgReactionsTable, pgMessageEditsTable
CREATE TABLE IF NOT EXISTS "message_reactions" (
"id" text PRIMARY KEY,
"message_id" text NOT NULL,
"channel_id" text NOT NULL,
"guild_id" text NOT NULL,
"user_id" text NOT NULL,
"username" text NOT NULL,
"emoji" text NOT NULL,
"emoji_id" text,
"animated" boolean NOT NULL DEFAULT false,
"reaction_type" text NOT NULL CHECK ("reaction_type" IN ('add', 'remove')),
"created_at" bigint NOT NULL
);
CREATE INDEX IF NOT EXISTS "idx_reactions_message_id" ON "message_reactions" ("message_id");
CREATE INDEX IF NOT EXISTS "idx_reactions_user_id" ON "message_reactions" ("user_id");
CREATE INDEX IF NOT EXISTS "idx_reactions_guild_created" ON "message_reactions" ("guild_id", "created_at");
CREATE TABLE IF NOT EXISTS "message_edits" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid(),
"message_id" text NOT NULL,
"old_content" text NOT NULL,
"edited_at" bigint NOT NULL
);
CREATE INDEX IF NOT EXISTS "idx_message_edits_message_id" ON "message_edits" ("message_id");
CREATE INDEX IF NOT EXISTS "idx_message_edits_edited_at" ON "message_edits" ("edited_at");
@@ -71,6 +71,13 @@
"when": 1781316000000, "when": 1781316000000,
"tag": "0009_add_reply_forward_crosspost", "tag": "0009_add_reply_forward_crosspost",
"breakpoints": true "breakpoints": true
},
{
"idx": 10,
"version": "7",
"when": 1781390000000,
"tag": "0010_add_reactions_and_edit_history",
"breakpoints": true
} }
] ]
} }
@@ -4,16 +4,29 @@ import { Client } from "discord.js-selfbot-v13";
import { inArray, lt } from "drizzle-orm"; import { inArray, lt } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres"; import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import { startPendingAIAnalysisWorker } from "../modules/ai-moderation/aiAnalyzer.js"; import { startPendingAIAnalysisWorker } from "../modules/ai-moderation/aiAnalyzer.js";
import { registerChannelTopicCapture } from "../modules/channel-topic/index.js";
import { CommandHandler } from "../modules/command-handler/commandHandler.js"; import { CommandHandler } from "../modules/command-handler/commandHandler.js";
import { import {
EventBroadcaster, EventBroadcaster,
RedisEventPublisher, RedisEventPublisher,
} from "../modules/event-broadcaster/index.js"; } from "../modules/event-broadcaster/index.js";
import {
startMetricsServer,
stopMetricsServer,
} from "../modules/gateway-metrics/index.js";
import { registerGuildMemberEvents } from "../modules/guild-member-events/index.js";
import { import {
registerMessageCapture, registerMessageCapture,
setEventBroadcaster as setMessageCaptureEventBroadcaster, setEventBroadcaster as setMessageCaptureEventBroadcaster,
} from "../modules/message-capture/messageCapture.js"; } from "../modules/message-capture/messageCapture.js";
import { getExpiredMessages } from "../modules/message-capture/messageStore.js"; import { getExpiredMessages } from "../modules/message-capture/messageStore.js";
import { registerReactionCapture } from "../modules/reaction-tracking/index.js";
import { registerThreadCapture } from "../modules/thread-tracking/index.js";
import { registerPresenceCapture } from "../modules/user-presence/index.js";
import {
startMuxerWorker,
stopMuxerWorker,
} from "../modules/voice-recording/muxer.js";
import { setEventBroadcaster as setRecorderEventBroadcaster } from "../modules/voice-recording/recorder.js"; import { setEventBroadcaster as setRecorderEventBroadcaster } from "../modules/voice-recording/recorder.js";
import { VoiceController } from "../modules/voice-recording/voiceController.js"; import { VoiceController } from "../modules/voice-recording/voiceController.js";
import { config } from "../shared/config/config.js"; import { config } from "../shared/config/config.js";
@@ -212,6 +225,7 @@ export async function initializeDiscordGateway() {
client, client,
eventBroadcaster, eventBroadcaster,
commandHandler, commandHandler,
stopMetricsServer,
}); });
try { try {
@@ -252,6 +266,16 @@ export async function initializeDiscordGateway() {
registerMessageCapture(client); registerMessageCapture(client);
startPendingAIAnalysisWorker(client, eventBroadcaster); startPendingAIAnalysisWorker(client, eventBroadcaster);
// Register new event captures
registerReactionCapture(client, eventBroadcaster);
registerThreadCapture(client, eventBroadcaster);
registerPresenceCapture(client, eventBroadcaster);
registerChannelTopicCapture(client, eventBroadcaster);
registerGuildMemberEvents(client, eventBroadcaster);
// Start background workers
startMuxerWorker();
// Start command handler after Discord is ready // Start command handler after Discord is ready
commandHandler.start(client, voiceController); commandHandler.start(client, voiceController);
logger.info("Command handler started"); logger.info("Command handler started");
@@ -282,6 +306,9 @@ export async function initializeDiscordGateway() {
gracefulShutdown("unhandledRejection"); gracefulShutdown("unhandledRejection");
}); });
// Start metrics server
startMetricsServer();
logger.info("Calling Discord client.login"); logger.info("Calling Discord client.login");
client client
.login(token) .login(token)
@@ -2,11 +2,14 @@ import type { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13"; import type { Client } from "discord.js-selfbot-v13";
import type { CommandHandler } from "../modules/command-handler/commandHandler.js"; import type { CommandHandler } from "../modules/command-handler/commandHandler.js";
import type { EventBroadcaster } from "../modules/event-broadcaster/index.js"; import type { EventBroadcaster } from "../modules/event-broadcaster/index.js";
import { stopMetricsServer } from "../modules/gateway-metrics/index.js";
import { stopMuxerWorker } from "../modules/voice-recording/muxer.js";
import type { VoiceController } from "../modules/voice-recording/voiceController.js"; import type { VoiceController } from "../modules/voice-recording/voiceController.js";
import type { closeDatabase } from "../shared/database/drizzle.js"; import type { closeDatabase } from "../shared/database/drizzle.js";
type Logger = ReturnType<typeof createChildLogger>; type Logger = ReturnType<typeof createChildLogger>;
type CloseDatabase = typeof closeDatabase; type CloseDatabase = typeof closeDatabase;
type StopMetricsServer = typeof stopMetricsServer;
export interface GracefulShutdownOptions { export interface GracefulShutdownOptions {
logger: Logger; logger: Logger;
@@ -15,6 +18,7 @@ export interface GracefulShutdownOptions {
client: Client; client: Client;
eventBroadcaster: EventBroadcaster; eventBroadcaster: EventBroadcaster;
commandHandler: CommandHandler; commandHandler: CommandHandler;
stopMetricsServer?: StopMetricsServer;
} }
export function createGracefulShutdown(options: GracefulShutdownOptions) { export function createGracefulShutdown(options: GracefulShutdownOptions) {
@@ -30,6 +34,8 @@ export function createGracefulShutdown(options: GracefulShutdownOptions) {
options.logger.info({ signal }, "Graceful shutdown initiated"); options.logger.info({ signal }, "Graceful shutdown initiated");
try { try {
options.stopMetricsServer?.();
stopMuxerWorker();
options.logger.info("Closing database..."); options.logger.info("Closing database...");
await options.closeDatabase(); await options.closeDatabase();
options.logger.info("Database closed"); options.logger.info("Database closed");
@@ -7,8 +7,11 @@ const logger = createChildLogger("channel-topic");
function isMonitoredGuild(guildId: string | null | undefined): boolean { function isMonitoredGuild(guildId: string | null | undefined): boolean {
if (!guildId) return false; if (!guildId) return false;
const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as
if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; | string[]
| undefined;
if (!guildIds || guildIds.length === 0)
return config.MONITOR_GUILD_ID === guildId;
return guildIds.includes(guildId); return guildIds.includes(guildId);
} }
@@ -184,6 +184,106 @@ export class EventBroadcaster {
}); });
} }
async reactionAdded(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing reaction_added");
await this.publisher.publish(EventChannels.REACTION_ADDED, {
type: "reaction_added",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async reactionRemoved(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing reaction_removed");
await this.publisher.publish(EventChannels.REACTION_REMOVED, {
type: "reaction_removed",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async threadCreated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing thread_created");
await this.publisher.publish(EventChannels.THREAD_CREATED, {
type: "thread_created",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async threadDeleted(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing thread_deleted");
await this.publisher.publish(EventChannels.THREAD_DELETED, {
type: "thread_deleted",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async threadUpdated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing thread_updated");
await this.publisher.publish(EventChannels.THREAD_UPDATED, {
type: "thread_updated",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async channelTopicUpdated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing channel_topic_updated");
await this.publisher.publish(EventChannels.CHANNEL_TOPIC_UPDATED, {
type: "channel_topic_updated",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async presenceUpdated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing presence_updated");
await this.publisher.publish(EventChannels.PRESENCE_UPDATED, {
type: "presence_updated",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async guildMemberAdded(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing guild_member_added");
await this.publisher.publish(EventChannels.GUILD_MEMBER_ADDED, {
type: "guild_member_added",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async guildMemberRemoved(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing guild_member_removed");
await this.publisher.publish(EventChannels.GUILD_MEMBER_REMOVED, {
type: "guild_member_removed",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async voiceAnalyzed(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing voice_analyzed");
await this.publisher.publish(EventChannels.VOICE_ANALYZED, {
type: "voice_analyzed",
data,
timestamp: Date.now(),
source: "discord-gateway",
});
}
async analysisQueueStatus(data: unknown): Promise<void> { async analysisQueueStatus(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing analysis_queue_status"); this.logger.debug({ data }, "Publishing analysis_queue_status");
await this.publisher.publish(EventChannels.ANALYSIS_QUEUE_STATUS, { await this.publisher.publish(EventChannels.ANALYSIS_QUEUE_STATUS, {
@@ -2,11 +2,21 @@ import {
DISCORD_ANALYSIS_QUEUE_STATUS, DISCORD_ANALYSIS_QUEUE_STATUS,
DISCORD_ATTACHMENT_CREATED, DISCORD_ATTACHMENT_CREATED,
DISCORD_ATTACHMENT_UPLOADED, DISCORD_ATTACHMENT_UPLOADED,
DISCORD_CHANNEL_TOPIC_UPDATED,
DISCORD_GUILD_MEMBER_ADDED,
DISCORD_GUILD_MEMBER_REMOVED,
DISCORD_MESSAGE_ANALYZED, DISCORD_MESSAGE_ANALYZED,
DISCORD_MESSAGE_CREATED, DISCORD_MESSAGE_CREATED,
DISCORD_MESSAGE_DELETED, DISCORD_MESSAGE_DELETED,
DISCORD_MESSAGE_UPDATED, DISCORD_MESSAGE_UPDATED,
DISCORD_PRESENCE_UPDATED,
DISCORD_REACTION_ADDED,
DISCORD_REACTION_REMOVED,
DISCORD_THREAD_CREATED,
DISCORD_THREAD_DELETED,
DISCORD_THREAD_UPDATED,
DISCORD_VOICE_ACTIVE_USER, DISCORD_VOICE_ACTIVE_USER,
DISCORD_VOICE_ANALYZED,
DISCORD_VOICE_PCM, DISCORD_VOICE_PCM,
DISCORD_VOICE_STARTED, DISCORD_VOICE_STARTED,
DISCORD_VOICE_STOPPED, DISCORD_VOICE_STOPPED,
@@ -30,6 +40,16 @@ export const EventChannels = {
VOICE_ACTIVE_USER: DISCORD_VOICE_ACTIVE_USER, // Active speaker state updates VOICE_ACTIVE_USER: DISCORD_VOICE_ACTIVE_USER, // Active speaker state updates
VOICE_PCM: DISCORD_VOICE_PCM, // Live PCM audio data stream VOICE_PCM: DISCORD_VOICE_PCM, // Live PCM audio data stream
ANALYSIS_QUEUE_STATUS: DISCORD_ANALYSIS_QUEUE_STATUS, ANALYSIS_QUEUE_STATUS: DISCORD_ANALYSIS_QUEUE_STATUS,
REACTION_ADDED: DISCORD_REACTION_ADDED,
REACTION_REMOVED: DISCORD_REACTION_REMOVED,
THREAD_CREATED: DISCORD_THREAD_CREATED,
THREAD_DELETED: DISCORD_THREAD_DELETED,
THREAD_UPDATED: DISCORD_THREAD_UPDATED,
CHANNEL_TOPIC_UPDATED: DISCORD_CHANNEL_TOPIC_UPDATED,
PRESENCE_UPDATED: DISCORD_PRESENCE_UPDATED,
GUILD_MEMBER_ADDED: DISCORD_GUILD_MEMBER_ADDED,
GUILD_MEMBER_REMOVED: DISCORD_GUILD_MEMBER_REMOVED,
VOICE_ANALYZED: DISCORD_VOICE_ANALYZED,
} as const; } as const;
export type EventChannelType = export type EventChannelType =
@@ -1,5 +1,6 @@
import http from "node:http"; import http from "node:http";
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import type { AppConfig as GatewayConfig } from "../../shared/config/config.js";
import { config } from "../../shared/config/config.js"; import { config } from "../../shared/config/config.js";
const logger = createChildLogger("gateway-metrics"); const logger = createChildLogger("gateway-metrics");
@@ -73,7 +74,9 @@ function formatMetrics(): string {
const lines: string[] = []; const lines: string[] = [];
for (const [fullName, metric] of metrics) { for (const [fullName, metric] of metrics) {
const baseName = fullName.includes("{") ? fullName.slice(0, fullName.indexOf("{")) : fullName; const baseName = fullName.includes("{")
? fullName.slice(0, fullName.indexOf("{"))
: fullName;
lines.push(`# HELP ${baseName} ${metric.help}`); lines.push(`# HELP ${baseName} ${metric.help}`);
lines.push(`# TYPE ${baseName} ${metric.type}`); lines.push(`# TYPE ${baseName} ${metric.type}`);
lines.push(`${fullName} ${metric.value}`); lines.push(`${fullName} ${metric.value}`);
@@ -85,7 +88,7 @@ function formatMetrics(): string {
export function startMetricsServer(): void { export function startMetricsServer(): void {
if (server) return; if (server) return;
const port = config.METRICS_PORT; const port = (config as any).METRICS_PORT ?? 9090;
logger.info({ port }, "Starting metrics HTTP server"); logger.info({ port }, "Starting metrics HTTP server");
server = http.createServer((req, res) => { server = http.createServer((req, res) => {
@@ -1,5 +1,9 @@
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import type { Client, GuildMember } from "discord.js-selfbot-v13"; import type {
Client,
GuildMember,
PartialGuildMember,
} from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js"; import { config } from "../../shared/config/config.js";
import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js"; import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js";
@@ -7,8 +11,11 @@ const logger = createChildLogger("guild-member-events");
function isMonitoredGuild(guildId: string | null | undefined): boolean { function isMonitoredGuild(guildId: string | null | undefined): boolean {
if (!guildId) return false; if (!guildId) return false;
const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as
if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; | string[]
| undefined;
if (!guildIds || guildIds.length === 0)
return config.MONITOR_GUILD_ID === guildId;
return guildIds.includes(guildId); return guildIds.includes(guildId);
} }
@@ -38,22 +45,25 @@ export function registerGuildMemberEvents(
await eventBroadcaster.guildMemberAdded(data).catch(() => {}); await eventBroadcaster.guildMemberAdded(data).catch(() => {});
}); });
client.on("guildMemberRemove", async (member: GuildMember) => { client.on(
if (!isMonitoredGuild(member.guild.id)) return; "guildMemberRemove",
async (member: GuildMember | PartialGuildMember) => {
if (!isMonitoredGuild(member.guild.id)) return;
const data = { const data = {
user_id: member.id, user_id: member.id,
username: member.user?.username ?? "unknown", username: (member.user as any)?.username ?? "unknown",
tag: member.user?.tag ?? null, tag: (member.user as any)?.tag ?? null,
guild_id: member.guild.id, guild_id: member.guild.id,
member_count: member.guild.memberCount, member_count: member.guild.memberCount,
removed_at: Date.now(), removed_at: Date.now(),
}; };
logger.info( logger.info(
{ userId: member.id, username: member.user?.username }, { userId: member.id, username: member.user?.username },
"Guild member removed", "Guild member removed",
); );
await eventBroadcaster.guildMemberRemoved(data).catch(() => {}); await eventBroadcaster.guildMemberRemoved(data).catch(() => {});
}); },
);
} }
@@ -85,7 +85,10 @@ function getTextCaptureTargets(): TextCaptureTarget[] {
const { EFFECTIVE_MONITOR_GUILD_IDS, TEXT_CHANNEL_ID } = config as any; const { EFFECTIVE_MONITOR_GUILD_IDS, TEXT_CHANNEL_ID } = config as any;
if (EFFECTIVE_MONITOR_GUILD_IDS?.length) { if (EFFECTIVE_MONITOR_GUILD_IDS?.length) {
if (TEXT_CHANNEL_ID) { if (TEXT_CHANNEL_ID) {
return EFFECTIVE_MONITOR_GUILD_IDS.map((guildId: string) => ({ guildId, channelId: TEXT_CHANNEL_ID })); return EFFECTIVE_MONITOR_GUILD_IDS.map((guildId: string) => ({
guildId,
channelId: TEXT_CHANNEL_ID,
}));
} }
return EFFECTIVE_MONITOR_GUILD_IDS.map((guildId: string) => ({ guildId })); return EFFECTIVE_MONITOR_GUILD_IDS.map((guildId: string) => ({ guildId }));
} }
@@ -99,7 +102,9 @@ function shouldCaptureForAnyTarget(
targets: TextCaptureTarget[], targets: TextCaptureTarget[],
): boolean { ): boolean {
if (targets.length === 0) return false; if (targets.length === 0) return false;
return targets.some((target) => shouldCaptureMessageLocation(message, target)); return targets.some((target) =>
shouldCaptureMessageLocation(message, target),
);
} }
function requireMessageGuildId(message: Message): string { function requireMessageGuildId(message: Message): string {
@@ -292,8 +297,7 @@ export function registerMessageCapture(client: Client): void {
}); });
client.on("messageUpdate", async (_oldMessage, newMessage) => { client.on("messageUpdate", async (_oldMessage, newMessage) => {
if (!shouldCaptureForAnyTarget(newMessage, targets)) if (!shouldCaptureForAnyTarget(newMessage, targets)) return;
return;
if (newMessage.author?.bot) return; if (newMessage.author?.bot) return;
if (isAgeRestrictedMessage(newMessage as Message)) return; if (isAgeRestrictedMessage(newMessage as Message)) return;
@@ -321,9 +325,14 @@ export function registerMessageCapture(client: Client): void {
// Save edit history snapshot before overwriting // Save edit history snapshot before overwriting
if (oldContent) { if (oldContent) {
insertMessageEdit(newMessage.id, oldContent, editedAt).catch((err: unknown) => { insertMessageEdit(newMessage.id, oldContent, editedAt).catch(
logger.error({ messageId: newMessage.id, error: err }, "Failed to save edit history"); (err: unknown) => {
}); logger.error(
{ messageId: newMessage.id, error: err },
"Failed to save edit history",
);
},
);
} }
await updateMessageAsEdited( await updateMessageAsEdited(
@@ -53,7 +53,11 @@ export class MessageStore {
// ── Edit History ──────────────────────────────────────────────────────── // ── Edit History ────────────────────────────────────────────────────────
insertMessageEdit(messageId: string, oldContent: string, editedAt: number): Promise<void> { insertMessageEdit(
messageId: string,
oldContent: string,
editedAt: number,
): Promise<void> {
return this.messages.insertMessageEdit(messageId, oldContent, editedAt); return this.messages.insertMessageEdit(messageId, oldContent, editedAt);
} }
@@ -313,7 +317,8 @@ export const insertMessageEdit = (
messageId: string, messageId: string,
oldContent: string, oldContent: string,
editedAt: number, editedAt: number,
): Promise<void> => getInstance().insertMessageEdit(messageId, oldContent, editedAt); ): Promise<void> =>
getInstance().insertMessageEdit(messageId, oldContent, editedAt);
// Messages // Messages
export const insertMessage = (message: MessageRecord): Promise<void> => export const insertMessage = (message: MessageRecord): Promise<void> =>
@@ -2,7 +2,10 @@ import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, or, type SQL } from "drizzle-orm"; import { and, desc, eq, or, type SQL } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres"; import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js"; import type * as schema from "../../shared/database/schema.js";
import { messagesTable, messageEditsTable } from "../../shared/database/schema.js"; import {
messageEditsTable,
messagesTable,
} from "../../shared/database/schema.js";
import type { MessageRecord } from "../message-capture/types.js"; import type { MessageRecord } from "../message-capture/types.js";
// ─── Shared Helpers ────────────────────────────────────────────────────────── // ─── Shared Helpers ──────────────────────────────────────────────────────────
@@ -152,11 +155,18 @@ export class MessagesCrud {
try { try {
await this.db await this.db
.insert(messageEditsTable) .insert(messageEditsTable)
.values({ message_id: messageId, old_content: oldContent, edited_at: editedAt }) .values({
message_id: messageId,
old_content: oldContent,
edited_at: editedAt,
})
.onConflictDoNothing(); .onConflictDoNothing();
} catch (error) { } catch (error) {
this.logger.error( this.logger.error(
{ messageId, error: error instanceof Error ? error.message : String(error) }, {
messageId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to insert message edit", "Failed to insert message edit",
); );
throw error; throw error;
@@ -128,7 +128,11 @@ export class MessagesDb {
// ── Edit History ───────────────────────────────────────────────────────── // ── Edit History ─────────────────────────────────────────────────────────
insertMessageEdit(messageId: string, oldContent: string, editedAt: number): Promise<void> { insertMessageEdit(
messageId: string,
oldContent: string,
editedAt: number,
): Promise<void> {
return this.crud.insertMessageEdit(messageId, oldContent, editedAt); return this.crud.insertMessageEdit(messageId, oldContent, editedAt);
} }
@@ -1,8 +1,14 @@
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import type { Client, MessageReaction, User } from "discord.js-selfbot-v13"; import type {
Client,
MessageReaction,
PartialMessageReaction,
PartialUser,
User,
} from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js";
import { getDatabase } from "../../shared/database/drizzle.js"; import { getDatabase } from "../../shared/database/drizzle.js";
import { reactionsTable } from "../../shared/database/schema.js"; import { reactionsTable } from "../../shared/database/schema.js";
import { config } from "../../shared/config/config.js";
import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js"; import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js";
const logger = createChildLogger("reaction-tracking"); const logger = createChildLogger("reaction-tracking");
@@ -11,12 +17,17 @@ const logger = createChildLogger("reaction-tracking");
function isMonitoredGuild(guildId: string | null | undefined): boolean { function isMonitoredGuild(guildId: string | null | undefined): boolean {
if (!guildId) return false; if (!guildId) return false;
const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as
if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; | string[]
| undefined;
if (!guildIds || guildIds.length === 0)
return config.MONITOR_GUILD_ID === guildId;
return guildIds.includes(guildId); return guildIds.includes(guildId);
} }
function getEmojiIdentifier(reaction: MessageReaction): { function getEmojiIdentifier(
reaction: MessageReaction | PartialMessageReaction,
): {
emoji: string; emoji: string;
emojiId: string | null; emojiId: string | null;
animated: boolean; animated: boolean;
@@ -39,8 +50,8 @@ function getEmojiIdentifier(reaction: MessageReaction): {
// ─── Event Handlers ────────────────────────────────────────────────────── // ─── Event Handlers ──────────────────────────────────────────────────────
async function handleReactionAdd( async function handleReactionAdd(
reaction: MessageReaction, reaction: MessageReaction | PartialMessageReaction,
user: User, user: User | PartialUser,
): Promise<void> { ): Promise<void> {
const guildId = reaction.message.guildId; const guildId = reaction.message.guildId;
if (!isMonitoredGuild(guildId)) return; if (!isMonitoredGuild(guildId)) return;
@@ -52,19 +63,22 @@ async function handleReactionAdd(
try { try {
const db = getDatabase(); const db = getDatabase();
await (db as any).insert(reactionsTable).values({ await (db as any)
id, .insert(reactionsTable)
message_id: reaction.message.id, .values({
channel_id: reaction.message.channelId, id,
guild_id: guildId, message_id: reaction.message.id,
user_id: user.id, channel_id: reaction.message.channelId,
username: user.username, guild_id: guildId,
emoji, user_id: user.id,
emoji_id: emojiId, username: user.username,
animated, emoji,
reaction_type: "add", emoji_id: emojiId,
created_at: now, animated,
}).onConflictDoNothing(); reaction_type: "add",
created_at: now,
})
.onConflictDoNothing();
logger.debug( logger.debug(
{ messageId: reaction.message.id, emoji, userId: user.id }, { messageId: reaction.message.id, emoji, userId: user.id },
@@ -79,8 +93,8 @@ async function handleReactionAdd(
} }
async function handleReactionRemove( async function handleReactionRemove(
reaction: MessageReaction, reaction: MessageReaction | PartialMessageReaction,
user: User, user: User | PartialUser,
): Promise<void> { ): Promise<void> {
const guildId = reaction.message.guildId; const guildId = reaction.message.guildId;
if (!isMonitoredGuild(guildId)) return; if (!isMonitoredGuild(guildId)) return;
@@ -92,19 +106,22 @@ async function handleReactionRemove(
try { try {
const db = getDatabase(); const db = getDatabase();
await (db as any).insert(reactionsTable).values({ await (db as any)
id, .insert(reactionsTable)
message_id: reaction.message.id, .values({
channel_id: reaction.message.channelId, id,
guild_id: guildId, message_id: reaction.message.id,
user_id: user.id, channel_id: reaction.message.channelId,
username: user.username, guild_id: guildId,
emoji, user_id: user.id,
emoji_id: emojiId, username: user.username,
animated, emoji,
reaction_type: "remove", emoji_id: emojiId,
created_at: now, animated,
}).onConflictDoNothing(); reaction_type: "remove",
created_at: now,
})
.onConflictDoNothing();
logger.debug( logger.debug(
{ messageId: reaction.message.id, emoji, userId: user.id }, { messageId: reaction.message.id, emoji, userId: user.id },
@@ -126,45 +143,61 @@ export function registerReactionCapture(
): void { ): void {
logger.info("Registering reaction capture"); logger.info("Registering reaction capture");
client.on("messageReactionAdd", async (reaction, user) => { client.on(
await handleReactionAdd(reaction, user); "messageReactionAdd",
async (
reaction: MessageReaction | PartialMessageReaction,
user: User | PartialUser,
) => {
await handleReactionAdd(reaction, user);
const guildId = reaction.message.guildId; const guildId = reaction.message.guildId;
if (!isMonitoredGuild(guildId)) return; if (!isMonitoredGuild(guildId)) return;
const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); const { emoji, emojiId, animated } = getEmojiIdentifier(reaction);
eventBroadcaster.reactionAdded({ eventBroadcaster
message_id: reaction.message.id, .reactionAdded({
channel_id: reaction.message.channelId, message_id: reaction.message.id,
guild_id: guildId, channel_id: reaction.message.channelId,
user_id: user.id, guild_id: guildId,
username: user.username, user_id: user.id,
emoji, username: user.username,
emoji_id: emojiId, emoji,
animated, emoji_id: emojiId,
created_at: Date.now(), animated,
}).catch(() => {}); created_at: Date.now(),
}); })
.catch(() => {});
},
);
client.on("messageReactionRemove", async (reaction, user) => { client.on(
await handleReactionRemove(reaction, user); "messageReactionRemove",
async (
reaction: MessageReaction | PartialMessageReaction,
user: User | PartialUser,
) => {
await handleReactionRemove(reaction, user);
const guildId = reaction.message.guildId; const guildId = reaction.message.guildId;
if (!isMonitoredGuild(guildId)) return; if (!isMonitoredGuild(guildId)) return;
const { emoji, emojiId, animated } = getEmojiIdentifier(reaction); const { emoji, emojiId, animated } = getEmojiIdentifier(reaction);
eventBroadcaster.reactionRemoved({ eventBroadcaster
message_id: reaction.message.id, .reactionRemoved({
channel_id: reaction.message.channelId, message_id: reaction.message.id,
guild_id: guildId, channel_id: reaction.message.channelId,
user_id: user.id, guild_id: guildId,
username: user.username, user_id: user.id,
emoji, username: user.username,
emoji_id: emojiId, emoji,
animated, emoji_id: emojiId,
created_at: Date.now(), animated,
}).catch(() => {}); created_at: Date.now(),
}); })
.catch(() => {});
},
);
} }
@@ -7,8 +7,11 @@ const logger = createChildLogger("thread-tracking");
function isMonitoredGuild(guildId: string | null | undefined): boolean { function isMonitoredGuild(guildId: string | null | undefined): boolean {
if (!guildId) return false; if (!guildId) return false;
const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as
if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; | string[]
| undefined;
if (!guildIds || guildIds.length === 0)
return config.MONITOR_GUILD_ID === guildId;
return guildIds.includes(guildId); return guildIds.includes(guildId);
} }
@@ -51,20 +54,23 @@ export function registerThreadCapture(
await eventBroadcaster.threadDeleted(data).catch(() => {}); await eventBroadcaster.threadDeleted(data).catch(() => {});
}); });
client.on("threadUpdate", async (_oldThread: ThreadChannel, newThread: ThreadChannel) => { client.on(
if (!isMonitoredGuild(newThread.guildId)) return; "threadUpdate",
async (_oldThread: ThreadChannel, newThread: ThreadChannel) => {
if (!isMonitoredGuild(newThread.guildId)) return;
const data = { const data = {
id: newThread.id, id: newThread.id,
guild_id: newThread.guildId, guild_id: newThread.guildId,
channel_id: newThread.parentId ?? newThread.guildId, channel_id: newThread.parentId ?? newThread.guildId,
name: newThread.name, name: newThread.name,
archived: (newThread as any).archived ?? false, archived: (newThread as any).archived ?? false,
rate_limit_per_user: (newThread as any).rateLimitPerUser ?? null, rate_limit_per_user: (newThread as any).rateLimitPerUser ?? null,
updated_at: Date.now(), updated_at: Date.now(),
}; };
logger.debug({ threadId: newThread.id }, "Thread updated"); logger.debug({ threadId: newThread.id }, "Thread updated");
await eventBroadcaster.threadUpdated(data).catch(() => {}); await eventBroadcaster.threadUpdated(data).catch(() => {});
}); },
);
} }
@@ -11,19 +11,25 @@ const PRESENCE_COOLDOWN_MS = 30_000;
function isMonitoredGuild(guildId: string | null | undefined): boolean { function isMonitoredGuild(guildId: string | null | undefined): boolean {
if (!guildId) return false; if (!guildId) return false;
const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as string[] | undefined; const guildIds = (config as any).EFFECTIVE_MONITOR_GUILD_IDS as
if (!guildIds || guildIds.length === 0) return config.MONITOR_GUILD_ID === guildId; | string[]
| undefined;
if (!guildIds || guildIds.length === 0)
return config.MONITOR_GUILD_ID === guildId;
return guildIds.includes(guildId); return guildIds.includes(guildId);
} }
function getStatus(presence: Presence): string { function getStatus(presence: Presence): string {
if (!presence) return "offline"; if (!presence) return "offline";
const status = presence.status; const status = presence.status;
if (status === "online" || status === "idle" || status === "dnd") return status; if (status === "online" || status === "idle" || status === "dnd")
return status;
return "offline"; return "offline";
} }
function getActivities(presence: Presence): Array<{ name: string; type: string }> { function getActivities(
presence: Presence,
): Array<{ name: string; type: string }> {
if (!presence?.activities) return []; if (!presence?.activities) return [];
return presence.activities.map((a) => ({ return presence.activities.map((a) => ({
name: a.name ?? "unknown", name: a.name ?? "unknown",
@@ -47,38 +53,42 @@ export function registerPresenceCapture(
): void { ): void {
logger.info("Registering presence capture"); logger.info("Registering presence capture");
client.on("presenceUpdate", async (_oldPresence: Presence | null, newPresence: Presence) => { client.on(
if (!newPresence?.guildId) return; "presenceUpdate",
if (!isMonitoredGuild(newPresence.guildId)) return; async (_oldPresence: Presence | null, newPresence: Presence) => {
const guildId = newPresence.guild?.id ?? null;
if (!guildId) return;
if (!isMonitoredGuild(guildId)) return;
const userId = newPresence.userId ?? newPresence.user?.id; const userId = newPresence.userId ?? newPresence.user?.id;
if (!userId) return; if (!userId) return;
// Cooldown check // Cooldown check
const now = Date.now(); const now = Date.now();
const lastUpdate = presenceCooldowns.get(userId); const lastUpdate = presenceCooldowns.get(userId);
if (lastUpdate && now - lastUpdate < PRESENCE_COOLDOWN_MS) return; if (lastUpdate && now - lastUpdate < PRESENCE_COOLDOWN_MS) return;
presenceCooldowns.set(userId, now); presenceCooldowns.set(userId, now);
const data = { const data = {
user_id: userId, user_id: userId,
username: newPresence.user?.username ?? "unknown", username: newPresence.user?.username ?? "unknown",
status: getStatus(newPresence), status: getStatus(newPresence),
activities: getActivities(newPresence), activities: getActivities(newPresence),
client_status: getClientStatus(newPresence), client_status: getClientStatus(newPresence),
guild_id: newPresence.guildId, guild_id: guildId,
last_changed: now, last_changed: now,
}; };
logger.debug({ userId, status: data.status }, "Presence updated"); logger.debug({ userId, status: data.status }, "Presence updated");
await eventBroadcaster.presenceUpdated(data).catch(() => {}); await eventBroadcaster.presenceUpdated(data).catch(() => {});
// Periodic cleanup of stale cooldown entries // Periodic cleanup of stale cooldown entries
if (presenceCooldowns.size > 1000) { if (presenceCooldowns.size > 1000) {
const threshold = now - PRESENCE_COOLDOWN_MS * 10; const threshold = now - PRESENCE_COOLDOWN_MS * 10;
for (const [uid, ts] of presenceCooldowns) { for (const [uid, ts] of presenceCooldowns) {
if (ts < threshold) presenceCooldowns.delete(uid); if (ts < threshold) presenceCooldowns.delete(uid);
}
} }
} },
}); );
} }
@@ -213,8 +213,9 @@ export function resolveMediaUrl(
} }
}); });
// -- stderr (capture for diagnostics) ---------------------------------- // -- stderr (capture for diagnostics, capped at 4KB) ----------------------------------
const MAX_STDERR = 4096;
if (proc.stderr) { if (proc.stderr) {
proc.stderr.on("data", (chunk: Buffer) => { proc.stderr.on("data", (chunk: Buffer) => {
stderrBuf += chunk.toString("utf8"); stderrBuf += chunk.toString("utf8");
@@ -295,6 +296,7 @@ export async function extractMediaInfo(url: string): Promise<MediaInfo> {
let stdoutBuf = ""; let stdoutBuf = "";
let stderrBuf = ""; let stderrBuf = "";
const MAX_STDERR = 4096;
if (proc.stdout) { if (proc.stdout) {
proc.stdout.on("data", (chunk: Buffer) => { proc.stdout.on("data", (chunk: Buffer) => {
@@ -304,7 +306,9 @@ export async function extractMediaInfo(url: string): Promise<MediaInfo> {
if (proc.stderr) { if (proc.stderr) {
proc.stderr.on("data", (chunk: Buffer) => { proc.stderr.on("data", (chunk: Buffer) => {
stderrBuf += chunk.toString("utf8"); if (stderrBuf.length < MAX_STDERR) {
stderrBuf += chunk.toString("utf8").slice(0, MAX_STDERR - stderrBuf.length);
}
}); });
} }
@@ -1,9 +1,10 @@
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import { eq } from "drizzle-orm"; import { eq } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres"; 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 { config } from "../../shared/config/config.js";
import { getDatabase } from "../../shared/database/drizzle.js";
import type * as schema from "../../shared/database/schema.js";
import { muxerJobsTable } from "../../shared/database/schema.js";
import { buildMuxFfmpegArgs, runFfmpeg } from "./ffmpegProcess.js"; import { buildMuxFfmpegArgs, runFfmpeg } from "./ffmpegProcess.js";
const logger = createChildLogger("muxer"); const logger = createChildLogger("muxer");
@@ -59,10 +60,7 @@ export function startMuxerWorker(): void {
pollTimer = setInterval(() => { pollTimer = setInterval(() => {
processNextJobs().catch((err: unknown) => { processNextJobs().catch((err: unknown) => {
logger.error( logger.error({ error: String(err) }, "Muxer worker tick failed");
{ error: String(err) },
"Muxer worker tick failed",
);
}); });
}, 10_000); }, 10_000);
} }
@@ -126,7 +124,9 @@ async function processJob(
const data = JSON.parse(job.data) as MuxerJobData; const data = JSON.parse(job.data) as MuxerJobData;
if (!data.inputs || data.inputs.length < 2) { if (!data.inputs || data.inputs.length < 2) {
throw new Error(`Muxer job ${job.id} needs at least 2 inputs, got ${data.inputs?.length ?? 0}`); throw new Error(
`Muxer job ${job.id} needs at least 2 inputs, got ${data.inputs?.length ?? 0}`,
);
} }
logger.info( logger.info(
@@ -156,10 +156,7 @@ async function processJob(
.set({ status: "completed", updatedAt: Date.now() }) .set({ status: "completed", updatedAt: Date.now() })
.where(eq(muxerJobsTable.id, job.id)); .where(eq(muxerJobsTable.id, job.id));
logger.info( logger.info({ jobId: job.id, output: data.output }, "Muxer job completed");
{ jobId: job.id, output: data.output },
"Muxer job completed",
);
} catch (error) { } catch (error) {
const errMsg = error instanceof Error ? error.message : String(error); const errMsg = error instanceof Error ? error.message : String(error);
const newAttempts = job.attempts + 1; const newAttempts = job.attempts + 1;
@@ -174,7 +171,10 @@ async function processJob(
updatedAt: Date.now(), updatedAt: Date.now(),
}) })
.where(eq(muxerJobsTable.id, job.id)); .where(eq(muxerJobsTable.id, job.id));
logger.error({ jobId: job.id, error: errMsg }, "Muxer job failed permanently"); logger.error(
{ jobId: job.id, error: errMsg },
"Muxer job failed permanently",
);
} else { } else {
await db await db
.update(muxerJobsTable) .update(muxerJobsTable)
@@ -1,2 +1,2 @@
export { triggerWebhook } from "./webhookNotifier.js";
export type { WebhookPayload } from "./webhookNotifier.js"; export type { WebhookPayload } from "./webhookNotifier.js";
export { triggerWebhook } from "./webhookNotifier.js";
@@ -1,4 +1,5 @@
import { createChildLogger } from "@bete/shared/logger"; import { createChildLogger } from "@bete/shared/logger";
import type { AppConfig as GatewayConfig } from "../../shared/config/config.js";
import { config } from "../../shared/config/config.js"; import { config } from "../../shared/config/config.js";
const logger = createChildLogger("webhook-notifier"); const logger = createChildLogger("webhook-notifier");
@@ -29,11 +30,16 @@ export async function triggerWebhook(
eventType: string, eventType: string,
payload: WebhookPayload, payload: WebhookPayload,
): Promise<void> { ): Promise<void> {
const urls = config.WEBHOOK_URLS; const urls = (config as any).WEBHOOK_URLS as string[] | undefined;
if (!urls || urls.length === 0) return; if (!urls || urls.length === 0) return;
const enabledEvents = config.WEBHOOK_EVENTS; const enabledEvents = (config as any).WEBHOOK_EVENTS as string[] | undefined;
if (enabledEvents.length > 0 && !enabledEvents.includes(eventType)) return; if (
enabledEvents &&
enabledEvents.length > 0 &&
!enabledEvents.includes(eventType)
)
return;
const body = JSON.stringify({ const body = JSON.stringify({
...payload, ...payload,
@@ -59,7 +65,11 @@ export async function triggerWebhook(
// ─── Internal ──────────────────────────────────────────────────────────── // ─── Internal ────────────────────────────────────────────────────────────
async function sendWebhook(url: string, body: string): Promise<void> { async function sendWebhook(
url: string | undefined,
body: string,
): Promise<void> {
if (!url) return;
let lastErr: Error | null = null; let lastErr: Error | null = null;
for (let attempt = 0; attempt < 3; attempt++) { for (let attempt = 0; attempt < 3; attempt++) {
try { try {
@@ -6,6 +6,7 @@ import { loadConfig as sharedLoadConfig } from "@bete/shared/config";
export type AppConfig = SharedAppConfig & { export type AppConfig = SharedAppConfig & {
EFFECTIVE_TEXT_GUILD_ID?: string; EFFECTIVE_TEXT_GUILD_ID?: string;
EFFECTIVE_VOICE_GUILD_ID?: string; EFFECTIVE_VOICE_GUILD_ID?: string;
EFFECTIVE_MONITOR_GUILD_IDS: string[];
}; };
export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig {
@@ -14,6 +15,9 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig {
...parsed, ...parsed,
EFFECTIVE_TEXT_GUILD_ID: parsed.TEXT_GUILD_ID ?? parsed.MONITOR_GUILD_ID, EFFECTIVE_TEXT_GUILD_ID: parsed.TEXT_GUILD_ID ?? parsed.MONITOR_GUILD_ID,
EFFECTIVE_VOICE_GUILD_ID: parsed.VOICE_GUILD_ID, EFFECTIVE_VOICE_GUILD_ID: parsed.VOICE_GUILD_ID,
EFFECTIVE_MONITOR_GUILD_IDS:
(parsed as any).EFFECTIVE_MONITOR_GUILD_IDS ??
(parsed.MONITOR_GUILD_ID ? [parsed.MONITOR_GUILD_ID] : []),
}; };
} }
@@ -390,6 +390,47 @@ export const pgUserProfilesTable = pgTable(
* Mascot Chat Messages Table (PostgreSQL) * Mascot Chat Messages Table (PostgreSQL)
* Stores AI mascot chat conversation history * Stores AI mascot chat conversation history
*/ */
export const pgReactionsTable = pgTable(
"message_reactions",
{
id: pgText("id").primaryKey(),
message_id: pgText("message_id").notNull(),
channel_id: pgText("channel_id").notNull(),
guild_id: pgText("guild_id").notNull(),
user_id: pgText("user_id").notNull(),
username: pgText("username").notNull(),
emoji: pgText("emoji").notNull(),
emoji_id: pgText("emoji_id"),
animated: pgBoolean("animated").notNull().default(false),
reaction_type: pgText("reaction_type", {
enum: ["add", "remove"],
}).notNull(),
created_at: pgBigint("created_at", { mode: "number" }).notNull(),
},
(table) => ({
messageIdIdx: pgIndex("idx_reactions_message_id").on(table.message_id),
userIdIdx: pgIndex("idx_reactions_user_id").on(table.user_id),
guildCreatedIdx: pgIndex("idx_reactions_guild_created").on(
table.guild_id,
table.created_at,
),
}),
);
export const pgMessageEditsTable = pgTable(
"message_edits",
{
id: pgUuid("id").defaultRandom().primaryKey(),
message_id: pgText("message_id").notNull(),
old_content: pgText("old_content").notNull(),
edited_at: pgBigint("edited_at", { mode: "number" }).notNull(),
},
(table) => ({
messageIdIdx: pgIndex("idx_message_edits_message_id").on(table.message_id),
editedAtIdx: pgIndex("idx_message_edits_edited_at").on(table.edited_at),
}),
);
export const pgMascotChatMessagesTable = pgTable( export const pgMascotChatMessagesTable = pgTable(
"mascot_chat_messages", "mascot_chat_messages",
{ {
@@ -428,6 +469,8 @@ export const correctedModerationsTable = pgCorrectedModerationsTable;
export const userReputationsTable = pgUserReputationsTable; export const userReputationsTable = pgUserReputationsTable;
export const channelCulturesTable = pgChannelCulturesTable; export const channelCulturesTable = pgChannelCulturesTable;
export const userProfilesTable = pgUserProfilesTable; export const userProfilesTable = pgUserProfilesTable;
export const reactionsTable = pgReactionsTable;
export const messageEditsTable = pgMessageEditsTable;
export const mascotChatMessagesTable = pgMascotChatMessagesTable; export const mascotChatMessagesTable = pgMascotChatMessagesTable;
// Export table types for use in queries // Export table types for use in queries