refactor: atomic, DRY, and logging improvements

- Shared Redis channel constants as single source of truth (redis-channels.ts)
- commandHandler.ts split into VoiceHandler, MediaHandler, GuildHandler,
  ModerationHandler with handler-registry.ts dispatch
- messageStore.ts (1322 lines) split into domain-specific DB files:
  messages.db.ts, attachments.db.ts, reviews.db.ts,
  moderation-actions.db.ts, retention.db.ts
- recorder.ts startSpeaking callback extracted into speakingHandler.ts,
  streamSetup.ts, segmentFinalizer.ts
- autoDeleteManager.ts split into autoDeleteEligibility.ts,
  autoDeleteNotify.ts, autoDeleteLogger.ts
- Added createChildLogger() logging across 8 service files
- Backend messages.repository.ts migrated from raw SQL to Drizzle ORM
- Fixed biome.json to exclude packages/**/dist/* from lint
- Fixed config.ts GUILD_ID pre-existing type error

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MythEclipse
2026-06-09 17:34:18 +07:00
co-authored by Claude Opus 4.8
parent 7108f6bb47
commit b68789fffc
37 changed files with 3620 additions and 2345 deletions
+2 -1
View File
@@ -10,7 +10,8 @@
"*.ts",
"!vendor/**",
"!.claude/**",
"!services/**/dist/**"
"!services/**/dist/**",
"!packages/**/dist/**"
]
},
"formatter": {
+1
View File
@@ -2,4 +2,5 @@ export * from "./config/index.js";
export * from "./errors/index.js";
export * from "./logger/index.js";
export * from "./moderation-types.js";
export * from "./redis-channels.js";
export * from "./utils/index.js";
+84
View File
@@ -0,0 +1,84 @@
// ---------------------------------------------------------------------------
// Redis Channel Constants — single source of truth
//
// All Redis channel names, status keys, and command types used for
// inter-service communication between discord-gateway and backend.
// ---------------------------------------------------------------------------
// ---------------------------------------------------------------------------
// Event channels (discord-gateway -> backend via pub/sub)
// ---------------------------------------------------------------------------
export const DISCORD_MESSAGE_CREATED = "discord:message:created";
export const DISCORD_MESSAGE_UPDATED = "discord:message:updated";
export const DISCORD_MESSAGE_DELETED = "discord:message:deleted";
export const DISCORD_MESSAGE_ANALYZED = "discord:message:analyzed";
export const DISCORD_ATTACHMENT_CREATED = "discord:attachment:created";
export const DISCORD_ATTACHMENT_UPLOADED = "discord:attachment:uploaded";
export const DISCORD_VOICE_STARTED = "discord:voice:started";
export const DISCORD_VOICE_STOPPED = "discord:voice:stopped";
export const DISCORD_VOICE_UPLOADED = "discord:voice:uploaded";
export const DISCORD_VOICE_ACTIVE_USER = "discord:voice:active_user";
export const DISCORD_VOICE_PCM = "discord:voice:pcm";
export const DISCORD_ANALYSIS_QUEUE_STATUS = "discord:analysis:queue_status";
// ---------------------------------------------------------------------------
// Command channels (backend -> discord-gateway)
// ---------------------------------------------------------------------------
export const BACKEND_COMMAND = "backend:command";
export const BACKEND_VOICE_TRANSMIT = "backend:voice:transmit";
export const BACKEND_COMMAND_REPLY_PREFIX = "backend:command:reply:";
// ---------------------------------------------------------------------------
// Status keys (set by discord-gateway, read by backend via Redis GET)
// ---------------------------------------------------------------------------
export const VOICE_STATUS_KEY = "voice:status";
export const MEDIA_STATUS_KEY = "media:status";
// ---------------------------------------------------------------------------
// Command types (used as the `type` field in CommandMessage envelopes)
// ---------------------------------------------------------------------------
export const COMMAND_VOICE_CONNECT = "voice:connect";
export const COMMAND_VOICE_DISCONNECT = "voice:disconnect";
export const COMMAND_VOICE_CHANNELS = "voice:channels";
export const COMMAND_VOICE_TRANSMIT_START = "voice:transmit:start";
export const COMMAND_VOICE_TRANSMIT_STOP = "voice:transmit:stop";
export const COMMAND_GUILDS_LIST = "guilds:list";
export const COMMAND_GUILDS_TEXT_CHANNELS = "guilds:text-channels";
export const COMMAND_MEDIA_QUEUE = "media:queue";
export const COMMAND_MEDIA_SKIP = "media:skip";
export const COMMAND_MEDIA_STOP = "media:stop";
export const COMMAND_MEDIA_VOLUME = "media:volume";
export const COMMAND_MODERATION_ACTION = "moderation:action";
// ---------------------------------------------------------------------------
// Event envelope — used by discord-gateway when publishing to Redis
// ---------------------------------------------------------------------------
export interface DiscordGatewayEvent {
type: string;
data: unknown;
timestamp: number;
source: string;
}
// ---------------------------------------------------------------------------
// Command envelope — used by backend when publishing to backend:command
// ---------------------------------------------------------------------------
export interface CommandMessage {
id: string;
type: string;
payload: Record<string, unknown>;
replyChannel: string;
}
export interface CommandReply<T = unknown> {
id: string;
success: boolean;
data?: T;
error?: string;
}
@@ -15,6 +15,10 @@ class MascotChatService {
context: MascotChatContext | undefined,
userId: string,
): Promise<string> {
logger.info(
{ userId, messageLength: message.length },
"processMessage called",
);
const recentContext = await this.getRecentConversationContext(userId);
const serverInsights = await mascotChatRepository.getServerInsights(
context?.guildId,
@@ -34,6 +38,7 @@ class MascotChatService {
}
async saveConversation(input: SaveConversationInput): Promise<void> {
logger.info({ userId: input.userId }, "saveConversation called");
await mascotChatRepository.saveConversation(input);
}
@@ -41,10 +46,12 @@ class MascotChatService {
userId: string,
limit: number,
): Promise<MascotChatHistoryRow[]> {
logger.debug({ userId, limit }, "getChatHistory called");
return mascotChatRepository.getChatHistory(userId, limit);
}
async clearChatHistory(userId: string): Promise<void> {
logger.info({ userId }, "clearChatHistory called");
await mascotChatRepository.clearChatHistory(userId);
}
@@ -1,3 +1,10 @@
import {
COMMAND_MEDIA_QUEUE,
COMMAND_MEDIA_SKIP,
COMMAND_MEDIA_STOP,
COMMAND_MEDIA_VOLUME,
MEDIA_STATUS_KEY,
} from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { publishCommand, readRedisStatus } from "../../shared/redis/index.js";
@@ -41,10 +48,11 @@ const DEFAULT_STATE: MediaState = {
// ---------------------------------------------------------------------------
/**
* Read media status from Redis key "media:status" set by discord-gateway.
* Read media status from Redis key `media:status` set by discord-gateway.
*/
export async function getStatus(): Promise<MediaState> {
const cached = await readRedisStatus("media:status");
logger.debug("getStatus called");
const cached = await readRedisStatus(MEDIA_STATUS_KEY);
if (cached) {
const rawPlaying = cached.playing;
@@ -71,8 +79,9 @@ export async function queue(
source: string,
mode: "music" | "screen" = "music",
): Promise<MediaState> {
logger.info({ source, mode }, "queue called");
const reply = await publishCommand<MediaState>(
"media:queue",
COMMAND_MEDIA_QUEUE,
{ source, mode },
DEFAULT_COMMAND_TIMEOUT_MS,
);
@@ -97,8 +106,9 @@ export async function queue(
* Skip current track via Redis command to discord-gateway.
*/
export async function skip(): Promise<MediaState> {
logger.info("skip called");
const reply = await publishCommand<MediaState>(
"media:skip",
COMMAND_MEDIA_SKIP,
{},
DEFAULT_COMMAND_TIMEOUT_MS,
);
@@ -120,8 +130,9 @@ export async function skip(): Promise<MediaState> {
* Stop playback via Redis command to discord-gateway.
*/
export async function stop(): Promise<MediaState> {
logger.info("stop called");
const reply = await publishCommand<MediaState>(
"media:stop",
COMMAND_MEDIA_STOP,
{},
DEFAULT_COMMAND_TIMEOUT_MS,
);
@@ -143,8 +154,9 @@ export async function stop(): Promise<MediaState> {
* Set volume via Redis command to discord-gateway.
*/
export async function setVolume(volume: number): Promise<MediaState> {
logger.info({ volume }, "setVolume called");
const reply = await publishCommand<MediaState>(
"media:volume",
COMMAND_MEDIA_VOLUME,
{ volume },
DEFAULT_COMMAND_TIMEOUT_MS,
);
@@ -1,6 +1,14 @@
import type { PageResult } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { getPool } from "../../shared/database/index.js";
import { and, desc, eq, inArray, lt, ne, type SQL } from "drizzle-orm";
import {
bigint as pgBigint,
integer as pgInteger,
real as pgReal,
pgTable,
text as pgText,
} from "drizzle-orm/pg-core";
import { getDatabase } from "../../shared/database/index.js";
import type {
MessageCreate,
MessageQuery,
@@ -9,6 +17,71 @@ import type {
const logger = createChildLogger("messages.repository");
/**
* Local table definitions mirroring services/discord-gateway/src/shared/database/schema.ts.
* These are query-building references only — schema source of truth remains in discord-gateway.
*/
const messages = pgTable("messages", {
id: pgText("id").primaryKey(),
guild_id: pgText("guild_id").notNull(),
channel_id: pgText("channel_id").notNull(),
thread_id: pgText("thread_id"),
user_id: pgText("user_id").notNull(),
username: pgText("username").notNull(),
avatar_url: pgText("avatar_url"),
content: pgText("content").notNull(),
edited_content: pgText("edited_content"),
created_at: pgBigint("created_at", { mode: "number" }).notNull(),
edited_at: pgBigint("edited_at", { mode: "number" }),
deleted_at: pgBigint("deleted_at", { mode: "number" }),
type: pgText("type", {
enum: ["text", "edited", "deleted"],
})
.notNull()
.default("text"),
metadata: pgText("metadata"),
ai_status: pgText("ai_status", {
enum: ["pending", "processing", "clean", "warn", "flagged", "error"],
})
.notNull()
.default("pending"),
ai_moderation_flags: pgText("ai_moderation_flags"),
ai_moderation_score: pgReal("ai_moderation_score"),
ai_analysis: pgText("ai_analysis"),
ai_categories: pgText("ai_categories"),
ai_severity: pgText("ai_severity", {
enum: ["none", "low", "medium", "high", "critical"],
}),
ai_confidence: pgReal("ai_confidence"),
ai_recommended_action: pgText("ai_recommended_action", {
enum: ["none", "monitor", "warn", "review", "delete", "escalate"],
}),
ai_analyzed_at: pgBigint("ai_analyzed_at", { mode: "number" }),
ai_error: pgText("ai_error"),
});
const attachments = pgTable("attachments", {
id: pgText("id").primaryKey(),
message_id: pgText("message_id").notNull(),
guild_id: pgText("guild_id").notNull(),
channel_id: pgText("channel_id").notNull(),
thread_id: pgText("thread_id"),
user_id: pgText("user_id").notNull(),
filename: pgText("filename").notNull(),
size: pgInteger("size").notNull(),
type: pgText("type").notNull(),
discord_url: pgText("discord_url").notNull(),
uploaded_url: pgText("uploaded_url"),
upload_status: pgText("upload_status", {
enum: ["pending", "uploaded", "failed"],
})
.notNull()
.default("pending"),
upload_error: pgText("upload_error"),
created_at: pgBigint("created_at", { mode: "number" }).notNull(),
uploaded_at: pgBigint("uploaded_at", { mode: "number" }),
});
export interface AttachmentResult {
id: string;
message_id: string;
@@ -60,45 +133,37 @@ export class MessagesRepository {
async findMany(
query: MessageQuery,
): Promise<PageResult<ReturnType<typeof mapMessageRow>>> {
const pool = getPool();
const db = getDatabase();
const limit = query.limit ?? 50;
const clauses: string[] = [];
const params: (string | number)[] = [];
let p = 1;
const conditions: SQL[] = [];
if (query.guildId) {
clauses.push(`guild_id = $${p}`);
params.push(query.guildId);
p++;
conditions.push(eq(messages.guild_id, query.guildId));
}
if (query.channelId) {
clauses.push(`channel_id = $${p}`);
params.push(query.channelId);
p++;
conditions.push(eq(messages.channel_id, query.channelId));
}
if (query.userId) {
clauses.push(`user_id = $${p}`);
params.push(query.userId);
p++;
conditions.push(eq(messages.user_id, query.userId));
}
if (query.status) {
clauses.push(`ai_status = $${p}`);
params.push(query.status);
p++;
conditions.push(eq(messages.ai_status, query.status));
}
if (query.cursor) {
clauses.push(`created_at < $${p}`);
params.push(Number(query.cursor));
p++;
conditions.push(lt(messages.created_at, Number(query.cursor)));
}
const where = clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : "";
const { rows } = await pool.query(
`SELECT * FROM messages ${where} ORDER BY created_at DESC LIMIT $${p}`,
[...params, limit + 1],
);
const where = conditions.length > 0 ? and(...conditions) : undefined;
const rows = await db
.select()
.from(messages)
.where(where)
.orderBy(desc(messages.created_at))
.limit(limit + 1);
const data = rows.slice(0, limit).map(mapMessageRow);
const data = rows
.slice(0, limit)
.map((r) => mapMessageRow(r as Record<string, unknown>));
const nextCursor =
rows.length > limit ? String(rows[limit].created_at) : null;
@@ -107,37 +172,39 @@ export class MessagesRepository {
}
async findById(id: string) {
const pool = getPool();
const { rows } = await pool.query(`SELECT * FROM messages WHERE id = $1`, [
id,
]);
const db = getDatabase();
const [row] = await db
.select()
.from(messages)
.where(eq(messages.id, id))
.limit(1);
if (rows.length === 0) return null;
return mapMessageRow(rows[0] as Record<string, unknown>);
if (!row) return null;
return mapMessageRow(row as Record<string, unknown>);
}
async findByChannel(
channelId: string,
query: MessageQuery,
): Promise<PageResult<ReturnType<typeof mapMessageRow>>> {
const pool = getPool();
const db = getDatabase();
const limit = query.limit ?? 50;
const clauses: string[] = ["channel_id = $1"];
const params: (string | number)[] = [channelId];
let p = 2;
const conditions: SQL[] = [eq(messages.channel_id, channelId)];
if (query.cursor) {
clauses.push(`created_at < $${p}`);
params.push(Number(query.cursor));
p++;
conditions.push(lt(messages.created_at, Number(query.cursor)));
}
const { rows } = await pool.query(
`SELECT * FROM messages ${clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY created_at DESC LIMIT $${p}`,
[...params, limit + 1],
);
const rows = await db
.select()
.from(messages)
.where(and(...conditions))
.orderBy(desc(messages.created_at))
.limit(limit + 1);
const data = rows.slice(0, limit).map(mapMessageRow);
const data = rows
.slice(0, limit)
.map((r) => mapMessageRow(r as Record<string, unknown>));
const nextCursor =
rows.length > limit ? String(rows[limit].created_at) : null;
@@ -145,74 +212,67 @@ export class MessagesRepository {
}
async create(data: MessageCreate) {
const pool = getPool();
const db = getDatabase();
const id = crypto.randomUUID();
const { rows } = await pool.query(
`INSERT INTO messages (
id, guild_id, channel_id, thread_id, user_id, username, avatar_url,
content, edited_content, created_at, edited_at, deleted_at, type,
metadata, ai_status
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
RETURNING *`,
[
id,
data.guildId,
data.channelId,
data.threadId ?? null,
data.userId,
data.username,
data.avatarUrl ?? null,
data.content,
null,
Date.now(),
null,
null,
data.type,
null,
"pending",
],
);
return mapMessageRow(rows[0] as Record<string, unknown>);
const [row] = await db
.insert(messages)
.values({
id,
guild_id: data.guildId,
channel_id: data.channelId,
thread_id: data.threadId ?? null,
user_id: data.userId,
username: data.username,
avatar_url: data.avatarUrl ?? null,
content: data.content,
edited_content: null,
created_at: Date.now(),
edited_at: null,
deleted_at: null,
type: data.type ?? "text",
metadata: null,
ai_status: "pending",
})
.returning();
return mapMessageRow(row as Record<string, unknown>);
}
async update(id: string, data: MessageUpdate) {
const pool = getPool();
const db = getDatabase();
// Map camelCase schema keys to snake_case DB columns
const columnMap: Record<keyof MessageUpdate, string> = {
editedContent: "edited_content",
aiStatus: "ai_status",
aiAnalysis: "ai_analysis",
aiCategories: "ai_categories",
aiSeverity: "ai_severity",
aiConfidence: "ai_confidence",
};
const setData: Partial<typeof messages.$inferInsert> = {};
const sets: string[] = [];
const params: unknown[] = [];
let p = 1;
const keys = Object.keys(data) as (keyof MessageUpdate)[];
for (const key of keys) {
const val = data[key];
if (val !== undefined) {
sets.push(`${columnMap[key]} = $${p}`);
params.push(val);
p++;
}
if (data.editedContent !== undefined) {
setData.edited_content = data.editedContent;
}
if (data.aiStatus !== undefined) {
setData.ai_status = data.aiStatus;
}
if (data.aiAnalysis !== undefined) {
setData.ai_analysis = data.aiAnalysis;
}
if (data.aiCategories !== undefined) {
setData.ai_categories = data.aiCategories;
}
if (data.aiSeverity !== undefined) {
setData.ai_severity = data.aiSeverity;
}
if (data.aiConfidence !== undefined) {
setData.ai_confidence = data.aiConfidence;
}
if (sets.length === 0) return this.findById(id);
if (Object.keys(setData).length === 0) return this.findById(id);
params.push(id);
const { rows } = await pool.query(
`UPDATE messages SET ${sets.join(", ")} WHERE id = $${p} RETURNING *`,
params,
);
const [row] = await db
.update(messages)
.set(setData)
.where(eq(messages.id, id))
.returning();
if (rows.length === 0) return null;
return mapMessageRow(rows[0] as Record<string, unknown>);
if (!row) return null;
return mapMessageRow(row as Record<string, unknown>);
}
/**
@@ -227,36 +287,27 @@ export class MessagesRepository {
channelId?: string;
messageIds?: string[];
}): Promise<number> {
const pool = getPool();
const clauses: string[] = ["ai_status = 'error'"];
const params: (string | number)[] = [];
let p = 1;
const db = getDatabase();
const conditions: SQL[] = [eq(messages.ai_status, "error")];
if (opts.messageIds && opts.messageIds.length > 0) {
const placeholders = opts.messageIds.map((_, i) => `$${p + i}`);
clauses.push(`id IN (${placeholders.join(", ")})`);
params.push(...opts.messageIds);
p += opts.messageIds.length;
conditions.push(inArray(messages.id, opts.messageIds));
}
if (opts.guildId) {
clauses.push(`guild_id = $${p}`);
params.push(opts.guildId);
p++;
conditions.push(eq(messages.guild_id, opts.guildId));
}
if (opts.channelId) {
clauses.push(`channel_id = $${p}`);
params.push(opts.channelId);
p++;
conditions.push(eq(messages.channel_id, opts.channelId));
}
const where = clauses.join(" AND ");
const { rowCount } = await pool.query(
`UPDATE messages SET ai_status = 'pending' WHERE ${where}`,
params,
);
const result = await db
.update(messages)
.set({ ai_status: "pending" })
.where(and(...conditions));
logger.info({ count: rowCount ?? 0, ...opts }, "Batch reanalyze triggered");
return rowCount ?? 0;
const count = result.rowCount ?? 0;
logger.info({ count, ...opts }, "Batch reanalyze triggered");
return count;
}
/**
@@ -264,13 +315,11 @@ export class MessagesRepository {
* Skips messages already in 'pending' state to avoid write amplification.
*/
async markForReanalysis(id: string): Promise<void> {
const pool = getPool();
await pool.query(
`UPDATE messages SET ai_status = 'pending'
WHERE id = $1 AND ai_status != 'pending'`,
[id],
);
logger.debug({ id }, "Message marked for re-analysis");
const db = getDatabase();
await db
.update(messages)
.set({ ai_status: "pending" })
.where(and(eq(messages.id, id), ne(messages.ai_status, "pending")));
}
/**
@@ -281,65 +330,64 @@ export class MessagesRepository {
channelId?: string,
limit: number = 20,
): Promise<Record<string, unknown>[]> {
const pool = getPool();
const db = getDatabase();
const conditions: SQL[] = [
inArray(messages.ai_status, ["warn", "flagged"]),
];
if (channelId) {
const { rows } = await pool.query(
`SELECT id, guild_id, channel_id, user_id, username, avatar_url,
content, type, created_at, ai_status, ai_severity,
ai_confidence, ai_analysis
FROM messages
WHERE ai_status IN ('warn', 'flagged')
AND channel_id = $1
ORDER BY created_at DESC
LIMIT $2`,
[channelId, limit],
);
return rows;
conditions.push(eq(messages.channel_id, channelId));
}
const { rows } = await pool.query(
`SELECT id, guild_id, channel_id, user_id, username, avatar_url,
content, type, created_at, ai_status, ai_severity,
ai_confidence, ai_analysis
FROM messages
WHERE ai_status IN ('warn', 'flagged')
ORDER BY created_at DESC
LIMIT $1`,
[limit],
);
return rows;
const rows = await db
.select({
id: messages.id,
guild_id: messages.guild_id,
channel_id: messages.channel_id,
user_id: messages.user_id,
username: messages.username,
avatar_url: messages.avatar_url,
content: messages.content,
type: messages.type,
created_at: messages.created_at,
ai_status: messages.ai_status,
ai_severity: messages.ai_severity,
ai_confidence: messages.ai_confidence,
ai_analysis: messages.ai_analysis,
})
.from(messages)
.where(and(...conditions))
.orderBy(desc(messages.created_at))
.limit(limit);
return rows as unknown as Record<string, unknown>[];
}
async delete(id: string): Promise<boolean> {
const pool = getPool();
const { rowCount } = await pool.query(
`DELETE FROM messages WHERE id = $1`,
[id],
);
return (rowCount ?? 0) > 0;
const db = getDatabase();
const result = await db.delete(messages).where(eq(messages.id, id));
return (result.rowCount ?? 0) > 0;
}
async getAttachmentsByChannel(
channelId: string,
query: MessageQuery,
): Promise<PageResult<AttachmentResult>> {
const pool = getPool();
const db = getDatabase();
const limit = query.limit ?? 50;
const clauses: string[] = ["channel_id = $1"];
const params: (string | number)[] = [channelId];
let p = 2;
const conditions: SQL[] = [eq(attachments.channel_id, channelId)];
if (query.cursor) {
clauses.push(`created_at < $${p}`);
params.push(Number(query.cursor));
p++;
conditions.push(lt(attachments.created_at, Number(query.cursor)));
}
const { rows } = await pool.query(
`SELECT * FROM attachments ${clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY created_at DESC LIMIT $${p}`,
[...params, limit + 1],
);
const rows = await db
.select()
.from(attachments)
.where(and(...conditions))
.orderBy(desc(attachments.created_at))
.limit(limit + 1);
const data = rows.map((r) => ({
id: String(r.id ?? ""),
@@ -6,6 +6,7 @@ const logger = createChildLogger("recordings.service");
export class RecordingsService {
async getRecent(limit = 50) {
logger.info({ limit }, "getRecent called");
const db = getDatabase();
logger.debug({ limit }, "Fetching recent voice recordings");
@@ -1,3 +1,11 @@
import {
COMMAND_GUILDS_LIST,
COMMAND_GUILDS_TEXT_CHANNELS,
COMMAND_VOICE_CHANNELS,
COMMAND_VOICE_CONNECT,
COMMAND_VOICE_DISCONNECT,
VOICE_STATUS_KEY,
} from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { getPool } from "../../shared/database/index.js";
import { publishCommand, readRedisStatus } from "../../shared/redis/index.js";
@@ -28,7 +36,8 @@ export interface VoiceStatus {
* Falls back to database (distinct guild_id from messages) if gateway unreachable.
*/
export async function getGuilds(): Promise<Guild[]> {
const reply = await publishCommand<Guild[]>("guilds:list", {});
logger.info("getGuilds called");
const reply = await publishCommand<Guild[]>(COMMAND_GUILDS_LIST, {});
if (reply?.success && reply.data && reply.data.length > 0) return reply.data;
// Fallback: Postgres with synthetic names
@@ -52,7 +61,8 @@ export async function getGuilds(): Promise<Guild[]> {
* Falls back to database if gateway unreachable.
*/
export async function getTextChannels(guildId: string): Promise<Channel[]> {
const reply = await publishCommand<Channel[]>("guilds:text-channels", {
logger.info({ guildId }, "getTextChannels called");
const reply = await publishCommand<Channel[]>(COMMAND_GUILDS_TEXT_CHANNELS, {
guildId,
});
if (reply?.success && reply.data && reply.data.length > 0) return reply.data;
@@ -79,7 +89,10 @@ export async function getTextChannels(guildId: string): Promise<Channel[]> {
* Get voice channels — query from discord-gateway via Redis command.
*/
export async function getVoiceChannels(guildId: string): Promise<Channel[]> {
const reply = await publishCommand<Channel[]>("voice:channels", { guildId });
logger.info({ guildId }, "getVoiceChannels called");
const reply = await publishCommand<Channel[]>(COMMAND_VOICE_CHANNELS, {
guildId,
});
return reply?.success && reply.data ? reply.data : [];
}
@@ -87,7 +100,8 @@ export async function getVoiceChannels(guildId: string): Promise<Channel[]> {
* Get current voice connection status from Redis cache set by discord-gateway.
*/
export async function getVoiceStatus(): Promise<VoiceStatus> {
const cached = await readRedisStatus("voice:status");
logger.debug("getVoiceStatus called");
const cached = await readRedisStatus(VOICE_STATUS_KEY);
if (cached) return cached as unknown as VoiceStatus;
return {
connected: false,
@@ -104,14 +118,15 @@ export async function connectVoice(
guildId: string,
channelId: string,
): Promise<VoiceStatus> {
const reply = await publishCommand<VoiceStatus>("voice:connect", {
logger.info({ guildId, channelId }, "connectVoice called");
const reply = await publishCommand<VoiceStatus>(COMMAND_VOICE_CONNECT, {
guildId,
channelId,
});
if (reply?.success && reply.data) return reply.data;
// Fallback: read from Redis status key
const cached = await readRedisStatus("voice:status");
const cached = await readRedisStatus(VOICE_STATUS_KEY);
return (
(cached as unknown as VoiceStatus) ?? {
connected: false,
@@ -126,10 +141,11 @@ export async function connectVoice(
* Disconnect from voice via Redis command to discord-gateway.
*/
export async function disconnectVoice(): Promise<VoiceStatus> {
const reply = await publishCommand<VoiceStatus>("voice:disconnect", {});
logger.info("disconnectVoice called");
const reply = await publishCommand<VoiceStatus>(COMMAND_VOICE_DISCONNECT, {});
if (reply?.success && reply.data) return reply.data;
const cached = await readRedisStatus("voice:status");
const cached = await readRedisStatus(VOICE_STATUS_KEY);
return (
(cached as unknown as VoiceStatus) ?? {
connected: false,
+9 -21
View File
@@ -1,28 +1,16 @@
import { randomUUID } from "node:crypto";
import {
BACKEND_COMMAND,
BACKEND_COMMAND_REPLY_PREFIX,
type CommandMessage,
type CommandReply,
} from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import Redis from "ioredis";
import { config } from "../config/index.js";
const logger = createChildLogger("redis.command-channel");
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export interface CommandMessage {
id: string;
type: string;
payload: Record<string, unknown>;
replyChannel: string;
}
export interface CommandReply<T = unknown> {
id: string;
success: boolean;
data?: T;
error?: string;
}
// ---------------------------------------------------------------------------
// Internal Redis clients (singletons)
// ---------------------------------------------------------------------------
@@ -74,7 +62,7 @@ export async function publishCommand<T = unknown>(
}
const id = randomUUID();
const replyChannel = `backend:command:reply:${id}`;
const replyChannel = `${BACKEND_COMMAND_REPLY_PREFIX}${id}`;
const command: CommandMessage = {
id,
type: commandType,
@@ -125,7 +113,7 @@ export async function publishCommand<T = unknown>(
.subscribe(replyChannel)
.then(() => {
pub
.publish("backend:command", JSON.stringify(command))
.publish(BACKEND_COMMAND, JSON.stringify(command))
.then(() => {
logger.debug({ id, commandType }, "Command published");
})
@@ -175,7 +163,7 @@ export async function publishCommandNoReply(
replyChannel: "",
};
await getPublisher().publish("backend:command", JSON.stringify(command));
await getPublisher().publish(BACKEND_COMMAND, JSON.stringify(command));
logger.debug({ id, commandType }, "Command published (no reply)");
}
+26 -12
View File
@@ -1,3 +1,17 @@
import {
DISCORD_ANALYSIS_QUEUE_STATUS,
DISCORD_ATTACHMENT_CREATED,
DISCORD_ATTACHMENT_UPLOADED,
DISCORD_MESSAGE_ANALYZED,
DISCORD_MESSAGE_CREATED,
DISCORD_MESSAGE_DELETED,
DISCORD_MESSAGE_UPDATED,
DISCORD_VOICE_ACTIVE_USER,
DISCORD_VOICE_PCM,
DISCORD_VOICE_STARTED,
DISCORD_VOICE_STOPPED,
DISCORD_VOICE_UPLOADED,
} from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import Redis from "ioredis";
import { config } from "../shared/config/index.js";
@@ -11,21 +25,21 @@ interface ChannelMapping {
}
const SUBSCRIPTIONS: ChannelMapping[] = [
{ channel: "discord:message:created", eventType: "message_created" },
{ channel: "discord:message:updated", eventType: "message_updated" },
{ channel: "discord:message:deleted", eventType: "message_deleted" },
{ channel: "discord:message:analyzed", eventType: "message_analyzed" },
{ channel: "discord:attachment:created", eventType: "attachment_created" },
{ channel: "discord:attachment:uploaded", eventType: "attachment_uploaded" },
{ channel: "discord:voice:started", eventType: "voice_recording_started" },
{ channel: "discord:voice:stopped", eventType: "voice_recording_stopped" },
{ channel: "discord:voice:uploaded", eventType: "voice_recording_uploaded" },
{ channel: DISCORD_MESSAGE_CREATED, eventType: "message_created" },
{ channel: DISCORD_MESSAGE_UPDATED, eventType: "message_updated" },
{ channel: DISCORD_MESSAGE_DELETED, eventType: "message_deleted" },
{ channel: DISCORD_MESSAGE_ANALYZED, eventType: "message_analyzed" },
{ channel: DISCORD_ATTACHMENT_CREATED, eventType: "attachment_created" },
{ channel: DISCORD_ATTACHMENT_UPLOADED, eventType: "attachment_uploaded" },
{ channel: DISCORD_VOICE_STARTED, eventType: "voice_recording_started" },
{ channel: DISCORD_VOICE_STOPPED, eventType: "voice_recording_stopped" },
{ channel: DISCORD_VOICE_UPLOADED, eventType: "voice_recording_uploaded" },
{
channel: "discord:analysis:queue_status",
channel: DISCORD_ANALYSIS_QUEUE_STATUS,
eventType: "analysis_queue_status",
},
{ channel: "discord:voice:active_user", eventType: "voice_active_user" },
{ channel: "discord:voice:pcm", eventType: "voice_pcm_data" },
{ channel: DISCORD_VOICE_ACTIVE_USER, eventType: "voice_active_user" },
{ channel: DISCORD_VOICE_PCM, eventType: "voice_pcm_data" },
];
let subscriber: Redis | null = null;
+3 -2
View File
@@ -1,4 +1,5 @@
import type { Server } from "node:http";
import { BACKEND_COMMAND, BACKEND_VOICE_TRANSMIT } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { WebSocket, WebSocketServer } from "ws";
import { setBroadcastFunctions } from "./broadcast.js";
@@ -92,7 +93,7 @@ export function createWebSocketServer(server: Server): WebSocketServer {
const publisher = getCommandPublisher();
publisher
.publish(
"backend:voice:transmit",
BACKEND_VOICE_TRANSMIT,
JSON.stringify({
type: "pcm",
buffer: message.buffer,
@@ -114,7 +115,7 @@ export function createWebSocketServer(server: Server): WebSocketServer {
const commandId = `cmd-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
publisher
.publish(
"backend:command",
BACKEND_COMMAND,
JSON.stringify({
id: commandId,
type: message.command,
@@ -0,0 +1,162 @@
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../shared/config/config.js";
import type {
AnalysisResult,
MessageRecord,
} from "../message-capture/types.js";
const logger = createChildLogger("auto-delete-eligibility");
/** Parse a config value that may be a JSON array string or a comma-separated list. */
export function parseStringList(value?: string | null): string[] {
if (!value) return [];
try {
const parsed = JSON.parse(value) as unknown;
return Array.isArray(parsed)
? parsed.filter((item): item is string => typeof item === "string")
: [];
} catch {
return value
.split(",")
.map((item) => item.trim())
.filter(Boolean);
}
}
/** Derive severity from legacy messages that lack structured AI fields. */
export function deriveSeverity(msg: MessageRecord): string {
if (msg.ai_severity) return msg.ai_severity;
const score = msg.ai_confidence ?? msg.ai_moderation_score ?? 0;
if (msg.ai_status === "flagged")
return score >= 0.9 ? "critical" : score >= 0.7 ? "high" : "medium";
if (msg.ai_status === "warn") return score >= 0.6 ? "medium" : "low";
return "none";
}
/** Derive recommended action from legacy messages that lack structured AI fields. */
export function deriveRecommendedAction(msg: MessageRecord): string {
if (msg.ai_recommended_action) return msg.ai_recommended_action;
const severity = deriveSeverity(msg);
if (
msg.ai_status === "flagged" &&
(severity === "critical" || severity === "high")
)
return "delete";
if (msg.ai_status === "flagged") return "review";
if (msg.ai_status === "warn") return "warn";
return "none";
}
/**
* Check whether a message qualifies for auto-deletion.
* Uses the structured `analysisResult` fields when provided, falling back
* to legacy message-level AI fields otherwise.
*/
export function isEligibleForAutoDelete(
message: MessageRecord,
analysisResult?: AnalysisResult,
): boolean {
// If analysisResult is provided, use its status field; otherwise use message.ai_status
const status = analysisResult?.status ?? message.ai_status;
if (status !== "flagged" && status !== "warn") {
logger.debug(
{ messageId: message.id, status },
"Message not eligible for auto-delete: status is not flagged or warn",
);
return false;
}
// Confidence check
const confidence =
analysisResult?.confidence ??
message.ai_confidence ??
message.ai_moderation_score ??
0;
if (confidence < config.AUTO_DELETE_MIN_CONFIDENCE) {
logger.debug(
{
messageId: message.id,
confidence,
threshold: config.AUTO_DELETE_MIN_CONFIDENCE,
},
"Message not eligible for auto-delete: confidence below threshold",
);
return false;
}
// Severity check
const severity = analysisResult?.severity ?? deriveSeverity(message);
const allowedSeverities = parseStringList(
config.AUTO_DELETE_ALLOWED_SEVERITIES,
);
if (allowedSeverities.length > 0 && !allowedSeverities.includes(severity)) {
logger.debug(
{ messageId: message.id, severity, allowed: allowedSeverities },
"Message not eligible for auto-delete: severity not in allowed list",
);
return false;
}
// Recommended action check
const recommendedAction =
analysisResult?.recommendedAction ?? deriveRecommendedAction(message);
if (recommendedAction !== "delete" && recommendedAction !== "escalate") {
logger.debug(
{ messageId: message.id, recommendedAction },
"Message not eligible for auto-delete: recommended action is not delete/escalate",
);
return false;
}
// Categories check
const allowedCategories = parseStringList(
config.AUTO_DELETE_ALLOWED_CATEGORIES,
);
if (allowedCategories.length > 0) {
const messageCategories =
analysisResult?.categories ??
parseStringList(message.ai_categories ?? message.ai_moderation_flags);
const hasAllowedCategory = messageCategories.some((cat) =>
allowedCategories.includes(cat),
);
if (!hasAllowedCategory) {
logger.debug(
{
messageId: message.id,
categories: messageCategories,
allowed: allowedCategories,
},
"Message not eligible for auto-delete: no allowed categories match",
);
return false;
}
}
// Excluded channels check
const excludedChannels = parseStringList(
config.AUTO_DELETE_EXCLUDED_CHANNEL_IDS,
);
if (excludedChannels.length > 0) {
const channelId = message.thread_id ?? message.channel_id;
if (excludedChannels.includes(channelId)) {
logger.debug(
{ messageId: message.id, channelId },
"Message not eligible for auto-delete: channel excluded",
);
return false;
}
}
// Excluded users check
const excludedUsers = parseStringList(config.AUTO_DELETE_EXCLUDED_USER_IDS);
if (excludedUsers.length > 0 && excludedUsers.includes(message.user_id)) {
logger.debug(
{ messageId: message.id, userId: message.user_id },
"Message not eligible for auto-delete: user excluded",
);
return false;
}
return true;
}
@@ -0,0 +1,59 @@
import { createChildLogger } from "@bete/shared/logger";
import type { Guild } from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js";
import type { MessageRecord } from "../message-capture/types.js";
interface ChannelWithSend {
send: (content: string | object, options?: unknown) => Promise<unknown>;
}
const logger = createChildLogger("auto-delete-logger");
/**
* Post a log message about the auto-deletion to the configured moderation log channel.
* If AUTO_DELETE_LOG_CHANNEL_ID is not set, this is a no-op.
* Failures (channel not found, missing permissions) are logged as warnings.
*/
export async function logDeletionToChannel(
guild: Guild,
message: MessageRecord,
channelId: string,
): Promise<void> {
if (!config.AUTO_DELETE_LOG_CHANNEL_ID) return;
try {
const logChannel = guild.channels.cache.get(
config.AUTO_DELETE_LOG_CHANNEL_ID,
);
if (
logChannel &&
"send" in logChannel &&
typeof (logChannel as ChannelWithSend).send === "function"
) {
const severity = message.ai_severity ?? "none";
const categories =
message.ai_categories ?? message.ai_moderation_flags ?? "—";
const snippet = (message.edited_content ?? message.content).substring(
0,
200,
);
await (logChannel as ChannelWithSend).send(
`**🧹 Auto-Delete** — Pesan dari <@${message.user_id}> di <#${channelId}>\n` +
`**Status:** ${message.ai_status}\n` +
`**Severitas:** ${severity}\n` +
`**Kategori:** ${categories}\n` +
`**Isi:** ${snippet}\n` +
`**Waktu:** <t:${Math.floor(Date.now() / 1000)}:R>`,
);
logger.info(
{ channelId, messageId: message.id },
"Deletion logged to channel",
);
}
} catch (logErr) {
logger.warn(
{ messageId: message.id, error: String(logErr) },
"Failed to log auto-delete to moderation channel",
);
}
}
@@ -3,178 +3,20 @@ import type { Client, PermissionString } from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js";
import { createModerationAction } from "../message-capture/messageStore.js";
import type { MessageRecord } from "../message-capture/types.js";
interface ChannelWithSend {
send: (content: string | object, options?: unknown) => Promise<unknown>;
}
import { isEligibleForAutoDelete } from "./autoDeleteEligibility.js";
import { logDeletionToChannel } from "./autoDeleteLogger.js";
import { sendDeletionNotification } from "./autoDeleteNotify.js";
const logger = createChildLogger("auto-delete-manager");
const parseStringList = (value?: string | null): string[] => {
if (!value) return [];
try {
const parsed = JSON.parse(value) as unknown;
return Array.isArray(parsed)
? parsed.filter((item): item is string => typeof item === "string")
: [];
} catch {
return value
.split(",")
.map((item) => item.trim())
.filter(Boolean);
}
};
/** Derive severity from legacy messages that lack structured AI fields. */
function deriveSeverity(msg: MessageRecord): string {
if (msg.ai_severity) return msg.ai_severity;
const score = msg.ai_confidence ?? msg.ai_moderation_score ?? 0;
if (msg.ai_status === "flagged")
return score >= 0.9 ? "critical" : score >= 0.7 ? "high" : "medium";
if (msg.ai_status === "warn") return score >= 0.6 ? "medium" : "low";
return "none";
}
/** Derive recommended action from legacy messages that lack structured AI fields. */
function deriveRecommendedAction(msg: MessageRecord): string {
if (msg.ai_recommended_action) return msg.ai_recommended_action;
const severity = deriveSeverity(msg);
if (
msg.ai_status === "flagged" &&
(severity === "critical" || severity === "high")
)
return "delete";
if (msg.ai_status === "flagged") return "review";
if (msg.ai_status === "warn") return "warn";
return "none";
}
function isAutoDeleteEligible(message: MessageRecord): boolean {
if (message.ai_status !== "flagged" && message.ai_status !== "warn")
return false;
const confidence = message.ai_confidence ?? message.ai_moderation_score ?? 0;
if (confidence < config.AUTO_DELETE_MIN_CONFIDENCE) {
logger.debug(
{
messageId: message.id,
confidence,
threshold: config.AUTO_DELETE_MIN_CONFIDENCE,
},
"Auto-delete skipped: confidence below threshold",
);
return false;
}
const severity = deriveSeverity(message);
const allowedSeverities = (config.AUTO_DELETE_ALLOWED_SEVERITIES || "")
.split(",")
.map((s) => s.trim())
.filter(Boolean);
if (allowedSeverities.length > 0 && !allowedSeverities.includes(severity)) {
logger.debug(
{ messageId: message.id, severity, allowed: allowedSeverities },
"Auto-delete skipped: severity not in allowed list",
);
return false;
}
const recommendedAction = deriveRecommendedAction(message);
if (recommendedAction !== "delete" && recommendedAction !== "escalate") {
logger.debug(
{ messageId: message.id, recommendedAction },
"Auto-delete skipped: recommended action is not delete/escalate",
);
return false;
}
const allowedCategories = parseStringList(
config.AUTO_DELETE_ALLOWED_CATEGORIES,
);
if (allowedCategories.length > 0) {
const messageCategories = parseStringList(
message.ai_categories ?? message.ai_moderation_flags,
);
const hasAllowedCategory = messageCategories.some((cat) =>
allowedCategories.includes(cat),
);
if (!hasAllowedCategory) {
logger.debug(
{
messageId: message.id,
categories: messageCategories,
allowed: allowedCategories,
},
"Auto-delete skipped: no allowed categories match",
);
return false;
}
}
const excludedChannels = parseStringList(
config.AUTO_DELETE_EXCLUDED_CHANNEL_IDS,
);
if (excludedChannels.length > 0) {
const channelId = message.thread_id ?? message.channel_id;
if (excludedChannels.includes(channelId)) {
logger.debug(
{ messageId: message.id, channelId },
"Auto-delete skipped: channel excluded",
);
return false;
}
}
const excludedUsers = parseStringList(config.AUTO_DELETE_EXCLUDED_USER_IDS);
if (excludedUsers.length > 0 && excludedUsers.includes(message.user_id)) {
logger.debug(
{ messageId: message.id, userId: message.user_id },
"Auto-delete skipped: user excluded",
);
return false;
}
return true;
}
async function logAutoDeleteAttempt(
message: MessageRecord,
result: AutoDeleteResult,
): Promise<void> {
try {
await createModerationAction({
message_id: message.id,
user_id: message.user_id,
guild_id: message.guild_id,
action_type: "delete_message",
reason: result.reason,
executed_by: "auto-delete-manager",
status: result.deleted
? "executed"
: result.reason === "dry_run"
? "executed"
: "failed",
error: result.reason === "error" ? result.reason : null,
executed_at:
result.deleted || result.reason === "dry_run" ? Date.now() : null,
});
} catch (error) {
logger.warn(
{
messageId: message.id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to persist auto-delete action log",
);
}
}
export interface AutoDeleteResult {
deleted: boolean;
skipped: boolean;
reason: string;
}
// ─── Error Handling Utilities ────────────────────────────────────────
function getErrorCode(error: unknown): number | string | undefined {
if (!error || typeof error !== "object") return undefined;
const maybeCode = (error as { code?: number | string }).code;
@@ -216,42 +58,89 @@ function hasPermissionApi(channel: unknown): channel is {
);
}
// ─── Database Action Log ─────────────────────────────────────────────
async function logAutoDeleteAttempt(
message: MessageRecord,
result: AutoDeleteResult,
): Promise<void> {
try {
await createModerationAction({
message_id: message.id,
user_id: message.user_id,
guild_id: message.guild_id,
action_type: "delete_message",
reason: result.reason,
executed_by: "auto-delete-manager",
status: result.deleted
? "executed"
: result.reason === "dry_run"
? "executed"
: "failed",
error: result.reason === "error" ? result.reason : null,
executed_at:
result.deleted || result.reason === "dry_run" ? Date.now() : null,
});
} catch (error) {
logger.warn(
{
messageId: message.id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to persist auto-delete action log",
);
}
}
// ─── Main Orchestrator ───────────────────────────────────────────────
export async function attemptAutoDeleteFlaggedMessage(
client: Client | undefined,
message: MessageRecord,
): Promise<AutoDeleteResult> {
logger.debug({ messageId: message.id }, "Processing message for auto-delete");
// ── Config gate ──────────────────────────────────────────────────
if (!config.AUTO_DELETE_FLAGGED_ENABLED) {
logger.debug({ messageId: message.id }, "Auto-delete disabled by config");
return { deleted: false, skipped: true, reason: "disabled" };
}
// ── Status gate ──────────────────────────────────────────────────
if (message.ai_status !== "flagged" && message.ai_status !== "warn") {
logger.debug(
{ messageId: message.id, status: message.ai_status },
"Auto-delete skipped: message not flagged or warned",
);
const result = {
const result: AutoDeleteResult = {
deleted: false,
skipped: true,
reason: "not_flagged_or_warn",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
return result;
}
if (!isAutoDeleteEligible(message)) {
// ── Eligibility gate ─────────────────────────────────────────────
if (!isEligibleForAutoDelete(message)) {
logger.debug(
{ messageId: message.id },
"Auto-delete skipped: not eligible (confidence/severity/action/category filter)",
);
const result = {
const result: AutoDeleteResult = {
deleted: false,
skipped: true,
reason: "not_eligible",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
return result;
}
// ── Client check ─────────────────────────────────────────────────
if (!client?.user?.id) {
logger.warn(
{ messageId: message.id },
@@ -260,6 +149,8 @@ export async function attemptAutoDeleteFlaggedMessage(
return { deleted: false, skipped: true, reason: "client_user_missing" };
}
// ── Deletion flow ────────────────────────────────────────────────
try {
const guild = client.guilds.cache.get(message.guild_id);
if (!guild) {
@@ -305,12 +196,14 @@ export async function attemptAutoDeleteFlaggedMessage(
};
}
// ── Dry run mode ───────────────────────────────────────────────
if (config.AUTO_DELETE_FLAGGED_DRY_RUN) {
const result = {
const result: AutoDeleteResult = {
deleted: false,
skipped: true,
reason: "dry_run",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
logger.info(
{ messageId: message.id, channelId },
@@ -319,75 +212,30 @@ export async function attemptAutoDeleteFlaggedMessage(
return result;
}
// ── Perform API deletion ───────────────────────────────────────
const discordMessage = await channel.messages.fetch(message.id);
await discordMessage.delete();
logger.info(
{ messageId: message.id, channelId },
"Message deleted from Discord",
);
// ── Notify user via DM ──
if (config.AUTO_DELETE_NOTIFY_USER) {
try {
const targetUser = await client.users.fetch(message.user_id);
if (targetUser) {
const reason =
message.ai_categories ?? message.ai_moderation_flags ?? "(unknown)";
await targetUser.send(
`Pesan Anda di **${guild.name}** telah dihapus oleh sistem moderasi otomatis.\n` +
`Alasan: ${reason}\n` +
`Jika Anda merasa ini adalah kesalahan, silakan hubungi admin server.`,
);
}
} catch (dmErr) {
// DM might fail if user has DMs disabled — not critical
logger.debug(
{
messageId: message.id,
userId: message.user_id,
error: String(dmErr),
},
"Failed to send DM notification for auto-deleted message",
);
}
}
// ── Notify user via DM ─────────────────────────────────────────
// ── Log to moderation channel ──
if (config.AUTO_DELETE_LOG_CHANNEL_ID) {
try {
const logChannel = guild.channels.cache.get(
config.AUTO_DELETE_LOG_CHANNEL_ID,
);
if (
logChannel &&
"send" in logChannel &&
typeof (logChannel as ChannelWithSend).send === "function"
) {
const severity = message.ai_severity ?? "none";
const categories =
message.ai_categories ?? message.ai_moderation_flags ?? "—";
const snippet = (message.edited_content ?? message.content).substring(
0,
200,
);
await (logChannel as ChannelWithSend).send(
`**🧹 Auto-Delete** — Pesan dari <@${message.user_id}> di <#${channelId}>\n` +
`**Status:** ${message.ai_status}\n` +
`**Severitas:** ${severity}\n` +
`**Kategori:** ${categories}\n` +
`**Isi:** ${snippet}\n` +
`**Waktu:** <t:${Math.floor(Date.now() / 1000)}:R>`,
);
}
} catch (logErr) {
logger.warn(
{ messageId: message.id, error: String(logErr) },
"Failed to log auto-delete to moderation channel",
);
}
}
await sendDeletionNotification(client, message, guild.name);
const result = {
// ── Log to moderation channel ──────────────────────────────────
await logDeletionToChannel(guild, message, channelId);
// ── Success ────────────────────────────────────────────────────
const result: AutoDeleteResult = {
deleted: true,
skipped: false,
reason: "deleted",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
logger.info(
{ messageId: message.id, channelId },
@@ -395,12 +243,14 @@ export async function attemptAutoDeleteFlaggedMessage(
);
return result;
} catch (error) {
// ── Already deleted ──────────────────────────────────────────────
if (isAlreadyDeletedError(error)) {
const result = {
const result: AutoDeleteResult = {
deleted: true,
skipped: false,
reason: "already_deleted",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
logger.info(
{ messageId: message.id, code: getErrorCode(error) },
@@ -409,11 +259,13 @@ export async function attemptAutoDeleteFlaggedMessage(
return result;
}
const result = {
// ── Unexpected error ─────────────────────────────────────────────
const result: AutoDeleteResult = {
deleted: false,
skipped: true,
reason: "error",
} as AutoDeleteResult;
};
await logAutoDeleteAttempt(message, result);
logger.error(
{
@@ -0,0 +1,46 @@
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js";
import type { MessageRecord } from "../message-capture/types.js";
const logger = createChildLogger("auto-delete-notify");
/**
* Send a DM notification to the user whose message was auto-deleted.
* If AUTO_DELETE_NOTIFY_USER is disabled, this is a no-op.
* DM failures (user has DMs disabled, etc.) are logged at debug level and swallowed.
*/
export async function sendDeletionNotification(
client: Client,
message: MessageRecord,
guildName: string,
): Promise<void> {
if (!config.AUTO_DELETE_NOTIFY_USER) return;
try {
const targetUser = await client.users.fetch(message.user_id);
if (targetUser) {
const reason: string =
message.ai_categories ?? message.ai_moderation_flags ?? "(unknown)";
await targetUser.send(
`Pesan Anda di **${guildName}** telah dihapus oleh sistem moderasi otomatis.\n` +
`Alasan: ${reason}\n` +
`Jika Anda merasa ini adalah kesalahan, silakan hubungi admin server.`,
);
logger.info(
{ userId: message.user_id, messageId: message.id },
"Deletion notification sent",
);
}
} catch (dmErr) {
// DM might fail if user has DMs disabled — not critical
logger.debug(
{
messageId: message.id,
userId: message.user_id,
error: String(dmErr),
},
"Failed to send DM notification for auto-deleted message",
);
}
}
@@ -37,6 +37,10 @@ export async function uploadAttachmentToTele(
filename: string,
contentType = "application/octet-stream",
): Promise<string> {
logger.debug(
{ filename, sizeBytes: fileBuffer.length },
"Starting attachment upload to tele",
);
try {
const result = await uploadToTele({
buffer: fileBuffer,
@@ -47,6 +51,10 @@ export async function uploadAttachmentToTele(
retries: 0,
});
logger.info(
{ filename, url: result.url },
"Attachment uploaded to tele successfully",
);
return result.url;
} catch (error) {
logger.error(
@@ -61,6 +69,7 @@ export async function uploadAttachmentToTele(
}
export async function downloadDiscordAttachment(url: string): Promise<Buffer> {
logger.debug({ url }, "Starting Discord attachment download");
try {
const response = await fetch(url, {
signal: AbortSignal.timeout(config.ATTACHMENT_UPLOAD_TIMEOUT_MS),
@@ -74,7 +83,12 @@ export async function downloadDiscordAttachment(url: string): Promise<Buffer> {
}
const buffer = await response.arrayBuffer();
return Buffer.from(buffer);
const result = Buffer.from(buffer);
logger.debug(
{ url, sizeBytes: result.length },
"Discord attachment downloaded successfully",
);
return result;
} catch (error) {
logger.error(
{ url, error: toErrorMessage(error) },
@@ -93,6 +107,7 @@ export async function processAttachmentUpload(
contentType?: string;
} = {},
): Promise<void> {
logger.info({ attachmentId, filename }, "processAttachmentUpload called");
try {
let currentDiscordUrl = discordUrl;
let buffer: Buffer;
@@ -103,6 +118,10 @@ export async function processAttachmentUpload(
throw error;
}
logger.warn(
{ attachmentId, filename },
"Discord URL expired, refreshing and retrying",
);
const freshUrl = await options.refreshDiscordUrl();
if (!freshUrl) throw error;
currentDiscordUrl = freshUrl;
@@ -111,6 +130,10 @@ export async function processAttachmentUpload(
}
const sizeMb = buffer.length / (1024 * 1024);
logger.debug(
{ attachmentId, sizeMb: sizeMb.toFixed(2) },
"Attachment size check",
);
if (sizeMb > config.ATTACHMENT_MAX_SIZE_MB) {
throw new Error(
`File size ${sizeMb.toFixed(2)}MB exceeds limit of ${config.ATTACHMENT_MAX_SIZE_MB}MB`,
@@ -124,6 +147,10 @@ export async function processAttachmentUpload(
);
await updateAttachmentAsUploaded(attachmentId, uploadedUrl, Date.now());
logger.info(
{ attachmentId, url: uploadedUrl },
"Attachment upload completed successfully",
);
} catch (error) {
const errorMsg = toErrorMessage(error);
await updateAttachmentAsFailedUpload(attachmentId, errorMsg);
@@ -1,11 +1,23 @@
import {
BACKEND_COMMAND,
type CommandMessage,
type CommandReply,
MEDIA_STATUS_KEY,
VOICE_STATUS_KEY,
} from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
import Redis from "ioredis";
import { config } from "../../shared/config/config.js";
import { createModerationAction } from "../message-capture/messageStore.js";
import { discordPlayer } from "../voice-recording/player.js";
import { voiceTransmitter } from "../voice-recording/transmitter.js";
import type { VoiceController } from "../voice-recording/voiceController.js";
import { GuildHandler } from "./guild.handler.js";
import {
type CommandHandlerFn,
createHandlerRegistry,
} from "./handler-registry.js";
import { MediaHandler } from "./media.handler.js";
import { ModerationHandler } from "./moderation.handler.js";
import { VoiceHandler } from "./voice.handler.js";
const logger = createChildLogger("command-handler");
@@ -13,20 +25,6 @@ const logger = createChildLogger("command-handler");
// Types
// ---------------------------------------------------------------------------
interface BackendCommand {
id: string;
type: string;
payload: Record<string, unknown>;
replyChannel: string;
}
interface CommandReply {
id: string;
success: boolean;
data: unknown;
error?: string;
}
interface VoiceStatusPayload {
connected: boolean;
activeGuildId: string | null;
@@ -34,21 +32,6 @@ interface VoiceStatusPayload {
activeChannelName: string | null;
}
interface MediaStatusPayload {
playing: boolean;
musicVolume: number;
current: unknown;
queue: unknown[];
}
// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------
const COMMAND_CHANNEL = "backend:command";
const VOICE_STATUS_KEY = "voice:status";
const MEDIA_STATUS_KEY = "media:status";
// ---------------------------------------------------------------------------
// CommandHandler
// ---------------------------------------------------------------------------
@@ -56,8 +39,12 @@ const MEDIA_STATUS_KEY = "media:status";
export class CommandHandler {
private redisSub: Redis;
private redisPub: Redis;
private client: Client | null = null;
private voiceController: VoiceController | null = null;
private registry: Map<string, CommandHandlerFn> = new Map();
private voiceHandler!: VoiceHandler;
private mediaHandler!: MediaHandler;
private guildHandler!: GuildHandler;
private moderationHandler!: ModerationHandler;
constructor() {
this.redisSub = new Redis(config.REDIS_URL);
@@ -79,9 +66,22 @@ export class CommandHandler {
* command channel. Must be called *after* the Discord client is created.
*/
start(client: Client, voiceController: VoiceController): void {
this.client = client;
this.voiceController = voiceController;
// Create domain-specific handlers with their dependencies
this.voiceHandler = new VoiceHandler(client, voiceController);
this.mediaHandler = new MediaHandler();
this.guildHandler = new GuildHandler(client);
this.moderationHandler = new ModerationHandler(client);
// Build the command registry
this.registry = createHandlerRegistry(
this.voiceHandler,
this.mediaHandler,
this.guildHandler,
this.moderationHandler,
);
this.redisSub.on("message", (_channel, message) => {
this.handleCommand(message).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
@@ -89,11 +89,11 @@ export class CommandHandler {
});
});
this.redisSub.subscribe(COMMAND_CHANNEL, (err) => {
this.redisSub.subscribe(BACKEND_COMMAND, (err) => {
if (err) {
logger.error({ error: err }, "Failed to subscribe to command channel");
} else {
logger.info(`Subscribed to Redis channel "${COMMAND_CHANNEL}"`);
logger.info(`Subscribed to Redis channel "${BACKEND_COMMAND}"`);
}
});
@@ -109,9 +109,9 @@ export class CommandHandler {
// ---- Command dispatch ----
private async handleCommand(raw: string): Promise<void> {
let cmd: BackendCommand;
let cmd: CommandMessage;
try {
cmd = JSON.parse(raw) as BackendCommand;
cmd = JSON.parse(raw) as CommandMessage;
} catch {
logger.warn({ raw }, "Received invalid JSON on command channel");
return;
@@ -119,54 +119,20 @@ export class CommandHandler {
logger.info({ commandId: cmd.id, type: cmd.type }, "Received command");
let reply: CommandReply;
let reply: CommandReply<unknown>;
try {
switch (cmd.type) {
case "voice:connect":
reply = await this.handleVoiceConnect(cmd);
break;
case "voice:disconnect":
reply = await this.handleVoiceDisconnect(cmd);
break;
case "voice:channels":
reply = await this.handleVoiceChannels(cmd);
break;
case "voice:transmit:start":
reply = await this.handleVoiceTransmitStart(cmd);
break;
case "voice:transmit:stop":
reply = await this.handleVoiceTransmitStop(cmd);
break;
case "guilds:list":
reply = await this.handleListGuilds(cmd);
break;
case "guilds:text-channels":
reply = await this.handleTextChannels(cmd);
break;
case "media:queue":
reply = await this.handleMediaQueue(cmd);
break;
case "media:skip":
reply = await this.handleMediaSkip(cmd);
break;
case "media:stop":
reply = await this.handleMediaStop(cmd);
break;
case "media:volume":
reply = await this.handleMediaVolume(cmd);
break;
case "moderation:action":
reply = await this.handleModerationAction(cmd);
break;
default:
logger.warn({ type: cmd.type }, "Unknown command type");
reply = {
id: cmd.id,
success: false,
data: null,
error: `Unknown command type: ${cmd.type}`,
};
const handler = this.registry.get(cmd.type);
if (handler) {
reply = await handler(cmd);
} else {
logger.warn({ type: cmd.type }, "Unknown command type");
reply = {
id: cmd.id,
success: false,
data: null,
error: `Unknown command type: ${cmd.type}`,
};
}
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
@@ -195,383 +161,6 @@ export class CommandHandler {
this.publishMediaStatus();
}
// ---- Command handlers ----
private async handleVoiceConnect(cmd: BackendCommand): Promise<CommandReply> {
if (!this.client || !this.voiceController) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
const channelId = String(cmd.payload.channelId ?? "");
if (!guildId || !channelId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId and channelId are required",
};
}
const status = await this.voiceController.connect(guildId, channelId);
return { id: cmd.id, success: true, data: status };
}
private async handleVoiceDisconnect(
cmd: BackendCommand,
): Promise<CommandReply> {
if (!this.voiceController) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const status = await this.voiceController.disconnect();
return { id: cmd.id, success: true, data: status };
}
private async handleVoiceChannels(
cmd: BackendCommand,
): Promise<CommandReply> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
if (!guildId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId is required",
};
}
try {
const guild = await this.client.guilds.fetch(guildId);
const channels = await guild.channels.fetch();
const voiceChannels = channels
.filter((c) => c?.type === "GUILD_VOICE")
.map((c) => ({
id: c.id,
name: c.name,
type: "voice" as const,
}));
return { id: cmd.id, success: true, data: voiceChannels };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
private async handleListGuilds(cmd: BackendCommand): Promise<CommandReply> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
try {
const guilds = this.client.guilds.cache.map((g) => ({
id: g.id,
name: g.name,
icon: g.iconURL() ?? null,
}));
return { id: cmd.id, success: true, data: guilds };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
private async handleTextChannels(cmd: BackendCommand): Promise<CommandReply> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
if (!guildId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId is required",
};
}
try {
const guild = await this.client.guilds.fetch(guildId);
const channels = await guild.channels.fetch();
const textChannels = channels
.filter((c) => c?.type === "GUILD_TEXT")
.map((c) => ({
id: c.id,
name: c.name,
type: "text" as const,
}));
return { id: cmd.id, success: true, data: textChannels };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
private getCurrentMediaStatus(): MediaStatusPayload {
return {
playing: discordPlayer.getStatus() === "playing",
musicVolume: discordPlayer.getMusicVolume(),
current: null,
queue: [],
};
}
private async handleMediaQueue(cmd: BackendCommand): Promise<CommandReply> {
// Media queueing is handled at a higher level (frontend / backend streams
// audio directly). Log the request for now.
logger.info("media:queue received — media queueing is handled externally");
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
private async handleMediaSkip(cmd: BackendCommand): Promise<CommandReply> {
discordPlayer.stop("music");
return { id: cmd.id, success: true, data: this.getCurrentMediaStatus() };
}
private async handleMediaStop(cmd: BackendCommand): Promise<CommandReply> {
discordPlayer.stop("music");
return { id: cmd.id, success: true, data: this.getCurrentMediaStatus() };
}
private async handleMediaVolume(cmd: BackendCommand): Promise<CommandReply> {
const volume = Number(cmd.payload.volume);
if (!Number.isFinite(volume)) {
return {
id: cmd.id,
success: false,
data: null,
error: "volume must be a number",
};
}
discordPlayer.setMusicVolume(volume);
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
private async handleVoiceTransmitStart(
cmd: BackendCommand,
): Promise<CommandReply> {
if (!discordPlayer.isConnected()) {
return {
id: cmd.id,
success: false,
data: null,
error: "Not connected to voice channel",
};
}
try {
// Create a new Redis connection for the transmitter
const transmitRedis = new Redis(config.REDIS_URL);
await voiceTransmitter.start(transmitRedis);
const status = voiceTransmitter.getStatus();
logger.info({ status }, "Voice transmit started");
return {
id: cmd.id,
success: true,
data: status,
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
logger.error({ error: message }, "Failed to start voice transmit");
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
private async handleVoiceTransmitStop(
cmd: BackendCommand,
): Promise<CommandReply> {
try {
await voiceTransmitter.stop();
logger.info("Voice transmit stopped");
return {
id: cmd.id,
success: true,
data: { status: "stopped" },
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
logger.error({ error: message }, "Failed to stop voice transmit");
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
private async handleModerationAction(
cmd: BackendCommand,
): Promise<CommandReply> {
const payload = cmd.payload as {
message_id?: string;
user_id?: string;
guild_id?: string;
channel_id?: string;
action_type?: string;
reason?: string;
executed_by?: string;
};
if (
!payload.message_id ||
!payload.user_id ||
!payload.guild_id ||
!payload.action_type
) {
return {
id: cmd.id,
success: false,
data: null,
error: "message_id, user_id, guild_id, and action_type are required",
};
}
const validActions = [
"delete_message",
"mute_user",
"warn_user",
"kick_user",
"ban_user",
] as const;
if (
!validActions.includes(
payload.action_type as (typeof validActions)[number],
)
) {
return {
id: cmd.id,
success: false,
data: null,
error: `Invalid action_type: ${payload.action_type}. Must be one of: ${validActions.join(", ")}`,
};
}
try {
// For delete_message, also actually delete via Discord if client is available
if (payload.action_type === "delete_message" && this.client) {
try {
const channelId = String(cmd.payload.channel_id ?? "");
if (channelId) {
const channel = await this.client.channels.fetch(channelId);
if (channel?.isText()) {
const msg = await channel.messages
.fetch(payload.message_id)
.catch(() => null);
if (msg) {
await msg.delete().catch((err: unknown) => {
logger.warn(
{ error: err, messageId: payload.message_id },
"Failed to delete message via Discord",
);
});
}
}
}
} catch (err) {
logger.warn(
{ error: err, messageId: payload.message_id },
"Failed to fetch channel/message for deletion",
);
}
}
const action = await createModerationAction({
message_id: payload.message_id,
user_id: payload.user_id,
guild_id: payload.guild_id,
action_type: payload.action_type as
| "delete_message"
| "mute_user"
| "warn_user"
| "kick_user"
| "ban_user",
reason: payload.reason ?? null,
executed_by: payload.executed_by ?? "command-handler",
status: "executed",
error: null,
executed_at: Date.now(),
});
logger.info(
{
actionId: action.id,
actionType: payload.action_type,
userId: payload.user_id,
},
"Moderation action executed",
);
return {
id: cmd.id,
success: true,
data: action,
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
logger.error(
{ error: message, commandId: cmd.id },
"Failed to execute moderation action",
);
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
// ---- Status publishing ----
private publishVoiceStatus(): void {
@@ -588,7 +177,10 @@ export class CommandHandler {
}
private publishMediaStatus(): void {
this.setKey(MEDIA_STATUS_KEY, JSON.stringify(this.getCurrentMediaStatus()));
this.setKey(
MEDIA_STATUS_KEY,
JSON.stringify(this.mediaHandler.getCurrentMediaStatus()),
);
}
/**
@@ -0,0 +1,86 @@
import { type CommandMessage, type CommandReply } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
// ---------------------------------------------------------------------------
// GuildHandler
// ---------------------------------------------------------------------------
export class GuildHandler {
private logger = createChildLogger("guild-handler");
constructor(private client: Client | null) {}
setClient(client: Client): void {
this.client = client;
}
async handleListGuilds(cmd: CommandMessage): Promise<CommandReply<unknown>> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
try {
const guilds = this.client.guilds.cache.map((g) => ({
id: g.id,
name: g.name,
icon: g.iconURL() ?? null,
}));
return { id: cmd.id, success: true, data: guilds };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
this.logger.error({ error: msg }, "Failed to list guilds");
return { id: cmd.id, success: false, data: null, error: msg };
}
}
async handleTextChannels(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
if (!guildId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId is required",
};
}
try {
const guild = await this.client.guilds.fetch(guildId);
const channels = await guild.channels.fetch();
const textChannels = channels
.filter((c) => c?.type === "GUILD_TEXT")
.map((c) => ({
id: c.id,
name: c.name,
type: "text" as const,
}));
return { id: cmd.id, success: true, data: textChannels };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
this.logger.error(
{ error: msg, guildId },
"Failed to list text channels",
);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
}
@@ -0,0 +1,83 @@
import {
COMMAND_GUILDS_LIST,
COMMAND_GUILDS_TEXT_CHANNELS,
COMMAND_MEDIA_QUEUE,
COMMAND_MEDIA_SKIP,
COMMAND_MEDIA_STOP,
COMMAND_MEDIA_VOLUME,
COMMAND_MODERATION_ACTION,
COMMAND_VOICE_CHANNELS,
COMMAND_VOICE_CONNECT,
COMMAND_VOICE_DISCONNECT,
COMMAND_VOICE_TRANSMIT_START,
COMMAND_VOICE_TRANSMIT_STOP,
type CommandMessage,
type CommandReply,
} from "@bete/shared";
import type { GuildHandler } from "./guild.handler.js";
import type { MediaHandler } from "./media.handler.js";
import type { ModerationHandler } from "./moderation.handler.js";
import type { VoiceHandler } from "./voice.handler.js";
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export type CommandHandlerFn = (
cmd: CommandMessage,
) => Promise<CommandReply<unknown>>;
// ---------------------------------------------------------------------------
// Registry factory
// ---------------------------------------------------------------------------
export function createHandlerRegistry(
voiceHandler: VoiceHandler,
mediaHandler: MediaHandler,
guildHandler: GuildHandler,
moderationHandler: ModerationHandler,
): Map<string, CommandHandlerFn> {
const registry = new Map<string, CommandHandlerFn>();
// Voice commands
registry.set(COMMAND_VOICE_CONNECT, (cmd) =>
voiceHandler.handleVoiceConnect(cmd),
);
registry.set(COMMAND_VOICE_DISCONNECT, (cmd) =>
voiceHandler.handleVoiceDisconnect(cmd),
);
registry.set(COMMAND_VOICE_CHANNELS, (cmd) =>
voiceHandler.handleVoiceChannels(cmd),
);
registry.set(COMMAND_VOICE_TRANSMIT_START, (cmd) =>
voiceHandler.handleVoiceTransmitStart(cmd),
);
registry.set(COMMAND_VOICE_TRANSMIT_STOP, (cmd) =>
voiceHandler.handleVoiceTransmitStop(cmd),
);
// Media commands
registry.set(COMMAND_MEDIA_QUEUE, (cmd) =>
mediaHandler.handleMediaQueue(cmd),
);
registry.set(COMMAND_MEDIA_SKIP, (cmd) => mediaHandler.handleMediaSkip(cmd));
registry.set(COMMAND_MEDIA_STOP, (cmd) => mediaHandler.handleMediaStop(cmd));
registry.set(COMMAND_MEDIA_VOLUME, (cmd) =>
mediaHandler.handleMediaVolume(cmd),
);
// Guild commands
registry.set(COMMAND_GUILDS_LIST, (cmd) =>
guildHandler.handleListGuilds(cmd),
);
registry.set(COMMAND_GUILDS_TEXT_CHANNELS, (cmd) =>
guildHandler.handleTextChannels(cmd),
);
// Moderation commands
registry.set(COMMAND_MODERATION_ACTION, (cmd) =>
moderationHandler.handleModerationAction(cmd),
);
return registry;
}
@@ -0,0 +1,78 @@
import { type CommandMessage, type CommandReply } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { discordPlayer } from "../voice-recording/player.js";
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export interface MediaStatusPayload {
playing: boolean;
musicVolume: number;
current: unknown;
queue: unknown[];
}
// ---------------------------------------------------------------------------
// MediaHandler
// ---------------------------------------------------------------------------
export class MediaHandler {
private logger = createChildLogger("media-handler");
getCurrentMediaStatus(): MediaStatusPayload {
return {
playing: discordPlayer.getStatus() === "playing",
musicVolume: discordPlayer.getMusicVolume(),
current: null,
queue: [],
};
}
async handleMediaQueue(cmd: CommandMessage): Promise<CommandReply<unknown>> {
this.logger.info(
"media:queue received — media queueing is handled externally",
);
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
async handleMediaSkip(cmd: CommandMessage): Promise<CommandReply<unknown>> {
discordPlayer.stop("music");
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
async handleMediaStop(cmd: CommandMessage): Promise<CommandReply<unknown>> {
discordPlayer.stop("music");
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
async handleMediaVolume(cmd: CommandMessage): Promise<CommandReply<unknown>> {
const volume = Number(cmd.payload.volume);
if (!Number.isFinite(volume)) {
return {
id: cmd.id,
success: false,
data: null,
error: "volume must be a number",
};
}
discordPlayer.setMusicVolume(volume);
return {
id: cmd.id,
success: true,
data: this.getCurrentMediaStatus(),
};
}
}
@@ -0,0 +1,140 @@
import { type CommandMessage, type CommandReply } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
import { createModerationAction } from "../message-capture/messageStore.js";
// ---------------------------------------------------------------------------
// ModerationHandler
// ---------------------------------------------------------------------------
export class ModerationHandler {
private logger = createChildLogger("moderation-handler");
constructor(private client: Client | null) {}
setClient(client: Client): void {
this.client = client;
}
async handleModerationAction(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
const payload = cmd.payload as {
message_id?: string;
user_id?: string;
guild_id?: string;
channel_id?: string;
action_type?: string;
reason?: string;
executed_by?: string;
};
if (
!payload.message_id ||
!payload.user_id ||
!payload.guild_id ||
!payload.action_type
) {
return {
id: cmd.id,
success: false,
data: null,
error: "message_id, user_id, guild_id, and action_type are required",
};
}
const validActions = [
"delete_message",
"mute_user",
"warn_user",
"kick_user",
"ban_user",
] as const;
if (
!validActions.includes(
payload.action_type as (typeof validActions)[number],
)
) {
return {
id: cmd.id,
success: false,
data: null,
error: `Invalid action_type: ${payload.action_type}. Must be one of: ${validActions.join(", ")}`,
};
}
try {
// For delete_message, also actually delete via Discord if client is available
if (payload.action_type === "delete_message" && this.client) {
try {
const channelId = String(cmd.payload.channel_id ?? "");
if (channelId) {
const channel = await this.client.channels.fetch(channelId);
if (channel?.isText()) {
const msg = await channel.messages
.fetch(payload.message_id)
.catch(() => null);
if (msg) {
await msg.delete().catch((err: unknown) => {
this.logger.warn(
{ error: err, messageId: payload.message_id },
"Failed to delete message via Discord",
);
});
}
}
}
} catch (err) {
this.logger.warn(
{ error: err, messageId: payload.message_id },
"Failed to fetch channel/message for deletion",
);
}
}
const action = await createModerationAction({
message_id: payload.message_id,
user_id: payload.user_id,
guild_id: payload.guild_id,
action_type: payload.action_type as
| "delete_message"
| "mute_user"
| "warn_user"
| "kick_user"
| "ban_user",
reason: payload.reason ?? null,
executed_by: payload.executed_by ?? "command-handler",
status: "executed",
error: null,
executed_at: Date.now(),
});
this.logger.info(
{
actionId: action.id,
actionType: payload.action_type,
userId: payload.user_id,
},
"Moderation action executed",
);
return {
id: cmd.id,
success: true,
data: action,
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
this.logger.error(
{ error: message, commandId: cmd.id },
"Failed to execute moderation action",
);
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
}
@@ -0,0 +1,174 @@
import { type CommandMessage, type CommandReply } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
import Redis from "ioredis";
import { config } from "../../shared/config/config.js";
import { discordPlayer } from "../voice-recording/player.js";
import { voiceTransmitter } from "../voice-recording/transmitter.js";
import type { VoiceController } from "../voice-recording/voiceController.js";
// ---------------------------------------------------------------------------
// VoiceHandler
// ---------------------------------------------------------------------------
export class VoiceHandler {
private logger = createChildLogger("voice-handler");
constructor(
private client: Client | null,
private voiceController: VoiceController | null,
) {}
setClient(client: Client): void {
this.client = client;
}
setVoiceController(voiceController: VoiceController): void {
this.voiceController = voiceController;
}
async handleVoiceConnect(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!this.client || !this.voiceController) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
const channelId = String(cmd.payload.channelId ?? "");
if (!guildId || !channelId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId and channelId are required",
};
}
const status = await this.voiceController.connect(guildId, channelId);
return { id: cmd.id, success: true, data: status };
}
async handleVoiceDisconnect(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!this.voiceController) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const status = await this.voiceController.disconnect();
return { id: cmd.id, success: true, data: status };
}
async handleVoiceChannels(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
if (!guildId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId is required",
};
}
try {
const guild = await this.client.guilds.fetch(guildId);
const channels = await guild.channels.fetch();
const voiceChannels = channels
.filter((c) => c?.type === "GUILD_VOICE")
.map((c) => ({
id: c.id,
name: c.name,
type: "voice" as const,
}));
return { id: cmd.id, success: true, data: voiceChannels };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
async handleVoiceTransmitStart(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!discordPlayer.isConnected()) {
return {
id: cmd.id,
success: false,
data: null,
error: "Not connected to voice channel",
};
}
try {
// Create a new Redis connection for the transmitter
const transmitRedis = new Redis(config.REDIS_URL);
await voiceTransmitter.start(transmitRedis);
const status = voiceTransmitter.getStatus();
this.logger.info({ status }, "Voice transmit started");
return {
id: cmd.id,
success: true,
data: status,
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
this.logger.error({ error: message }, "Failed to start voice transmit");
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
async handleVoiceTransmitStop(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
try {
await voiceTransmitter.stop();
this.logger.info("Voice transmit stopped");
return {
id: cmd.id,
success: true,
data: { status: "stopped" },
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
this.logger.error({ error: message }, "Failed to stop voice transmit");
return {
id: cmd.id,
success: false,
data: null,
error: message,
};
}
}
}
@@ -1,4 +1,4 @@
import type { CustomLogger } from "@bete/shared/logger";
import { type CustomLogger, createChildLogger } from "@bete/shared/logger";
import Redis from "ioredis";
import { type DiscordGatewayEvent, EventChannels } from "./eventTypes.js";
@@ -37,12 +37,14 @@ export class RedisEventPublisher {
export class EventBroadcaster {
private publisher: RedisEventPublisher;
private logger = createChildLogger("event-broadcaster");
constructor(publisher: RedisEventPublisher) {
this.publisher = publisher;
}
async messageCreated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing message_created");
await this.publisher.publish(EventChannels.MESSAGE_CREATED, {
type: "message_created",
data,
@@ -52,6 +54,7 @@ export class EventBroadcaster {
}
async messageUpdated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing message_updated");
await this.publisher.publish(EventChannels.MESSAGE_UPDATED, {
type: "message_updated",
data,
@@ -61,6 +64,7 @@ export class EventBroadcaster {
}
async messageDeleted(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing message_deleted");
await this.publisher.publish(EventChannels.MESSAGE_DELETED, {
type: "message_deleted",
data,
@@ -70,6 +74,7 @@ export class EventBroadcaster {
}
async messageAnalyzed(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing message_analyzed");
await this.publisher.publish(EventChannels.MESSAGE_ANALYZED, {
type: "message_analyzed",
data,
@@ -79,6 +84,7 @@ export class EventBroadcaster {
}
async attachmentCreated(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing attachment_created");
await this.publisher.publish(EventChannels.ATTACHMENT_CREATED, {
type: "attachment_created",
data,
@@ -88,6 +94,7 @@ export class EventBroadcaster {
}
async attachmentUploaded(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing attachment_uploaded");
await this.publisher.publish(EventChannels.ATTACHMENT_UPLOADED, {
type: "attachment_uploaded",
data,
@@ -97,6 +104,7 @@ export class EventBroadcaster {
}
async voiceRecordingStarted(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing voice_recording_started");
await this.publisher.publish(EventChannels.VOICE_STARTED, {
type: "voice_recording_started",
data,
@@ -106,6 +114,7 @@ export class EventBroadcaster {
}
async voiceRecordingStopped(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing voice_recording_stopped");
await this.publisher.publish(EventChannels.VOICE_STOPPED, {
type: "voice_recording_stopped",
data,
@@ -115,6 +124,7 @@ export class EventBroadcaster {
}
async voiceRecordingUploaded(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing voice_recording_uploaded");
await this.publisher.publish(EventChannels.VOICE_UPLOADED, {
type: "voice_recording_uploaded",
data,
@@ -134,6 +144,10 @@ export class EventBroadcaster {
userId: string,
metadata?: Record<string, unknown>,
): Promise<void> {
this.logger.debug(
{ userId, pcmSize: pcmBuffer.length },
"Publishing voice_pcm_data",
);
await this.publisher.publish(EventChannels.VOICE_PCM, {
type: "voice_pcm_data",
data: {
@@ -155,6 +169,10 @@ export class EventBroadcaster {
userId: string,
data: { username: string; avatar: string; speaking: boolean },
): Promise<void> {
this.logger.debug(
{ userId, speaking: data.speaking },
"Publishing voice_active_user",
);
await this.publisher.publish(EventChannels.VOICE_ACTIVE_USER, {
type: "voice_active_user",
data: {
@@ -167,6 +185,7 @@ export class EventBroadcaster {
}
async analysisQueueStatus(data: unknown): Promise<void> {
this.logger.debug({ data }, "Publishing analysis_queue_status");
await this.publisher.publish(EventChannels.ANALYSIS_QUEUE_STATUS, {
type: "analysis_queue_status",
data,
@@ -176,6 +195,7 @@ export class EventBroadcaster {
}
async close(): Promise<void> {
this.logger.debug("Closing event broadcaster");
await this.publisher.close();
}
}
@@ -1,24 +1,35 @@
export interface DiscordGatewayEvent {
type: string;
data: unknown;
timestamp: number;
source: string;
}
import {
DISCORD_ANALYSIS_QUEUE_STATUS,
DISCORD_ATTACHMENT_CREATED,
DISCORD_ATTACHMENT_UPLOADED,
DISCORD_MESSAGE_ANALYZED,
DISCORD_MESSAGE_CREATED,
DISCORD_MESSAGE_DELETED,
DISCORD_MESSAGE_UPDATED,
DISCORD_VOICE_ACTIVE_USER,
DISCORD_VOICE_PCM,
DISCORD_VOICE_STARTED,
DISCORD_VOICE_STOPPED,
DISCORD_VOICE_UPLOADED,
type DiscordGatewayEvent,
} from "@bete/shared";
export type { DiscordGatewayEvent };
export const EventChannels = {
MESSAGE_CREATED: "discord:message:created",
MESSAGE_UPDATED: "discord:message:updated",
MESSAGE_DELETED: "discord:message:deleted",
MESSAGE_ANALYZED: "discord:message:analyzed",
ATTACHMENT_CREATED: "discord:attachment:created",
ATTACHMENT_UPLOADED: "discord:attachment:uploaded",
VOICE_STARTED: "discord:voice:started",
VOICE_STOPPED: "discord:voice:stopped",
VOICE_UPLOADED: "discord:voice:uploaded",
MESSAGE_CREATED: DISCORD_MESSAGE_CREATED,
MESSAGE_UPDATED: DISCORD_MESSAGE_UPDATED,
MESSAGE_DELETED: DISCORD_MESSAGE_DELETED,
MESSAGE_ANALYZED: DISCORD_MESSAGE_ANALYZED,
ATTACHMENT_CREATED: DISCORD_ATTACHMENT_CREATED,
ATTACHMENT_UPLOADED: DISCORD_ATTACHMENT_UPLOADED,
VOICE_STARTED: DISCORD_VOICE_STARTED,
VOICE_STOPPED: DISCORD_VOICE_STOPPED,
VOICE_UPLOADED: DISCORD_VOICE_UPLOADED,
// Real-time voice streaming channels
VOICE_ACTIVE_USER: "discord:voice:active_user", // Active speaker state updates
VOICE_PCM: "discord:voice:pcm", // Live PCM audio data stream
ANALYSIS_QUEUE_STATUS: "discord:analysis:queue_status",
VOICE_ACTIVE_USER: DISCORD_VOICE_ACTIVE_USER, // Active speaker state updates
VOICE_PCM: DISCORD_VOICE_PCM, // Live PCM audio data stream
ANALYSIS_QUEUE_STATUS: DISCORD_ANALYSIS_QUEUE_STATUS,
} as const;
export type EventChannelType =
@@ -0,0 +1,185 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, inArray, or, type SQL } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js";
import { attachmentsTable } from "../../shared/database/schema.js";
import type { AttachmentRecord } from "../message-capture/types.js";
// ─── AttachmentsDb Class ────────────────────────────────────────────────────
export class AttachmentsDb {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("attachments-db");
}
async insertAttachment(attachment: AttachmentRecord): Promise<void> {
this.logger.debug(
{ attachmentId: attachment.id },
"insertAttachment entry",
);
try {
await this.db
.insert(attachmentsTable)
.values(attachment)
.onConflictDoNothing();
} catch (error) {
this.logger.error(
{
attachmentId: attachment.id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to insert attachment",
);
throw error;
}
}
async getAttachmentsByChannel(
channelId: string,
limit: number = 50,
offset: number = 0,
guildId?: string,
): Promise<AttachmentRecord[]> {
this.logger.debug(
{ channelId, limit, offset },
"getAttachmentsByChannel entry",
);
try {
const conditions: SQL[] = [
or(
eq(attachmentsTable.channel_id, channelId),
eq(attachmentsTable.thread_id, channelId),
) as SQL,
];
if (guildId) {
conditions.push(eq(attachmentsTable.guild_id, guildId));
}
const rows = await this.db
.select()
.from(attachmentsTable)
.where(and(...conditions))
.orderBy(desc(attachmentsTable.created_at))
.limit(limit)
.offset(offset);
return rows as AttachmentRecord[];
} catch (error) {
this.logger.error(
{
channelId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get attachments by channel",
);
throw error;
}
}
async updateAttachmentAsUploaded(
attachmentId: string,
uploadedUrl: string,
uploadedAt: number,
): Promise<void> {
this.logger.debug({ attachmentId }, "updateAttachmentAsUploaded entry");
try {
await this.db
.update(attachmentsTable)
.set({
uploaded_url: uploadedUrl,
upload_status: "uploaded",
uploaded_at: uploadedAt,
})
.where(eq(attachmentsTable.id, attachmentId));
} catch (error) {
this.logger.error(
{
attachmentId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update attachment as uploaded",
);
throw error;
}
}
async updateAttachmentDiscordUrl(
attachmentId: string,
discordUrl: string,
): Promise<void> {
this.logger.debug({ attachmentId }, "updateAttachmentDiscordUrl entry");
try {
await this.db
.update(attachmentsTable)
.set({ discord_url: discordUrl })
.where(eq(attachmentsTable.id, attachmentId));
} catch (error) {
this.logger.error(
{
attachmentId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update attachment Discord URL",
);
throw error;
}
}
async updateAttachmentAsFailedUpload(
attachmentId: string,
error: string,
): Promise<void> {
this.logger.debug({ attachmentId }, "updateAttachmentAsFailedUpload entry");
try {
await this.db
.update(attachmentsTable)
.set({
upload_status: "failed",
upload_error: error,
})
.where(eq(attachmentsTable.id, attachmentId));
} catch (error) {
this.logger.error(
{
attachmentId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update attachment as failed",
);
throw error;
}
}
async getAttachmentsForMessages(
messageIds: string[],
): Promise<AttachmentRecord[]> {
this.logger.debug(
{ messageIdsCount: messageIds.length },
"getAttachmentsForMessages entry",
);
try {
if (messageIds.length === 0) return [];
const rows = await this.db
.select()
.from(attachmentsTable)
.where(inArray(attachmentsTable.message_id, messageIds));
return rows as AttachmentRecord[];
} catch (error) {
this.logger.error(
{
messageIds,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get attachments for messages",
);
throw error;
}
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,826 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import {
and,
asc,
desc,
eq,
inArray,
isNull,
or,
type SQL,
sql,
} from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js";
import { messagesTable } from "../../shared/database/schema.js";
import { decodeCursor, encodeCursor } from "../message-capture/pagination.js";
import type {
MessageQuery,
MessageRecord,
PageResult,
} from "../message-capture/types.js";
// ─── Helpers ────────────────────────────────────────────────────────────────
function channelOrThreadCondition(channelId: string): SQL {
return or(
eq(messagesTable.channel_id, channelId),
eq(messagesTable.thread_id, channelId),
) as SQL;
}
function buildListMessageConditions(query: MessageQuery): SQL[] {
const conditions: SQL[] = [];
if (query.guildId) {
conditions.push(eq(messagesTable.guild_id, query.guildId));
}
if (query.channelId) {
conditions.push(channelOrThreadCondition(query.channelId));
}
if (query.threadId) {
conditions.push(eq(messagesTable.thread_id, query.threadId));
}
if (query.userId) {
conditions.push(eq(messagesTable.user_id, query.userId));
}
if (query.status && query.status.length > 0) {
conditions.push(sql`${messagesTable.ai_status} in ${query.status}`);
}
if (query.q) {
const pattern = `%${query.q.toLowerCase()}%`;
conditions.push(sql`lower(${messagesTable.content}) like ${pattern}`);
}
const cursorData = decodeCursor(query.cursor);
if (cursorData) {
conditions.push(
sql`(${messagesTable.created_at} < ${cursorData.created_at} or (${messagesTable.created_at} = ${cursorData.created_at} and ${messagesTable.id} < ${cursorData.id}))`,
);
}
return conditions;
}
function pageRows<T extends { created_at: number; id: string }>(
rows: unknown[],
limit: number,
): PageResult<T> {
const hasMore = rows.length > limit;
const data = rows.slice(0, limit) as T[];
const lastItem = data[data.length - 1];
const nextCursor =
hasMore && lastItem
? encodeCursor({ created_at: lastItem.created_at, id: lastItem.id })
: null;
return { data, nextCursor };
}
function pageMessages(
rows: unknown[],
limit: number,
): PageResult<MessageRecord> {
return pageRows<MessageRecord>(rows, limit);
}
function stringifyAIList(
value: string[] | string | null | undefined,
): string | null {
if (value == null) return null;
return Array.isArray(value) ? JSON.stringify(value) : value;
}
// ─── AIAnalysisUpdate interface ────────────────────────────────────────────
export interface AIAnalysisUpdate {
status: "pending" | "processing" | "clean" | "warn" | "flagged" | "error";
flags?: string | null;
score?: number | null;
analysis?: string | null;
categories?: string[] | string | null;
severity?: MessageRecord["ai_severity"] | null;
confidence?: number | null;
recommendedAction?: MessageRecord["ai_recommended_action"] | null;
analyzedAt?: number | null;
error?: string | null;
}
// ─── MessagesDb Class ──────────────────────────────────────────────────────
export class MessagesDb {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-db");
}
// ── CRUD ──────────────────────────────────────────────────────────────
async insertMessage(message: MessageRecord): Promise<void> {
this.logger.debug({ messageId: message.id }, "insertMessage entry");
try {
await this.db
.insert(messagesTable)
.values(message as any)
.onConflictDoNothing();
} catch (error) {
this.logger.error(
{
messageId: message.id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to insert message",
);
throw error;
}
}
async upsertMessageForCapture(message: MessageRecord): Promise<boolean> {
this.logger.debug(
{ messageId: message.id },
"upsertMessageForCapture entry",
);
try {
const messageWithAIStatus = {
...message,
ai_status: "pending" as const,
};
const rows = await this.db
.insert(messagesTable)
.values(messageWithAIStatus as any)
.onConflictDoNothing()
.returning({ id: messagesTable.id });
return rows.length > 0;
} catch (error) {
this.logger.error(
{
messageId: message.id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to upsert message for capture",
);
throw error;
}
}
async updateMessageAsEdited(
messageId: string,
editedContent: string,
editedAt: number,
): Promise<void> {
this.logger.debug({ messageId }, "updateMessageAsEdited entry");
try {
await this.db
.update(messagesTable)
.set({
edited_content: editedContent,
edited_at: editedAt,
type: "edited",
ai_status: "pending",
ai_moderation_flags: null,
ai_moderation_score: null,
ai_analysis: null,
ai_categories: null,
ai_severity: null,
ai_confidence: null,
ai_recommended_action: null,
ai_analyzed_at: null,
ai_error: null,
})
.where(eq(messagesTable.id, messageId));
} catch (error) {
this.logger.error(
{
messageId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update message as edited",
);
throw error;
}
}
async updateMessageAsDeleted(
messageId: string,
deletedAt: number,
): Promise<void> {
this.logger.debug({ messageId }, "updateMessageAsDeleted entry");
try {
await this.db
.update(messagesTable)
.set({
deleted_at: deletedAt,
type: "deleted",
})
.where(eq(messagesTable.id, messageId));
} catch (error) {
this.logger.error(
{
messageId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update message as deleted",
);
throw error;
}
}
async getMessagesByChannel(
channelId: string,
limit: number = 50,
offset: number = 0,
guildId?: string,
): Promise<MessageRecord[]> {
this.logger.debug(
{ channelId, limit, offset, guildId },
"getMessagesByChannel entry",
);
try {
const conditions: SQL[] = [
or(
eq(messagesTable.channel_id, channelId),
eq(messagesTable.thread_id, channelId),
) as SQL,
];
if (guildId) {
conditions.push(eq(messagesTable.guild_id, guildId));
}
const rows = await this.db
.select()
.from(messagesTable)
.where(and(...conditions))
.orderBy(desc(messagesTable.created_at), desc(messagesTable.id))
.limit(limit)
.offset(offset);
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
channelId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get messages by channel",
);
throw error;
}
}
async getMessageById(messageId: string): Promise<MessageRecord | null> {
this.logger.debug({ messageId }, "getMessageById entry");
try {
const rows = await this.db
.select()
.from(messagesTable)
.where(eq(messagesTable.id, messageId));
return (rows[0] as MessageRecord) ?? null;
} catch (error) {
this.logger.error(
{
messageId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get message by id",
);
throw error;
}
}
// ── AI Analysis ───────────────────────────────────────────────────────
async updateMessageAIAnalysis(
messageId: string,
result: AIAnalysisUpdate,
): Promise<MessageRecord | null> {
this.logger.debug({ messageId }, "updateMessageAIAnalysis entry");
try {
await this.db
.update(messagesTable)
.set({
ai_status: result.status,
ai_moderation_flags: result.flags ?? null,
ai_moderation_score: result.score ?? null,
ai_analysis: result.analysis ?? null,
ai_categories: stringifyAIList(result.categories),
ai_severity: result.severity ?? null,
ai_confidence: result.confidence ?? result.score ?? null,
ai_recommended_action: result.recommendedAction ?? null,
ai_analyzed_at: result.analyzedAt ?? Date.now(),
ai_error: result.error ?? null,
})
.where(eq(messagesTable.id, messageId));
const rows = await this.db
.select()
.from(messagesTable)
.where(eq(messagesTable.id, messageId));
return (rows[0] as MessageRecord) ?? null;
} catch (error) {
this.logger.error(
{
messageId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update message AI analysis",
);
throw error;
}
}
async updateMessagesAIAnalysisBulk(
updates: Array<{ messageId: string; result: AIAnalysisUpdate }>,
): Promise<MessageRecord[]> {
this.logger.debug(
{ count: updates.length },
"updateMessagesAIAnalysisBulk entry",
);
if (updates.length === 0) return [];
try {
const now = Date.now();
await this.db.transaction(async (tx) => {
for (const { messageId, result } of updates) {
await tx
.update(messagesTable)
.set({
ai_status: result.status,
ai_moderation_flags: result.flags ?? null,
ai_moderation_score: result.score ?? null,
ai_analysis: result.analysis ?? null,
ai_categories: stringifyAIList(result.categories),
ai_severity: result.severity ?? null,
ai_confidence: result.confidence ?? result.score ?? null,
ai_recommended_action: result.recommendedAction ?? null,
ai_analyzed_at: result.analyzedAt ?? now,
ai_error: result.error ?? null,
})
.where(eq(messagesTable.id, messageId));
}
});
const ids = updates.map(({ messageId }) => messageId);
const rows = await this.db
.select()
.from(messagesTable)
.where(inArray(messagesTable.id, ids));
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
error: error instanceof Error ? error.message : String(error),
},
"Failed to bulk update messages AI analysis",
);
throw error;
}
}
async getPendingAIAnalysisMessages(
limit: number = 25,
): Promise<MessageRecord[]> {
this.logger.debug({ limit }, "getPendingAIAnalysisMessages entry");
try {
const rows = await this.db
.select()
.from(messagesTable)
.where(
and(
eq(messagesTable.ai_status, "pending"),
isNull(messagesTable.deleted_at),
),
)
.orderBy(asc(messagesTable.created_at))
.limit(limit);
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to get pending AI analysis messages",
);
throw error;
}
}
// ── Listing / Pagination ──────────────────────────────────────────────
async listMessages(query: MessageQuery): Promise<PageResult<MessageRecord>> {
this.logger.debug({ query }, "listMessages entry");
try {
const conditions = buildListMessageConditions(query);
const rows = await this.db
.select()
.from(messagesTable)
.where(conditions.length > 0 ? and(...conditions) : undefined)
.orderBy(desc(messagesTable.created_at), desc(messagesTable.id))
.limit(query.limit + 1);
return pageMessages(rows, query.limit);
} catch (error) {
this.logger.error(
{
query,
error: error instanceof Error ? error.message : String(error),
},
"Failed to list messages",
);
throw error;
}
}
async listReviewMessages(
query: Omit<MessageQuery, "status">,
): Promise<PageResult<MessageRecord>> {
return this.listMessages({
...query,
status: ["warn", "flagged", "error"],
});
}
// ── Conversation Context ──────────────────────────────────────────────
async getConversationContextBefore(input: {
channelId: string;
threadId: string | null;
beforeCreatedAt: number;
limit: number;
}): Promise<MessageRecord[]> {
this.logger.debug(
{ channelId: input.channelId, threadId: input.threadId },
"getConversationContextBefore entry",
);
try {
const { channelId, threadId, beforeCreatedAt, limit } = input;
const locationCondition = threadId
? eq(messagesTable.thread_id, threadId)
: eq(messagesTable.channel_id, channelId);
const rows = await this.db
.select()
.from(messagesTable)
.where(
and(
locationCondition,
sql`${messagesTable.created_at} < ${beforeCreatedAt}`,
isNull(messagesTable.deleted_at),
),
)
.orderBy(desc(messagesTable.created_at))
.limit(limit);
return (rows as MessageRecord[]).reverse();
} catch (error) {
this.logger.error(
{
channelId: input.channelId,
threadId: input.threadId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get conversation context before",
);
throw error;
}
}
async getPendingMessagesByConversation(
conversationKey: string,
limit: number = 200,
): Promise<MessageRecord[]> {
this.logger.debug(
{ conversationKey, limit },
"getPendingMessagesByConversation entry",
);
try {
const rows = await this.db.transaction(async (tx) => {
const pendingIdsQuery = tx
.select({ id: messagesTable.id })
.from(messagesTable)
.where(
and(
or(
eq(messagesTable.thread_id, conversationKey),
eq(messagesTable.channel_id, conversationKey),
),
eq(messagesTable.ai_status, "pending"),
isNull(messagesTable.deleted_at),
),
)
.orderBy(asc(messagesTable.created_at))
.limit(limit)
.for("update", { skipLocked: true });
const pendingIds = (await pendingIdsQuery) as Array<{ id: string }>;
if (pendingIds.length === 0) return [];
return await tx
.update(messagesTable)
.set({ ai_status: "processing", ai_analyzed_at: Date.now() })
.where(
inArray(
messagesTable.id,
pendingIds.map((r) => r.id),
),
)
.returning();
});
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
conversationKey,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get pending messages by conversation",
);
throw error;
}
}
// ── Conversation Keys ─────────────────────────────────────────────────
async getPendingConversationKeys(limit: number = 500): Promise<string[]> {
this.logger.debug({ limit }, "getPendingConversationKeys entry");
try {
const rows = (await this.db
.selectDistinct({
thread_id: messagesTable.thread_id,
channel_id: messagesTable.channel_id,
})
.from(messagesTable)
.where(
and(
eq(messagesTable.ai_status, "pending"),
isNull(messagesTable.deleted_at),
),
)
.limit(limit)) as Array<{
thread_id: string | null;
channel_id: string;
}>;
const keys: string[] = [];
for (const row of rows) {
const key = row.thread_id || row.channel_id;
if (key && !keys.includes(key)) {
keys.push(key);
}
}
return keys;
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to get pending conversation keys",
);
throw error;
}
}
async getConversationKeysWithIncompleteAnalysis(
limit: number = 200,
): Promise<string[]> {
this.logger.debug(
{ limit },
"getConversationKeysWithIncompleteAnalysis entry",
);
try {
const rows = (await this.db
.selectDistinct({
thread_id: messagesTable.thread_id,
channel_id: messagesTable.channel_id,
})
.from(messagesTable)
.where(
and(
eq(messagesTable.ai_status, "error"),
sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`,
sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`,
isNull(messagesTable.deleted_at),
),
)
.limit(limit)) as Array<{
thread_id: string | null;
channel_id: string;
}>;
const keys: string[] = [];
for (const row of rows) {
const key = row.thread_id || row.channel_id;
if (key && !keys.includes(key)) {
keys.push(key);
}
}
return keys;
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to get conversation keys with incomplete analysis",
);
throw error;
}
}
async getIncompleteMessagesByConversation(
conversationKey: string,
limit: number = 500,
): Promise<MessageRecord[]> {
this.logger.debug(
{ conversationKey, limit },
"getIncompleteMessagesByConversation entry",
);
try {
const rows = await this.db.transaction(async (tx) => {
const pendingIdsQuery = tx
.select({ id: messagesTable.id })
.from(messagesTable)
.where(
and(
or(
eq(messagesTable.thread_id, conversationKey),
eq(messagesTable.channel_id, conversationKey),
),
eq(messagesTable.ai_status, "error"),
sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`,
sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`,
isNull(messagesTable.deleted_at),
),
)
.orderBy(asc(messagesTable.created_at))
.limit(limit)
.for("update", { skipLocked: true });
const pendingIds = (await pendingIdsQuery) as Array<{ id: string }>;
if (pendingIds.length === 0) return [];
return await tx
.update(messagesTable)
.set({ ai_status: "processing", ai_analyzed_at: Date.now() })
.where(
inArray(
messagesTable.id,
pendingIds.map((r) => r.id),
),
)
.returning();
});
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
conversationKey,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get incomplete messages by conversation",
);
throw error;
}
}
// ── Search ────────────────────────────────────────────────────────────
async searchMessages(input: {
query: string;
channelId?: string;
guildId?: string;
limit?: number;
}): Promise<MessageRecord[]> {
this.logger.debug({ query: input.query }, "searchMessages entry");
try {
const { query, channelId, guildId, limit = 20 } = input;
const searchPattern = `%${query}%`;
const conditions: (SQL | undefined)[] = [
isNull(messagesTable.deleted_at),
];
if (guildId) {
conditions.push(eq(messagesTable.guild_id, guildId));
}
if (channelId) {
conditions.push(channelOrThreadCondition(channelId));
}
conditions.push(
or(
sql`${messagesTable.content} LIKE ${searchPattern}`,
sql`${messagesTable.edited_content} LIKE ${searchPattern}`,
),
);
const validConditions = conditions.filter(
(c): c is SQL => c !== undefined,
);
const rows = await this.db
.select()
.from(messagesTable)
.where(and(...validConditions))
.orderBy(desc(messagesTable.created_at))
.limit(limit);
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
query: input.query,
channelId: input.channelId,
guildId: input.guildId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to search messages",
);
throw error;
}
}
// ── Retention / Recovery ──────────────────────────────────────────────
async getExpiredMessages(retentionDays: number): Promise<MessageRecord[]> {
this.logger.debug({ retentionDays }, "getExpiredMessages entry");
try {
const cutoffTime = Date.now() - retentionDays * 24 * 60 * 60 * 1000;
const rows = await this.db
.select()
.from(messagesTable)
.where(
and(
sql`${messagesTable.created_at} < ${cutoffTime}`,
isNull(messagesTable.deleted_at),
),
)
.limit(1000);
return rows as MessageRecord[];
} catch (error) {
this.logger.error(
{
retentionDays,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get expired messages",
);
throw error;
}
}
async revertStuckProcessingMessages(
timeoutMs: number = 300000,
): Promise<number> {
this.logger.debug({ timeoutMs }, "revertStuckProcessingMessages entry");
try {
const cutoffTime = Date.now() - timeoutMs;
const rows = await this.db
.update(messagesTable)
.set({ ai_status: "pending", ai_analyzed_at: null })
.where(
and(
eq(messagesTable.ai_status, "processing"),
sql`${messagesTable.ai_analyzed_at} < ${cutoffTime}`,
),
)
.returning({ id: messagesTable.id });
if (Array.isArray(rows) && rows.length > 0) {
this.logger.info(
{
count: rows.length,
messageIds: rows.map((r: { id: string }) => r.id),
},
"Reverted stuck processing messages back to pending",
);
}
return Array.isArray(rows) ? rows.length : 0;
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to revert stuck processing messages",
);
return 0;
}
}
}
@@ -0,0 +1,163 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, type SQL, sql } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js";
import { moderationActionsTable } from "../../shared/database/schema.js";
import { decodeCursor, encodeCursor } from "../message-capture/pagination.js";
import type { ModerationAction, PageResult } from "../message-capture/types.js";
// ─── Helpers ────────────────────────────────────────────────────────────────
function pageRows<T extends { created_at: number; id: string }>(
rows: unknown[],
limit: number,
): PageResult<T> {
const hasMore = rows.length > limit;
const data = rows.slice(0, limit) as T[];
const lastItem = data[data.length - 1];
const nextCursor =
hasMore && lastItem
? encodeCursor({ created_at: lastItem.created_at, id: lastItem.id })
: null;
return { data, nextCursor };
}
// ─── ModerationActionsDb Class ──────────────────────────────────────────────
export class ModerationActionsDb {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("moderation-actions-db");
}
async createModerationAction(
action: Omit<ModerationAction, "id" | "created_at">,
): Promise<ModerationAction> {
this.logger.debug(
{ guildId: action.guild_id },
"createModerationAction entry",
);
try {
const id = `action-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
const created_at = Date.now();
const rows = await this.db
.insert(moderationActionsTable)
.values({
...action,
id,
created_at,
})
.returning();
return rows[0] as ModerationAction;
} catch (error) {
this.logger.error(
{
guildId: action.guild_id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to create moderation action",
);
throw error;
}
}
async getModerationAction(id: string): Promise<ModerationAction | null> {
this.logger.debug({ actionId: id }, "getModerationAction entry");
try {
const rows = await this.db
.select()
.from(moderationActionsTable)
.where(eq(moderationActionsTable.id, id));
return (rows[0] as ModerationAction) || null;
} catch (error) {
this.logger.error(
{
actionId: id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get moderation action",
);
throw error;
}
}
async listModerationActions(query: {
guildId?: string;
status?: string[];
cursor?: string;
limit: number;
}): Promise<PageResult<ModerationAction>> {
this.logger.debug({ query }, "listModerationActions entry");
try {
const limit = Math.max(1, Math.min(query.limit || 50, 100));
const conditions: SQL[] = [];
if (query.guildId) {
conditions.push(eq(moderationActionsTable.guild_id, query.guildId));
}
if (query.status && query.status.length > 0) {
conditions.push(
sql`${moderationActionsTable.status} in ${query.status}`,
);
}
const cursorData = decodeCursor(query.cursor);
if (cursorData) {
conditions.push(
sql`(${moderationActionsTable.created_at} < ${cursorData.created_at} or (${moderationActionsTable.created_at} = ${cursorData.created_at} and ${moderationActionsTable.id} < ${cursorData.id}))`,
);
}
const rows = await this.db
.select()
.from(moderationActionsTable)
.where(conditions.length > 0 ? and(...conditions) : undefined)
.orderBy(
desc(moderationActionsTable.created_at),
desc(moderationActionsTable.id),
)
.limit(limit + 1);
return pageRows<ModerationAction>(rows, limit);
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to list moderation actions",
);
throw error;
}
}
async updateModerationAction(
id: string,
updates: Partial<Omit<ModerationAction, "id" | "created_at">>,
): Promise<ModerationAction | null> {
this.logger.debug({ actionId: id }, "updateModerationAction entry");
try {
const rows = (await this.db
.update(moderationActionsTable)
.set(updates)
.where(eq(moderationActionsTable.id, id))
.returning()) as ModerationAction[];
return rows[0] || null;
} catch (error) {
this.logger.error(
{
actionId: id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update moderation action",
);
throw error;
}
}
}
@@ -0,0 +1,88 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { eq } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js";
import { retentionPoliciesTable } from "../../shared/database/schema.js";
import type { RetentionPolicy } from "../message-capture/types.js";
// ─── RetentionDb Class ──────────────────────────────────────────────────────
export class RetentionDb {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("retention-db");
}
async getRetentionPolicy(guildId: string): Promise<RetentionPolicy | null> {
this.logger.debug({ guildId }, "getRetentionPolicy entry");
try {
const rows = await this.db
.select()
.from(retentionPoliciesTable)
.where(eq(retentionPoliciesTable.guild_id, guildId));
return (rows[0] as RetentionPolicy) || null;
} catch (error) {
this.logger.error(
{
guildId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get retention policy",
);
throw error;
}
}
async upsertRetentionPolicy(
policy: Omit<RetentionPolicy, "created_at" | "updated_at">,
): Promise<RetentionPolicy> {
this.logger.debug(
{ guildId: policy.guild_id },
"upsertRetentionPolicy entry",
);
try {
const now = Date.now();
const existing = await this.getRetentionPolicy(policy.guild_id);
if (existing) {
const rows = (await this.db
.update(retentionPoliciesTable)
.set({
...policy,
updated_at: now,
})
.where(eq(retentionPoliciesTable.id, existing.id))
.returning()) as RetentionPolicy[];
return rows[0] as RetentionPolicy;
}
const id = `policy-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
const rows = (await this.db
.insert(retentionPoliciesTable)
.values({
...policy,
id,
created_at: now,
updated_at: now,
})
.returning()) as RetentionPolicy[];
return rows[0] as RetentionPolicy;
} catch (error) {
this.logger.error(
{
guildId: policy.guild_id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to upsert retention policy",
);
throw error;
}
}
}
@@ -0,0 +1,165 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, type SQL, sql } from "drizzle-orm";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import type * as schema from "../../shared/database/schema.js";
import { messageReviewsTable } from "../../shared/database/schema.js";
import { decodeCursor, encodeCursor } from "../message-capture/pagination.js";
import type { MessageReview, PageResult } from "../message-capture/types.js";
// ─── Helpers ────────────────────────────────────────────────────────────────
function pageRows<T extends { created_at: number; id: string }>(
rows: unknown[],
limit: number,
): PageResult<T> {
const hasMore = rows.length > limit;
const data = rows.slice(0, limit) as T[];
const lastItem = data[data.length - 1];
const nextCursor =
hasMore && lastItem
? encodeCursor({ created_at: lastItem.created_at, id: lastItem.id })
: null;
return { data, nextCursor };
}
// ─── ReviewsDb Class ────────────────────────────────────────────────────────
export class ReviewsDb {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("reviews-db");
}
async createMessageReview(
review: Omit<MessageReview, "id" | "created_at">,
): Promise<MessageReview> {
this.logger.debug(
{ messageId: review.message_id },
"createMessageReview entry",
);
try {
const id = `review-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
const created_at = Date.now();
const rows = await this.db
.insert(messageReviewsTable)
.values({
...review,
id,
created_at,
})
.returning();
return rows[0] as MessageReview;
} catch (error) {
this.logger.error(
{
messageId: review.message_id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to create message review",
);
throw error;
}
}
async getMessageReview(id: string): Promise<MessageReview | null> {
this.logger.debug({ reviewId: id }, "getMessageReview entry");
try {
const rows = await this.db
.select()
.from(messageReviewsTable)
.where(eq(messageReviewsTable.id, id));
return (rows[0] as MessageReview) || null;
} catch (error) {
this.logger.error(
{
reviewId: id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to get message review",
);
throw error;
}
}
async listMessageReviews(query: {
guildId?: string;
channelId?: string;
status?: string[];
cursor?: string;
limit: number;
}): Promise<PageResult<MessageReview>> {
this.logger.debug({ query }, "listMessageReviews entry");
try {
const limit = Math.max(1, Math.min(query.limit || 50, 100));
const conditions: SQL[] = [];
if (query.guildId) {
conditions.push(eq(messageReviewsTable.guild_id, query.guildId));
}
if (query.channelId) {
conditions.push(eq(messageReviewsTable.channel_id, query.channelId));
}
if (query.status && query.status.length > 0) {
conditions.push(sql`${messageReviewsTable.status} in ${query.status}`);
}
const cursorData = decodeCursor(query.cursor);
if (cursorData) {
conditions.push(
sql`(${messageReviewsTable.created_at} < ${cursorData.created_at} or (${messageReviewsTable.created_at} = ${cursorData.created_at} and ${messageReviewsTable.id} < ${cursorData.id}))`,
);
}
const rows = await this.db
.select()
.from(messageReviewsTable)
.where(conditions.length > 0 ? and(...conditions) : undefined)
.orderBy(
desc(messageReviewsTable.created_at),
desc(messageReviewsTable.id),
)
.limit(limit + 1);
return pageRows<MessageReview>(rows, limit);
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
"Failed to list message reviews",
);
throw error;
}
}
async updateMessageReview(
id: string,
updates: Partial<Omit<MessageReview, "id" | "created_at">>,
): Promise<MessageReview | null> {
this.logger.debug({ reviewId: id }, "updateMessageReview entry");
try {
const rows = (await this.db
.update(messageReviewsTable)
.set(updates)
.where(eq(messageReviewsTable.id, id))
.returning()) as MessageReview[];
return rows[0] || null;
} catch (error) {
this.logger.error(
{
reviewId: id,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update message review",
);
throw error;
}
}
}
@@ -1,10 +1,8 @@
import { promises as fsPromises } from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import { retryWithBackoff } from "@bete/shared/utils";
import {
type DiscordGatewayAdapterCreator,
EndBehaviorType,
entersState,
getVoiceConnection,
joinVoiceChannel,
@@ -14,19 +12,12 @@ import {
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import { config } from "../../shared/config/config.js";
import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js";
import { PacketFilter } from "./packetFilter.js";
import { OpusDecoder } from "./recorder/decoder.js";
import {
collectUserMetadata,
createSegmentMetadata,
} from "./recorder/metadata.js";
import { SegmentManager } from "./recorder/segment.js";
import {
createRecordingSession,
finalizeRecordingSession,
type RecordingSession,
} from "./recorder/sessionRecording.js";
import { uploadRecordingSegment } from "./recorder/uploader.js";
import { createSpeakingHandler } from "./recorder/speakingHandler.js";
const logger = createChildLogger("recorder");
@@ -132,174 +123,18 @@ export async function startRecording(
const receiver = connection.receiver;
// Dengarkan siapapun yang mulai bicara
receiver.speaking.on("start", async (userId) => {
if (userId === client.user?.id) return;
const userMetadata = await collectUserMetadata(client, userId, channel);
if (userMetadata.bot) return;
logger.debug(
{ userId, username: userMetadata.username },
"Voice activity detected",
);
// Notify webserver
_eventBroadcaster?.voiceActiveUser(userId, {
username: userMetadata.username,
avatar: userMetadata.avatarUrl,
speaking: true,
});
// Skip if user already has an active stream
if (receiver.subscriptions.has(userId)) return;
const userDir = path.join(recordingsDir, userId);
await fsPromises.mkdir(userDir, { recursive: true }).catch(() => {
// Directory already exists, ignore
});
try {
// Subscribe to the audio stream FIRST, then immediately attach all event
// handlers before piping — prevents race condition where initial packets
// arrive before listeners are registered.
const audioStream = receiver.subscribe(userId, {
end: {
behavior: EndBehaviorType.AfterSilence,
duration: config.AUDIO_STREAM_SILENCE_DURATION_MS,
},
});
const packetFilterForOgg = new PacketFilter(
config.PACKET_FILTER_MIN_SIZE,
);
const segmentManager = new SegmentManager(
userDir,
config.RECORDING_SEGMENT_MS,
);
// --- Web broadcast: prism decoder with safe restart and cooldown ---
const decoder = new OpusDecoder({
cooldownMs: config.DECODER_COOLDOWN_MS,
rotateMs: config.DECODER_ROTATE_MS,
onData: (pcm) => {
// Downsample 48kHz stereo → 24kHz mono (left channel, every 2nd sample)
const outBuf = Buffer.alloc(pcm.length / 4);
for (let i = 0; i < outBuf.length / 2; i++) {
outBuf.writeInt16LE(pcm.readInt16LE(i * 8), i * 2);
}
_eventBroadcaster?.voicePcmData(outBuf, userId);
},
});
// Attach all audioStream event handlers BEFORE pipe() to avoid data loss
audioStream.on("data", (chunk: Buffer) => {
if (chunk.length < 8) return;
segmentManager.rotateIfNeeded(packetFilterForOgg);
decoder.rotateIfNeeded();
decoder.write(chunk);
});
audioStream.on("end", () => {
segmentManager.close(packetFilterForOgg);
decoder.destroy();
_eventBroadcaster?.voiceActiveUser(userId, {
username: userMetadata.username,
avatar: userMetadata.avatarUrl,
speaking: false,
});
});
audioStream.on("error", (error: Error) => {
segmentManager.close(packetFilterForOgg);
decoder.destroy();
logger.error({ userId, error: error.message }, "Audio stream error");
});
// Now pipe for OGG recording (safe — event handlers already attached)
const oggPacketStream = audioStream.pipe(packetFilterForOgg);
const activeSession = activeSessions.get(channel.guild.id);
let currentSegment = segmentManager.open(oggPacketStream);
currentSegment.out.on("finish", () => {
if (config.VERBOSE) {
logger.info({ filename: currentSegment.filename }, "Segment saved");
}
const endTime = currentSegment.endTime ?? Date.now();
if (activeSession) {
activeSession.registerSegment({
user: userMetadata,
oggPath: currentSegment.filename,
jsonPath: currentSegment.jsonFilename,
startTime: currentSegment.startTime,
endTime,
});
}
const metadata = createSegmentMetadata(
userMetadata,
currentSegment,
activeSession?.sessionId ?? `${userId}-0`,
activeSession?.sessionId ?? `${channel.guild.id}-${channel.id}-0`,
activeSession?.startTime ?? 0,
config.RECORDING_SEGMENT_MS,
);
fsPromises
.writeFile(
currentSegment.jsonFilename,
JSON.stringify(metadata, null, 2),
)
.then(() => {
if (config.VERBOSE) {
logger.info(
{ jsonFile: currentSegment.jsonFilename },
"Metadata saved",
);
}
})
.catch((err: unknown) => {
logger.error(
{ error: err instanceof Error ? err.message : String(err) },
"Failed to write segment metadata",
);
});
// Trigger async voice segment upload
const segmentId = `${userId}-${currentSegment.startTime}`;
uploadRecordingSegment({
id: segmentId,
oggPath: currentSegment.filename,
userId: userMetadata.userId,
username: userMetadata.username,
avatarUrl: userMetadata.avatarUrl,
guildId: channel.guild.id,
channelId: channel.id,
channelName: channel.name,
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error(
{ segmentId, error: msg },
"Upload segment trigger failed",
);
});
});
currentSegment.out.on("error", (err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error({ userId, error: msg }, "File write error");
});
packetFilterForOgg.on("error", (err) => {
segmentManager.close(oggPacketStream);
logger.error({ userId, error: err.message }, "PacketFilter error");
});
} catch (e) {
logger.error(
{ userId, error: e instanceof Error ? e.message : String(e) },
"Failed to create stream",
);
}
// Use the extracted speaking handler for voice activity
const speakingHandler = createSpeakingHandler({
client,
channel,
receiver,
eventBroadcaster: _eventBroadcaster,
activeSessions,
recordingsDir,
});
receiver.speaking.on("start", speakingHandler);
// Handle unexpected disconnection
connection.on(VoiceConnectionStatus.Disconnected, async () => {
if (config.VERBOSE) {
@@ -0,0 +1,102 @@
import { promises as fsPromises } from "node:fs";
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../../shared/config/config.js";
import type {
SegmentState,
UserMetadata,
} from "../../message-capture/types.js";
import { createSegmentMetadata } from "./metadata.js";
import type { RecordingSession } from "./sessionRecording.js";
import { uploadRecordingSegment } from "./uploader.js";
const logger = createChildLogger("segment-finalizer");
export interface SegmentFinalizerInput {
currentSegment: SegmentState;
userMetadata: UserMetadata;
activeSession: RecordingSession | undefined;
guildId: string;
channelId: string;
channelName: string;
}
/**
* Handles the completion of an OGG segment:
* - Logs the saved segment (if VERBOSE)
* - Registers the segment with the active recording session
* - Writes the metadata JSON file alongside the OGG file
* - Triggers async upload of the segment to external storage
*
* This function is fire-and-forget for the metadata write and upload;
* errors are caught and logged without throwing.
*/
export function finalizeSegment(input: SegmentFinalizerInput): void {
const {
currentSegment,
userMetadata,
activeSession,
guildId,
channelId,
channelName,
} = input;
const endTime = currentSegment.endTime ?? Date.now();
if (config.VERBOSE) {
logger.info({ filename: currentSegment.filename }, "Segment saved");
}
// Register segment with the active recording session
if (activeSession) {
activeSession.registerSegment({
user: userMetadata,
oggPath: currentSegment.filename,
jsonPath: currentSegment.jsonFilename,
startTime: currentSegment.startTime,
endTime,
});
}
// Write metadata JSON (async, fire-and-forget)
const metadata = createSegmentMetadata(
userMetadata,
currentSegment,
activeSession?.sessionId ?? `${userMetadata.userId}-0`,
activeSession?.sessionId ?? `${guildId}-${channelId}-0`,
activeSession?.startTime ?? 0,
config.RECORDING_SEGMENT_MS,
);
fsPromises
.writeFile(currentSegment.jsonFilename, JSON.stringify(metadata, null, 2))
.then(() => {
if (config.VERBOSE) {
logger.info(
{ jsonFile: currentSegment.jsonFilename },
"Metadata saved",
);
}
})
.catch((err: unknown) => {
logger.error(
{ error: err instanceof Error ? err.message : String(err) },
"Failed to write segment metadata",
);
});
// Trigger async voice segment upload (fire-and-forget)
const segmentId = `${userMetadata.userId}-${currentSegment.startTime}`;
uploadRecordingSegment({
id: segmentId,
oggPath: currentSegment.filename,
userId: userMetadata.userId,
username: userMetadata.username,
avatarUrl: userMetadata.avatarUrl,
guildId,
channelId,
channelName,
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error({ segmentId, error: msg }, "Upload segment trigger failed");
});
}
@@ -0,0 +1,148 @@
import { promises as fsPromises } from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { VoiceConnection } from "@discordjs/voice";
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import type { EventBroadcaster } from "../../event-broadcaster/eventBroadcaster.js";
import { collectUserMetadata } from "./metadata.js";
import { finalizeSegment } from "./segmentFinalizer.js";
import type { RecordingSession } from "./sessionRecording.js";
import { setupUserStream } from "./streamSetup.js";
const logger = createChildLogger("speaking-handler");
export interface SpeakingHandlerContext {
client: Client;
channel: VoiceChannel;
receiver: VoiceConnection["receiver"];
eventBroadcaster: EventBroadcaster | undefined;
activeSessions: Map<string, RecordingSession>;
recordingsDir: string;
}
/**
* Creates the event handler for `receiver.speaking.on("start", handler)`.
*
* The returned handler manages the full lifecycle for a user who starts speaking:
* 1. Validates the user (skip bot self, skip already-subscribed)
* 2. Collects user metadata and notifies the event broadcaster
* 3. Sets up the audio stream, decoder, packet filter, and segment manager
* 4. Attaches stream event handlers (data, end, error) BEFORE piping
* 5. Pipes audio through the packet filter for OGG recording
* 6. Handles segment completion (metadata write, upload trigger)
*/
export function createSpeakingHandler(
ctx: SpeakingHandlerContext,
): (userId: string) => Promise<void> {
const {
client,
channel,
receiver,
eventBroadcaster,
activeSessions,
recordingsDir,
} = ctx;
return async (userId: string) => {
// Skip the bot's own audio
if (userId === client.user?.id) return;
const userMetadata = await collectUserMetadata(client, userId, channel);
if (userMetadata.bot) return;
logger.debug(
{ userId, username: userMetadata.username },
"Voice activity detected",
);
// Notify webserver / WebSocket clients
eventBroadcaster?.voiceActiveUser(userId, {
username: userMetadata.username,
avatar: userMetadata.avatarUrl,
speaking: true,
});
// Skip if user already has an active stream subscription
if (receiver.subscriptions.has(userId)) return;
// Ensure per-user recording directory
const userDir = path.join(recordingsDir, userId);
await fsPromises.mkdir(userDir, { recursive: true }).catch(() => {
// Directory already exists, ignore
});
try {
// Step 1: Set up stream components (subscribe, decoder, filter, segment
// manager). NOTE: pipe() is NOT called here — we attach event handlers
// first to prevent data loss from race conditions.
const { audioStream, packetFilter, segmentManager, decoder } =
setupUserStream({
userId,
receiver,
userDir,
onPcmData: (pcm) => {
eventBroadcaster?.voicePcmData(pcm, userId);
},
});
// Step 2: Attach all audioStream event handlers BEFORE pipe()
audioStream.on("data", (chunk: Buffer) => {
if (chunk.length < 8) return;
segmentManager.rotateIfNeeded(packetFilter);
decoder.rotateIfNeeded();
decoder.write(chunk);
});
audioStream.on("end", () => {
segmentManager.close(packetFilter);
decoder.destroy();
eventBroadcaster?.voiceActiveUser(userId, {
username: userMetadata.username,
avatar: userMetadata.avatarUrl,
speaking: false,
});
});
audioStream.on("error", (error: Error) => {
segmentManager.close(packetFilter);
decoder.destroy();
logger.error({ userId, error: error.message }, "Audio stream error");
});
// Step 3: Now pipe for OGG recording (safe — event handlers attached)
const oggPacketStream = audioStream.pipe(packetFilter);
// Step 4: Open the first segment
const activeSession = activeSessions.get(channel.guild.id);
let currentSegment = segmentManager.open(oggPacketStream);
// Step 5: Handle segment file completion
currentSegment.out.on("finish", () => {
finalizeSegment({
currentSegment,
userMetadata,
activeSession,
guildId: channel.guild.id,
channelId: channel.id,
channelName: channel.name,
});
});
currentSegment.out.on("error", (err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error({ userId, error: msg }, "File write error");
});
// Step 6: Handle packet filter errors
packetFilter.on("error", (err) => {
segmentManager.close(oggPacketStream);
logger.error({ userId, error: err.message }, "PacketFilter error");
});
} catch (e) {
logger.error(
{ userId, error: e instanceof Error ? e.message : String(e) },
"Failed to create stream",
);
}
};
}
@@ -0,0 +1,74 @@
import { createChildLogger } from "@bete/shared/logger";
import type { VoiceConnection } from "@discordjs/voice";
import { EndBehaviorType } from "@discordjs/voice";
import { config } from "../../../shared/config/config.js";
import { PacketFilter } from "../packetFilter.js";
import { OpusDecoder } from "./decoder.js";
import { SegmentManager } from "./segment.js";
const logger = createChildLogger("stream-setup");
export interface StreamSetupInput {
userId: string;
receiver: VoiceConnection["receiver"];
userDir: string;
onPcmData: (pcm: Buffer) => void;
}
export interface StreamSetupResult {
audioStream: NodeJS.ReadableStream;
packetFilter: PacketFilter;
segmentManager: SegmentManager;
decoder: OpusDecoder;
}
/**
* Creates the audio stream subscription, decoder, packet filter, and segment
* manager for a user who started speaking.
*
* NOTE: This function does NOT pipe the audio stream through the packet filter.
* The caller must attach event handlers to `audioStream` BEFORE calling
* `audioStream.pipe(packetFilter)` to prevent data loss from race conditions.
*/
export function setupUserStream(input: StreamSetupInput): StreamSetupResult {
const { userId, receiver, userDir, onPcmData } = input;
logger.debug({ userId }, "Setting up user audio stream");
// Subscribe to the audio stream from the Discord voice receiver
const audioStream = receiver.subscribe(userId, {
end: {
behavior: EndBehaviorType.AfterSilence,
duration: config.AUDIO_STREAM_SILENCE_DURATION_MS,
},
});
const packetFilter = new PacketFilter(config.PACKET_FILTER_MIN_SIZE);
const segmentManager = new SegmentManager(
userDir,
config.RECORDING_SEGMENT_MS,
);
// Create decoder for web broadcast (PCM downsampling)
const decoder = new OpusDecoder({
cooldownMs: config.DECODER_COOLDOWN_MS,
rotateMs: config.DECODER_ROTATE_MS,
onData: (pcm: Buffer) => {
// Downsample 48kHz stereo -> 24kHz mono (left channel, every 2nd sample)
const outBuf = Buffer.alloc(pcm.length / 4);
for (let i = 0; i < outBuf.length / 2; i++) {
outBuf.writeInt16LE(pcm.readInt16LE(i * 8), i * 2);
}
onPcmData(outBuf);
},
});
logger.debug({ userId }, "User audio stream setup complete");
return {
audioStream,
packetFilter,
segmentManager,
decoder,
};
}
@@ -1,5 +1,6 @@
import { spawn } from "node:child_process";
import { PassThrough } from "node:stream";
import { BACKEND_VOICE_TRANSMIT } from "@bete/shared";
import { createChildLogger } from "@bete/shared/logger";
import { StreamType } from "@discordjs/voice";
import type Redis from "ioredis";
@@ -19,12 +20,13 @@ export class VoiceTransmitter {
private pcmStream: PassThrough | null = null;
private ffmpegProcess: ReturnType<typeof spawn> | null = null;
private isActive = false;
private readonly TRANSMIT_CHANNEL = "backend:voice:transmit";
private readonly TRANSMIT_CHANNEL = BACKEND_VOICE_TRANSMIT;
/**
* Start listening for PCM audio data from Redis and stream to Discord
*/
async start(redis: Redis): Promise<void> {
logger.info("Transmitter start requested");
if (this.isActive) {
logger.warn("Voice transmitter already active");
return;
@@ -143,6 +145,7 @@ export class VoiceTransmitter {
* Stop transmitting and clean up resources
*/
async stop(): Promise<void> {
logger.info("Transmitter stop requested");
if (!this.isActive) return;
this.isActive = false;
@@ -40,6 +40,7 @@ export class VoiceController {
constructor(private readonly client: Client) {}
getStatus(): VoiceStatus {
logger.debug("getStatus called");
const connection = this.activeGuildId
? getVoiceConnection(this.activeGuildId)
: undefined;
@@ -54,12 +55,14 @@ export class VoiceController {
}
listGuilds(): GuildSummary[] {
logger.info("listGuilds called");
return this.client.guilds.cache
.map((guild) => ({ id: guild.id, name: guild.name }))
.sort((a, b) => a.name.localeCompare(b.name));
}
async listVoiceChannels(guildId: string): Promise<VoiceChannelSummary[]> {
logger.info({ guildId }, "listVoiceChannels called");
const guild = this.getGuild(guildId);
await guild.channels.fetch().catch(() => null);
@@ -70,6 +73,7 @@ export class VoiceController {
}
async listWatchableChannels(guildId: string): Promise<ChannelSummary[]> {
logger.info({ guildId }, "listWatchableChannels called");
const guild = this.getGuild(guildId);
await guild.channels.fetch().catch(() => null);
@@ -84,6 +88,7 @@ export class VoiceController {
}
async connect(guildId: string, channelId: string): Promise<VoiceStatus> {
logger.info({ guildId, channelId }, "connect called");
if (!this.client.isReady()) {
throw new AppError(
"Discord client is not ready",
@@ -155,6 +160,7 @@ export class VoiceController {
}
async disconnect(): Promise<VoiceStatus> {
logger.info("disconnect called");
if (this.activeGuildId) {
stopRecording(this.activeGuildId);
}
@@ -13,7 +13,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig {
return {
...parsed,
EFFECTIVE_TEXT_GUILD_ID: parsed.TEXT_GUILD_ID ?? parsed.MONITOR_GUILD_ID,
EFFECTIVE_VOICE_GUILD_ID: parsed.VOICE_GUILD_ID ?? parsed.GUILD_ID,
EFFECTIVE_VOICE_GUILD_ID: parsed.VOICE_GUILD_ID,
};
}