refactor: atomic, DRY, and logging improvements across codebase

- Split llmModerationClient.ts (2170 lines) into 5 focused sub-modules
- Split aiAnalyzer.ts (1282 lines) into 4 modular pipelines
- Split messages.db.ts (826 lines) into 5 domain-specific modules
- Moved shared schema to @bete/shared, eliminated backend duplication
- Added createChildLogger to all voice-recording and AI moderation modules
- Extracted tryCommandThenFallback, normalizeMediaState, DEFAULT_VOICE_STATUS
- Created shared pagination.ts utility, eliminated 5+ cursor-pagination duplications
- Created shared messageMapper.ts for row mapping
- Standardized backend error handling with asyncHandler
- Added frontend createLogger utility and useAsyncAction hook
- Added structured logging to frontend hooks, socket, and API client

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MythEclipse
2026-06-09 19:46:08 +07:00
co-authored by Claude Opus 4.8
parent b68789fffc
commit 07032ab521
61 changed files with 3808 additions and 3043 deletions
@@ -1,3 +1,4 @@
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../shared/config/config.js";
import { initializeDatabase } from "../../shared/database/drizzle.js";
import {
@@ -15,6 +16,8 @@ import {
runSimpleTextFallback,
} from "./llmModerationClient.js";
const logger = createChildLogger("aiAnalysisWorker");
let dbInitialized = false;
let dbInitPromise: Promise<any> | null = null;
@@ -70,13 +73,9 @@ export default async function workerRouter(
if (!config.AI_LLM_API_KEY) {
const errorMsg =
"AI_LLM_API_KEY is missing from environment. Worker cannot process moderation requests without credentials.";
console.error(
JSON.stringify({
level: "ERROR",
context: "aiAnalysisWorker",
error: errorMsg,
timestamp: new Date().toISOString(),
}),
logger.error(
{ error: errorMsg },
"AI_LLM_API_KEY is missing from environment",
);
if (job.type === "batch") {
@@ -113,15 +112,9 @@ export default async function workerRouter(
} catch (error) {
const errorMessage = error instanceof Error ? error.message : String(error);
const errorStack = error instanceof Error ? error.stack : undefined;
console.error(
JSON.stringify({
level: "ERROR",
context: "aiAnalysisWorker",
type: job.type,
error: errorMessage,
stack: errorStack,
timestamp: new Date().toISOString(),
}),
logger.error(
{ type: job.type, error: errorMessage, stack: errorStack },
"Worker job failed",
);
if (job.type === "batch") {
return {
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,399 @@
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../shared/config/config.js";
import { isAgeRestrictedMetadata } from "../message-capture/messageMetadata.js";
import { updateMessagesAIAnalysisBulk } from "../message-capture/messageStore.js";
import type { MessageRecord } from "../message-capture/types.js";
import {
broadcastAnalysisCompleted,
conversationErrorCooldown,
conversationProcessing,
LAST_ERROR,
recordConversationBatchFailure,
resetConversationBatchFailures,
scheduleAutoDelete,
workerPool,
} from "./circuitBreaker.js";
import { estimateTokens } from "./conversationContext.js";
import { enqueueIndividualFallbacks } from "./individualFallbackProcessor.js";
const logger = createChildLogger("batch-processor");
export interface AnalysisWorkerResponse {
ok: boolean;
conversationKey: string;
rows: MessageRecord[];
error?: string;
}
// ---------------------------------------------------------------------------
// Observability
// ---------------------------------------------------------------------------
export let activeRequests = 0;
// ---------------------------------------------------------------------------
// Exported helpers
// ---------------------------------------------------------------------------
/**
* Picks a batch of messages within a token budget.
* `tokensPerMessage` accounts for JSON structure overhead around each entry.
* Uses a rough character-based token estimate (avoids async formatMessageForPrompt
* since this function runs in a synchronous promise chain).
*/
export function pickBatchWithinBudget(
messages: MessageRecord[],
maxTokens: number,
tokensPerMessage: number,
): MessageRecord[] {
const batch: MessageRecord[] = [];
let usedTokens = 0;
for (const msg of messages) {
const content = msg.edited_content ?? msg.content;
// Accurate token count via tiktoken (+ overhead for JSON structure)
const msgTokens = estimateTokens(content) + tokensPerMessage;
if (usedTokens + msgTokens <= maxTokens) {
batch.push(msg);
usedTokens += msgTokens;
}
}
return batch;
}
// ---------------------------------------------------------------------------
// Age-restricted message helpers
// ---------------------------------------------------------------------------
export function isAgeRestrictedMessage(message: MessageRecord): boolean {
return isAgeRestrictedMetadata(message.metadata);
}
export function buildAgeRestrictedSkipResult(): {
status: "clean";
flags: string | null;
score: number;
analysis: string;
categories: string[];
severity: "none";
confidence: number;
recommendedAction: "none";
analyzedAt: number;
error: null;
} {
return {
status: "clean",
flags: JSON.stringify(["age_restricted"]),
score: 0,
analysis: "Skipped moderation for age-restricted content.",
categories: ["age_restricted"],
severity: "none",
confidence: 1,
recommendedAction: "none",
analyzedAt: Date.now(),
error: null,
};
}
export async function skipAgeRestrictedMessages(
messages: MessageRecord[],
): Promise<MessageRecord[]> {
const ageRestrictedMessages = messages.filter(isAgeRestrictedMessage);
if (ageRestrictedMessages.length === 0) {
return messages;
}
const skippedRows = await updateMessagesAIAnalysisBulk(
ageRestrictedMessages.map((message) => ({
messageId: message.id,
result: buildAgeRestrictedSkipResult(),
})),
);
for (const row of skippedRows) {
broadcastAnalysisCompleted(row);
}
const skippedIds = new Set(
ageRestrictedMessages.map((message) => message.id),
);
return messages.filter((message) => !skippedIds.has(message.id));
}
// ---------------------------------------------------------------------------
// Batch pipeline
// ---------------------------------------------------------------------------
async function postBatchReputationUpdate(rows: MessageRecord[]): Promise<void> {
for (const row of rows) {
if (row.ai_status === "clean") {
import("./userReputationStore.js")
.then((store) => store.recordCleanMessage(row.user_id, row.guild_id))
.catch((e) =>
logger.error({ error: e }, "Failed to record clean message streak"),
);
} else if (row.ai_status === "flagged" && row.ai_severity !== "none") {
import("./userReputationStore.js")
.then((store) =>
store.recordInfraction(
row.user_id,
row.guild_id,
row.ai_severity as "low" | "medium" | "high" | "critical",
),
)
.catch((e) =>
logger.error({ error: e }, "Failed to record infraction penalty"),
);
}
}
}
export async function processBatch(
conversationKey: string,
messages: MessageRecord[],
processingStartedAt: number,
): Promise<void> {
if (messages.length === 0) {
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
return;
}
const cooldownUntil = conversationErrorCooldown.get(conversationKey) ?? 0;
if (Date.now() < cooldownUntil) {
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
return;
}
activeRequests++;
let shouldScheduleNext = false;
try {
const result = (await workerPool.run({
type: "batch",
conversationKey,
messages,
})) as AnalysisWorkerResponse;
// Do not broadcast or auto-delete if it's an API failure that will be reverted.
for (const row of result.rows) {
let isApiFailure = false;
if (row.ai_status === "error") {
try {
const flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[];
isApiFailure = flags.includes("analysis_api_failed");
} catch {}
}
if (!isApiFailure) {
broadcastAnalysisCompleted(row);
scheduleAutoDelete(row);
}
}
// Post-batch reputation updates (fire-and-forget)
postBatchReputationUpdate(
result.rows.filter((r) => {
if (r.ai_status === "error") {
try {
const flags = JSON.parse(r.ai_moderation_flags ?? "[]") as string[];
return !flags.includes("analysis_api_failed");
} catch {
return false;
}
}
return true;
}),
);
if (!result.ok) {
recordConversationBatchFailure(conversationKey);
// Batch failed entirely -- fall back all messages to individual queue
logger.warn(
{
conversationKey,
messageCount: messages.length,
error: result.error,
},
"Batch failed entirely -- routing all messages to individual fallback queue",
);
enqueueIndividualFallbacks(messages);
LAST_ERROR.value = result.error ?? "Analysis worker failed";
conversationErrorCooldown.set(
conversationKey,
Date.now() + config.AI_ANALYSIS_ERROR_COOLDOWN_MS,
);
logger.error(
{
conversationKey,
error: LAST_ERROR.value,
messageCount: messages.length,
messageIds: messages.map((m) => m.id),
cooldownUntil: new Date(
Date.now() + config.AI_ANALYSIS_ERROR_COOLDOWN_MS,
).toISOString(),
timestamp: new Date().toISOString(),
},
"Batch analysis failed, will retry after cooldown",
);
return;
}
// Batch succeeded -- check for messages the LLM silently dropped or failed
const incompleteMessages: MessageRecord[] = [];
const parseFailedMessages: MessageRecord[] = [];
const apiFailedMessages: MessageRecord[] = [];
for (const msg of messages) {
const row = result.rows.find((r) => r.id === msg.id);
if (!row) {
incompleteMessages.push(msg);
continue;
}
if (row.ai_status === "error") {
let flags: string[] = [];
try {
flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[];
} catch {}
if (flags.includes("analysis_incomplete")) {
incompleteMessages.push(msg);
} else if (flags.includes("analysis_parse_failed")) {
parseFailedMessages.push(msg);
} else if (flags.includes("analysis_api_failed")) {
apiFailedMessages.push(msg);
}
}
}
const messagesForIndividualQueue = [
...incompleteMessages,
...parseFailedMessages,
];
if (messagesForIndividualQueue.length > 0) {
logger.warn(
{
conversationKey,
count: messagesForIndividualQueue.length,
ids: messagesForIndividualQueue.map((m) => m.id),
totalBatchSize: messages.length,
},
"Batch returned incomplete or unparseable results -- fanning out to individual fallback queue",
);
enqueueIndividualFallbacks(messagesForIndividualQueue);
}
if (apiFailedMessages.length > 0) {
logger.warn(
{
conversationKey,
count: apiFailedMessages.length,
ids: apiFailedMessages.map((m) => m.id),
},
"Batch returned API failures -- reverting to pending to put back in queue",
);
// Revert to pending so they are picked up again
const revertedRows = await updateMessagesAIAnalysisBulk(
apiFailedMessages.map((msg) => ({
messageId: msg.id,
result: {
status: "pending",
flags: null,
score: null,
analysis: null,
categories: null,
severity: null,
confidence: null,
recommendedAction: null,
analyzedAt: null,
error: null,
},
})),
).catch((err) => {
logger.error(
{ error: String(err) },
"Failed to revert API failures to pending",
);
return [];
});
for (const row of revertedRows) {
broadcastAnalysisCompleted(row);
}
// Trigger conversation cooldown
recordConversationBatchFailure(conversationKey);
const existingCooldown =
conversationErrorCooldown.get(conversationKey) ?? 0;
const newCooldown = Date.now() + config.AI_ANALYSIS_ERROR_COOLDOWN_MS;
if (newCooldown > existingCooldown) {
conversationErrorCooldown.set(conversationKey, newCooldown);
}
// Release the processing lock immediately so the cooldown timer controls retry
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
// Do NOT schedule next -- let the cooldown gate it
shouldScheduleNext = false;
}
if (apiFailedMessages.length === 0) {
resetConversationBatchFailures(conversationKey);
conversationErrorCooldown.delete(conversationKey);
}
shouldScheduleNext = true;
} catch (error) {
recordConversationBatchFailure(conversationKey);
logger.warn(
{ conversationKey, messageCount: messages.length },
"Batch threw exception -- routing all messages to individual fallback queue",
);
enqueueIndividualFallbacks(messages);
LAST_ERROR.value = error instanceof Error ? error.message : String(error);
const errorStack = error instanceof Error ? error.stack : undefined;
const existingCatchCooldown =
conversationErrorCooldown.get(conversationKey) ?? 0;
const newCatchCooldown = Date.now() + config.AI_ANALYSIS_ERROR_COOLDOWN_MS;
if (newCatchCooldown > existingCatchCooldown) {
conversationErrorCooldown.set(conversationKey, newCatchCooldown);
}
logger.error(
{
conversationKey,
error: LAST_ERROR.value,
stack: errorStack,
messageCount: messages.length,
messageIds: messages.map((m) => m.id),
cooldownUntil: new Date(
Date.now() + config.AI_ANALYSIS_ERROR_COOLDOWN_MS,
).toISOString(),
timestamp: new Date().toISOString(),
},
"Analysis worker failed, will retry after cooldown",
);
} finally {
activeRequests--;
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
if (shouldScheduleNext) {
setImmediate(() => {
// Dynamic import to avoid circular dependency at module scope
import("./batchScheduler.js").then((m) =>
m.scheduleConversationAnalysis(conversationKey),
);
});
}
}
}
@@ -0,0 +1,140 @@
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../shared/config/config.js";
import { getPendingMessagesByConversation } from "../message-capture/messageStore.js";
import type { MessageRecord } from "../message-capture/types.js";
import {
pickBatchWithinBudget,
processBatch,
skipAgeRestrictedMessages,
} from "./batchProcessor.js";
import {
conversationConsecutiveErrors,
conversationDebounceTimers,
conversationErrorCooldown,
conversationProcessing,
isConversationProcessingLocked,
MAX_CONSECUTIVE_ERRORS,
} from "./circuitBreaker.js";
const logger = createChildLogger("batch-scheduler");
// ---------------------------------------------------------------------------
// Scheduling
// ---------------------------------------------------------------------------
/**
* Schedules a debounced analysis run for a conversation.
*
* FIX #3: The async work inside setTimeout is now wrapped in an explicit
* .catch() so DB errors don't produce unhandled promise rejections.
* FIX #6: Calls pickBatchWithinBudget after fetching messages so token budget
* is respected before handing the batch to the LLM.
* FIX #7: Unified single-timer path -- always clear-and-reset one timer per
* conversation key regardless of whether a cooldown is active. The delay is
* simply max(cooldownRemainder+500, debounce) so the same timer serves both
* the "throttled by error cooldown" and "normal debounce" cases, eliminating
* the previous two-path logic that could leave both timers live simultaneously.
*/
export function scheduleConversationAnalysis(conversationKey: string): void {
if (isConversationProcessingLocked(conversationKey)) {
return;
}
const convoCooldown = conversationErrorCooldown.get(conversationKey) ?? 0;
const convoErrors = conversationConsecutiveErrors.get(conversationKey) ?? 0;
// Hard-block: circuit breaker threshold reached AND cooldown still active.
if (convoErrors >= MAX_CONSECUTIVE_ERRORS && Date.now() < convoCooldown) {
return;
}
// Unified delay: honour the cooldown window if active, otherwise use the
// normal debounce interval. Always clear-and-reset so only ONE timer is
// ever pending per conversation key regardless of call source.
const now = Date.now();
const delayMs =
convoCooldown > now
? convoCooldown - now + 500
: config.AI_ANALYSIS_DEBOUNCE_MS;
const existingTimer = conversationDebounceTimers.get(conversationKey);
if (existingTimer) {
clearTimeout(existingTimer);
}
const timer = setTimeout(() => {
conversationDebounceTimers.delete(conversationKey);
// FIX TOCTOU: Set lock synchronously BEFORE the async DB fetch starts
if (isConversationProcessingLocked(conversationKey)) {
return;
}
const processingStartedAt = Date.now();
conversationProcessing.set(conversationKey, processingStartedAt);
// FIX #3: explicit .catch() -- no async arrow function to avoid unhandled rejection.
getPendingMessagesByConversation(
conversationKey,
config.AI_ANALYSIS_MAX_BATCH_SIZE,
)
.then(async (messages: MessageRecord[]) => {
if (messages.length === 0) {
if (
conversationProcessing.get(conversationKey) === processingStartedAt
) {
conversationProcessing.delete(conversationKey);
}
return;
}
const processableMessages = await skipAgeRestrictedMessages(messages);
if (processableMessages.length === 0) {
if (
conversationProcessing.get(conversationKey) === processingStartedAt
) {
conversationProcessing.delete(conversationKey);
}
return;
}
// FIX #6: trim to token budget before sending to LLM.
let trimmed = pickBatchWithinBudget(
processableMessages,
config.AI_ANALYSIS_MAX_TARGET_TOKENS,
50,
);
// FIX #10: if every message individually exceeds the token budget,
// fall back to the first message alone.
if (trimmed.length === 0 && processableMessages.length > 0) {
trimmed = processableMessages.slice(0, 1);
logger.warn(
{
conversationKey,
messageId: processableMessages[0]?.id,
tokenBudget: config.AI_ANALYSIS_MAX_TARGET_TOKENS,
},
"All messages exceed token budget -- processing first message alone to avoid stuck-pending deadlock",
);
}
return processBatch(conversationKey, trimmed, processingStartedAt);
})
.catch((err: unknown) => {
if (
conversationProcessing.get(conversationKey) === processingStartedAt
) {
conversationProcessing.delete(conversationKey);
}
logger.error(
{
conversationKey,
error: err instanceof Error ? err.message : String(err),
},
"Failed to fetch or dispatch pending messages for scheduled analysis",
);
});
}, delayMs);
conversationDebounceTimers.set(conversationKey, timer);
}
@@ -1,3 +1,4 @@
import { createChildLogger } from "@bete/shared/logger";
import { eq } from "drizzle-orm";
import { getDatabase } from "../../shared/database/drizzle.js";
import {
@@ -5,6 +6,8 @@ import {
channelCulturesTable,
} from "../../shared/database/schema.js";
const logger = createChildLogger("channelCultureStore");
/**
* Fetch the AI-generated culture summary for a channel.
*/
@@ -18,6 +21,11 @@ export async function getChannelCulture(
.where(eq(channelCulturesTable.channel_id, channelId))
.limit(1);
if (existing[0]) {
logger.debug({ channelId }, "Channel culture lookup: found");
} else {
logger.debug({ channelId }, "Channel culture lookup: not found");
}
return existing[0] || null;
}
@@ -46,4 +54,9 @@ export async function updateChannelCulture(
last_analyzed_at: Date.now(),
},
});
logger.debug(
{ channelId, guildId, cultureSummary },
"Channel culture updated",
);
}
@@ -0,0 +1,235 @@
import { existsSync } from "node:fs";
import { availableParallelism } from "node:os";
import { fileURLToPath } from "node:url";
import { createChildLogger } from "@bete/shared/logger";
import type { Client } from "discord.js-selfbot-v13";
import { LRUCache } from "lru-cache";
import { Piscina } from "piscina";
import { config } from "../../shared/config/config.js";
import type { EventBroadcaster } from "../event-broadcaster/index.js";
import type { MessageRecord } from "../message-capture/types.js";
import { attemptAutoDeleteFlaggedMessage } from "./autoDeleteManager.js";
const logger = createChildLogger("circuit-breaker");
// ---------------------------------------------------------------------------
// Piscina worker pool (shared by batch + individual pipelines)
// ---------------------------------------------------------------------------
function getAnalysisWorkerUrl(): URL {
const candidates = [
new URL("./aiAnalysisWorker.js", import.meta.url),
new URL("../aiAnalysisWorker.js", import.meta.url),
new URL("./aiAnalysisWorker.ts", import.meta.url),
];
for (const candidate of candidates) {
if (existsSync(fileURLToPath(candidate))) {
return candidate;
}
}
return candidates[2];
}
export const workerPool = new Piscina({
filename: fileURLToPath(getAnalysisWorkerUrl()),
execArgv: process.execArgv,
maxThreads: config.PISCINA_MAX_THREADS ?? availableParallelism(),
});
/**
* Gets the conversation key for a message (thread_id or channel_id).
*/
export function getConversationKey(message: MessageRecord): string {
return message.thread_id || message.channel_id;
}
// ---------------------------------------------------------------------------
// Shared observable state
// ---------------------------------------------------------------------------
/** Redis EventBroadcaster -- set externally so sub-modules can publish events. */
export let _redisEventBroadcaster: EventBroadcaster | undefined;
/** Discord client reference -- needed for auto-delete actions. */
export let moderationClient: Client | undefined;
export function setSharedEventBroadcaster(
eb: EventBroadcaster | undefined,
): void {
_redisEventBroadcaster = eb;
}
export function setModerationClient(mc: Client | undefined): void {
moderationClient = mc;
}
/**
* Per-message in-flight guard for the auto-delete side-effect.
* (LRU-backed to prevent unbounded growth)
*/
export const autoDeleteInFlight = new LRUCache<string, true>({ max: 10000 });
/** Last recorded error across all pipelines. */
export const LAST_ERROR: { value: string | null } = { value: null };
// ---------------------------------------------------------------------------
// Batch circuit breaker state
// ---------------------------------------------------------------------------
export const conversationConsecutiveErrors = new LRUCache<string, number>({
max: 10000,
});
export const MAX_CONSECUTIVE_ERRORS = 5;
export const CONVERSATION_CB_COOLDOWN_MS = 60000;
export const conversationErrorCooldown = new LRUCache<string, number>({
max: 10000,
});
// ---------------------------------------------------------------------------
// Scheduling / timing state (shared so sub-modules can access without cycles)
// ---------------------------------------------------------------------------
/** Debounce timer handle per conversation key. */
export const conversationDebounceTimers = new LRUCache<string, NodeJS.Timeout>({
max: 10000,
dispose: (value) => {
clearTimeout(value);
},
});
/** Timestamp of when processing started per conversation key. */
export const conversationProcessing = new LRUCache<string, number>({
max: 10000,
});
// ---------------------------------------------------------------------------
// Conversation lock helper
// ---------------------------------------------------------------------------
export function isConversationProcessingLocked(
conversationKey: string,
): boolean {
const startedAt = conversationProcessing.get(conversationKey);
return Boolean(
startedAt &&
Date.now() - startedAt < config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS,
);
}
// ---------------------------------------------------------------------------
// Alert system
// ---------------------------------------------------------------------------
export type CircuitBreakerAlert = {
type: "conversation_cb" | "individual_cb" | "sustained_error";
conversationKey?: string;
consecutiveErrors: number;
message: string;
lastError?: string | null;
};
const alertHandlers: Array<(alert: CircuitBreakerAlert) => void> = [];
/**
* Register an alert handler (e.g., for webhook integration).
*/
export function onCircuitBreakerAlert(
handler: (alert: CircuitBreakerAlert) => void,
): void {
alertHandlers.push(handler);
}
export function fireAlert(alert: CircuitBreakerAlert): void {
logger.warn(alert, `CB Alert: ${alert.type} -- ${alert.message}`);
for (const handler of alertHandlers) {
try {
handler(alert);
} catch {
// handler errors are non-critical
}
}
}
// ---------------------------------------------------------------------------
// Circuit breaker helpers
// ---------------------------------------------------------------------------
export function recordConversationBatchFailure(conversationKey: string): void {
const nextCount =
(conversationConsecutiveErrors.get(conversationKey) ?? 0) + 1;
conversationConsecutiveErrors.set(conversationKey, nextCount);
if (nextCount >= MAX_CONSECUTIVE_ERRORS) {
conversationErrorCooldown.set(
conversationKey,
Date.now() + CONVERSATION_CB_COOLDOWN_MS,
);
fireAlert({
type: "conversation_cb",
conversationKey,
consecutiveErrors: nextCount,
message: `Conversation ${conversationKey} circuit breaker triggered after ${nextCount} consecutive errors`,
lastError: LAST_ERROR.value,
});
conversationConsecutiveErrors.set(conversationKey, 0);
}
}
export function resetConversationBatchFailures(conversationKey: string): void {
conversationConsecutiveErrors.delete(conversationKey);
}
// ---------------------------------------------------------------------------
// Broadcast & auto-delete helpers
// ---------------------------------------------------------------------------
export function broadcastAnalysisCompleted(row: MessageRecord): void {
if (_redisEventBroadcaster) {
_redisEventBroadcaster.messageAnalyzed(row).catch((err: unknown) =>
logger.warn(
{
messageId: row.id,
error: err instanceof Error ? err.message : String(err),
},
"Failed to publish message_analyzed via Redis EventBroadcaster",
),
);
}
}
export function scheduleAutoDelete(row: MessageRecord): void {
if (row.ai_status !== "flagged" && row.ai_status !== "warn") return;
if (autoDeleteInFlight.has(row.id)) {
logger.debug(
{ messageId: row.id },
"Auto-delete skipped: already in-flight for this message",
);
return;
}
autoDeleteInFlight.set(row.id, true);
const run = () => {
attemptAutoDeleteFlaggedMessage(moderationClient, row)
.catch((error: unknown) => {
logger.error(
{
messageId: row.id,
error: error instanceof Error ? error.message : String(error),
},
"Unexpected auto-delete error",
);
})
.finally(() => {
autoDeleteInFlight.delete(row.id);
});
};
if (config.AUTO_DELETE_FLAGGED_DELAY_MS > 0) {
setTimeout(run, config.AUTO_DELETE_FLAGGED_DELAY_MS);
return;
}
setImmediate(run);
}
@@ -1,6 +1,9 @@
import { createChildLogger } from "@bete/shared/logger";
import pLimit from "p-limit";
import { config } from "../../shared/config/config.js";
const logger = createChildLogger("concurrencyLimiter");
/**
* Concurrency limiter for LLM API calls.
*
@@ -9,6 +12,42 @@ import { config } from "../../shared/config/config.js";
*/
const llmSemaphore = pLimit(config.AI_LLM_MAX_CONCURRENT ?? 5);
let activeCount = 0;
let pendingCount = 0;
// Track queue state changes for logging
function updateCounts(): void {
// p-limit exposes queueSize and activeCount via constructor internals,
// but we track via our wrapper to avoid depending on internals.
}
export async function withLlmConcurrency<T>(fn: () => Promise<T>): Promise<T> {
return llmSemaphore(fn);
const queuedAt = activeCount + pendingCount;
pendingCount++;
logger.debug(
{ activeCount, pendingCount, maxConcurrent: config.AI_LLM_MAX_CONCURRENT },
"Queuing LLM request",
);
return llmSemaphore(async () => {
pendingCount--;
activeCount++;
if (activeCount >= (config.AI_LLM_MAX_CONCURRENT ?? 5)) {
logger.warn(
{ activeCount, maxConcurrent: config.AI_LLM_MAX_CONCURRENT },
"LLM concurrency limit reached",
);
}
try {
return await fn();
} finally {
activeCount--;
logger.debug(
{ activeCount, pendingCount },
"LLM request completed, concurrency slot released",
);
}
});
}
@@ -1,7 +1,10 @@
import { createChildLogger } from "@bete/shared/logger";
import { encoding_for_model as encodingForModel } from "tiktoken";
import { formatMediaEvidenceForPrompt } from "../message-capture/messageMetadata.js";
import type { MessageRecord } from "../message-capture/types.js";
const logger = createChildLogger("conversationContext");
export interface ConversationContextInput {
contextBefore: MessageRecord[];
targets: MessageRecord[];
@@ -29,7 +32,12 @@ function formatTimestamp(ms: number): string {
*/
export function estimateTokens(text: string): number {
// Use tiktoken for accurate token counting (+15 overhead for JSON structure)
return getEncoder().encode(text).length + 15;
const tokens = getEncoder().encode(text).length + 15;
logger.debug(
{ tokenEstimate: tokens, textLength: text.length },
"Estimated tokens for text",
);
return tokens;
}
/**
@@ -81,5 +89,14 @@ export function buildConversationContext(
}
}
logger.debug(
{
targetCount: targets.length,
contextCount: selectedContextLines.length,
usedTokens,
maxTokens,
},
"Conversation context built",
);
return selectedContextLines;
}
@@ -0,0 +1,76 @@
import { createChildLogger } from "@bete/shared/logger";
const log = createChildLogger("imageMimeSniffer");
/**
* Sniff the first bytes of a buffer to determine if it is a supported image
* format. Returns the canonical MIME type string on success, or null if the
* bytes are not a recognizable image.
*/
export function sniffImageMimeType(buf: Buffer): string | null {
if (buf.length < 12) return null;
if (buf[0] === 0xff && buf[1] === 0xd8 && buf[2] === 0xff) {
return "image/jpeg";
}
if (
buf[0] === 0x89 &&
buf[1] === 0x50 &&
buf[2] === 0x4e &&
buf[3] === 0x47 &&
buf[4] === 0x0d &&
buf[5] === 0x0a &&
buf[6] === 0x1a &&
buf[7] === 0x0a
) {
return "image/png";
}
if (
buf[0] === 0x47 &&
buf[1] === 0x49 &&
buf[2] === 0x46 &&
buf[3] === 0x38
) {
return "image/gif";
}
if (
buf[0] === 0x52 &&
buf[1] === 0x49 &&
buf[2] === 0x46 &&
buf[3] === 0x46 &&
buf[8] === 0x57 &&
buf[9] === 0x45 &&
buf[10] === 0x42 &&
buf[11] === 0x50
) {
return "image/webp";
}
if (
buf.length >= 12 &&
buf[4] === 0x66 &&
buf[5] === 0x74 &&
buf[6] === 0x79 &&
buf[7] === 0x70
) {
const brand = buf.subarray(8, 12).toString("ascii");
if (brand.startsWith("avif") || brand.startsWith("avis")) {
return "image/avif";
}
if (
brand.startsWith("mif1") ||
brand.startsWith("heic") ||
brand.startsWith("heis")
) {
return "image/heic";
}
}
return null;
}
// Keep log referenced so TS does not tree-shake the logger init
log.debug("imageMimeSniffer loaded");
@@ -0,0 +1,346 @@
import { createChildLogger } from "@bete/shared/logger";
import { LRUCache } from "lru-cache";
import { config } from "../../shared/config/config.js";
import { updateMessagesAIAnalysisBulk } from "../message-capture/messageStore.js";
import type {
AnalysisResult,
MessageRecord,
} from "../message-capture/types.js";
import {
broadcastAnalysisCompleted,
fireAlert,
getConversationKey,
LAST_ERROR,
scheduleAutoDelete,
workerPool,
} from "./circuitBreaker.js";
import { logModerationError } from "./responseLogger.js";
const logger = createChildLogger("individual-fallback");
// ---------------------------------------------------------------------------
// Individual fallback queue state
// ---------------------------------------------------------------------------
/** IDs currently being processed one-by-one (LRU-backed, max 10k entries). */
export const individualInFlight = new LRUCache<string, true>({ max: 10000 });
/**
* Per-conversation count of in-flight individual messages.
* (LRU-backed to prevent unbounded growth)
*/
export const individualInFlightByConversation = new LRUCache<string, number>({
max: 10000,
});
/** Last-touched timestamp for pruning stale entries (LRU-backed). */
export const individualInFlightLastTouched = new LRUCache<string, number>({
max: 10000,
});
/** Counter for observability. */
export let activeIndividualRequests = 0;
// ---------------------------------------------------------------------------
// Individual fallback circuit breaker (independent of batch CB)
// ---------------------------------------------------------------------------
let individualConsecutiveErrors = 0;
export let individualCooldownUntil = 0;
const INDIVIDUAL_COOLDOWN_MS = 60000;
// ---------------------------------------------------------------------------
// Individual fallback pipeline
// ---------------------------------------------------------------------------
/**
* Processes a single message via the Piscina worker pool (offloaded from
* main thread to avoid blocking the event loop).
*/
async function processIndividualFallback(
message: MessageRecord,
): Promise<void> {
const { id: messageId } = message;
const conversationKey = getConversationKey(message);
activeIndividualRequests++;
individualInFlightByConversation.set(
conversationKey,
(individualInFlightByConversation.get(conversationKey) ?? 0) + 1,
);
individualInFlightLastTouched.set(conversationKey, Date.now());
let exhaustedOnIncomplete = false;
try {
// Run the LLM-heavy work in the worker thread
const workerResult = (await workerPool.run({
type: "individual",
message,
skipNormalAnalysis: false,
} as unknown)) as
| { ok: true; results: AnalysisResult[] }
| { ok: false; results: AnalysisResult[]; error: string };
let analysisResult: { results: AnalysisResult[] } | null = null;
let usedSimpleFallback = false;
if (workerResult.ok) {
const stillIncomplete = workerResult.results.some((r) =>
r.flags.includes("analysis_incomplete"),
);
if (stillIncomplete) {
exhaustedOnIncomplete = true;
analysisResult = null;
} else {
analysisResult = workerResult;
}
}
// Step 2: If normal analysis failed, try SIMPLE fallback via worker
if (!analysisResult) {
logger.info(
{ messageId },
"Normal analysis failed -- trying simple text fallback via worker",
);
const simpleResult = (await workerPool.run({
type: "individual",
message,
skipNormalAnalysis: true,
} as unknown)) as
| { ok: true; results: AnalysisResult[] }
| { ok: false; results: AnalysisResult[]; error: string };
if (simpleResult.ok) {
analysisResult = simpleResult;
usedSimpleFallback = true;
exhaustedOnIncomplete = false;
}
}
if (!analysisResult) {
throw new Error(
`Both normal and simple analysis failed for message ${messageId}`,
);
}
if (usedSimpleFallback) {
logger.info(
{ messageId, status: analysisResult.results[0]?.status },
"Used simple text fallback for individual message (via worker)",
);
}
// Main thread: DB writes + broadcast
const updates = analysisResult.results.map((r) => ({
messageId: r.messageId,
result: {
status: r.status,
flags: JSON.stringify(r.flags),
score: r.score,
analysis: r.analysis,
categories: r.categories,
severity: r.severity,
confidence: r.confidence,
recommendedAction: r.recommendedAction,
analyzedAt: Date.now(),
error: null,
},
}));
const rows = await updateMessagesAIAnalysisBulk(updates);
for (const row of rows) {
broadcastAnalysisCompleted(row);
scheduleAutoDelete(row);
// Update reputation autonomously
if (row.ai_status === "clean") {
import("./userReputationStore.js")
.then((store) => store.recordCleanMessage(row.user_id, row.guild_id))
.catch((e) =>
logger.error(
{ error: e },
"Failed to record clean message streak in fallback",
),
);
} else if (row.ai_status === "flagged" && row.ai_severity !== "none") {
import("./userReputationStore.js")
.then((store) =>
store.recordInfraction(
row.user_id,
row.guild_id,
row.ai_severity as "low" | "medium" | "high" | "critical",
),
)
.catch((e) =>
logger.error(
{ error: e },
"Failed to record infraction penalty in fallback",
),
);
}
}
const resultSummary = analysisResult.results[0];
logModerationError([messageId], config.AI_LLM_MODEL, new Error("Success"), {
phase: "individual_fallback",
status: resultSummary?.status,
flags: resultSummary?.flags,
severity: resultSummary?.severity,
confidence: resultSummary?.confidence,
});
individualConsecutiveErrors = 0;
logger.debug(
{ messageId, status: analysisResult.results[0]?.status },
"Individual fallback analysis complete (via worker)",
);
} catch (error) {
individualConsecutiveErrors++;
if (
individualConsecutiveErrors >= config.AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD
) {
individualCooldownUntil = Date.now() + INDIVIDUAL_COOLDOWN_MS;
fireAlert({
type: "individual_cb",
consecutiveErrors: individualConsecutiveErrors,
message: `Individual fallback circuit breaker triggered after ${individualConsecutiveErrors} consecutive errors`,
lastError: LAST_ERROR.value,
});
}
LAST_ERROR.value = error instanceof Error ? error.message : String(error);
logModerationError(
[messageId],
config.AI_LLM_MODEL,
error as Error | string,
{
phase: "individual_fallback",
conversationKey,
exhaustedOnIncomplete,
},
);
if (exhaustedOnIncomplete) {
await updateMessagesAIAnalysisBulk([
{
messageId,
result: {
status: "error",
flags: JSON.stringify(["individual_analysis_exhausted"]),
score: 0,
analysis:
"Individual fallback exhausted all retries: LLM consistently dropped this message even in single-target mode",
categories: ["individual_analysis_exhausted"],
severity: "none",
confidence: 0,
recommendedAction: "review",
analyzedAt: Date.now(),
error: LAST_ERROR.value,
},
},
]).catch((dbErr: unknown) => {
logger.error(
{ messageId, error: String(dbErr) },
"Failed to write terminal exhausted status",
);
});
logger.warn(
{ messageId },
"Individual fallback exhausted -- marked as individual_analysis_exhausted",
);
} else {
logger.error(
{
messageId,
error: LAST_ERROR.value,
stack: error instanceof Error ? error.stack : undefined,
},
"Individual fallback analysis failed (transient) -- will be retried",
);
}
} finally {
activeIndividualRequests--;
individualInFlight.delete(messageId);
const prev = individualInFlightByConversation.get(conversationKey) ?? 1;
if (prev <= 1) {
individualInFlightByConversation.delete(conversationKey);
individualInFlightLastTouched.delete(conversationKey);
} else {
individualInFlightByConversation.set(conversationKey, prev - 1);
individualInFlightLastTouched.set(conversationKey, Date.now());
}
}
}
// ---------------------------------------------------------------------------
// Enqueue individual fallbacks
// ---------------------------------------------------------------------------
/**
* Fans out message records to the individual fallback queue.
*
* FIX #1: Checks concurrency cap before admitting new work.
* FIX #5: Checks individual circuit breaker before admitting new work.
*/
export function enqueueIndividualFallbacks(messages: MessageRecord[]): void {
// FIX #5: Honour the individual circuit breaker.
if (Date.now() < individualCooldownUntil) {
logger.warn(
{
until: new Date(individualCooldownUntil).toISOString(),
skipped: messages.length,
},
"Individual fallback circuit breaker active -- messages will be recovered later",
);
return;
}
// FIX #5: Enforce concurrency cap
const maxConcurrent = config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT ?? 50;
const availableSlots = Math.max(0, maxConcurrent - activeIndividualRequests);
if (availableSlots <= 0) {
logger.debug(
{ maxConcurrent, active: activeIndividualRequests },
"Individual fallback concurrency cap reached -- messages will be recovered later",
);
return;
}
const newMessages = messages
.filter((m) => !individualInFlight.has(m.id))
.slice(0, availableSlots);
if (newMessages.length === 0) return;
logger.debug(
{
count: newMessages.length,
messageIds: newMessages.map((m) => m.id),
},
"Enqueueing individual fallback analysis for batch-incomplete messages",
);
for (const msg of newMessages) {
individualInFlight.set(msg.id, true);
processIndividualFallback(msg).catch((err: unknown) => {
logger.error(
{ messageId: msg.id, error: String(err) },
"Unexpected uncaught error escaping processIndividualFallback",
);
individualInFlight.delete(msg.id);
const ck = getConversationKey(msg);
const prev = individualInFlightByConversation.get(ck) ?? 1;
if (prev <= 1) {
individualInFlightByConversation.delete(ck);
individualInFlightLastTouched.delete(ck);
} else {
individualInFlightByConversation.set(ck, prev - 1);
individualInFlightLastTouched.set(ck, Date.now());
}
});
}
}
@@ -0,0 +1,81 @@
import { createChildLogger } from "@bete/shared/logger";
const log = createChildLogger("jsonExtractor");
/**
* Helper to extract JSON from a potentially conversational or markdown-wrapped string.
*/
export function extractJson(content: string): unknown {
const codeBlockRegex = /```(?:json)?\s*([\s\S]*?)\s*```/g;
const matches = content.matchAll(codeBlockRegex);
for (const match of matches) {
const codeContent = match[1].trim();
try {
const parsed = JSON.parse(codeContent);
if (parsed && typeof parsed === "object") {
return parsed;
}
} catch (err) {
log.debug(
{ err: err instanceof Error ? err.message : String(err) },
"Failed to parse JSON from code block — trying next block",
);
}
}
for (let start = 0; start < content.length; start++) {
const firstChar = content[start];
if (firstChar !== "{" && firstChar !== "[") continue;
const stack = [firstChar];
let inString = false;
let escaped = false;
for (let i = start + 1; i < content.length; i++) {
const char = content[i];
if (inString) {
if (escaped) {
escaped = false;
} else if (char === "\\") {
escaped = true;
} else if (char === '"') {
inString = false;
}
continue;
}
if (char === '"') {
inString = true;
continue;
}
if (char === "{" || char === "[") {
stack.push(char);
continue;
}
const last = stack[stack.length - 1];
if ((char === "}" && last === "{") || (char === "]" && last === "[")) {
stack.pop();
if (stack.length === 0) {
const candidate = content.slice(start, i + 1);
try {
const parsed = JSON.parse(candidate);
if (parsed && typeof parsed === "object") {
return parsed;
}
} catch (err) {
log.debug(
{ err: err instanceof Error ? err.message : String(err) },
"Failed to parse JSON candidate — trying next position",
);
}
break;
}
}
}
}
throw new Error("No JSON object found in response");
}
@@ -2,7 +2,6 @@ import { createChildLogger } from "@bete/shared/logger";
import { delay, retryWithBackoff } from "@bete/shared/utils";
import { LRUCache } from "lru-cache";
import type { ChatCompletion } from "openai/resources/chat/completions";
import { z } from "zod";
import { config } from "../../shared/config/config.js";
import { resizeImageForVision } from "../attachment-upload/imageResizer.js";
import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js";
@@ -46,33 +45,31 @@ import {
import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js";
import { initializeUserReputation } from "./userReputationStore.js";
const SeveritySchema = z.enum(["none", "low", "medium", "high", "critical"]);
const RecommendedActionSchema = z.enum([
"none",
"monitor",
"warn",
"review",
"delete",
"escalate",
]);
export { sniffImageMimeType } from "./imageMimeSniffer.js";
export { extractJson } from "./jsonExtractor.js";
export {
parseModerationResponse,
sanitizeErrorMessage,
} from "./moderationResponseParser.js";
// Re-export all symbols from sub-modules to preserve public API
export {
ModerationResponseSchema,
RecommendedActionSchema,
ResultItemSchema,
SeveritySchema,
} from "./moderationSchemas.js";
export {
clampScore,
DEFERRAL_ANALYSIS_PATTERN,
DEFERRAL_EXCEPTION_PATTERN,
deriveRecommendedAction,
deriveSeverity,
hasDeferralAnalysis,
} from "./severityDeriver.js";
const ResultItemSchema = z.object({
message_id: z.union([z.string(), z.number()]).transform(String),
status: z.enum(["clean", "warn", "flagged"]),
flags: z.array(z.string()).optional(),
score: z.number(),
analysis: z.string().nullable().optional(),
categories: z.array(z.string()).optional(),
severity: SeveritySchema.optional(),
confidence: z.number().optional(),
recommended_action: RecommendedActionSchema.optional(),
policy_version: z.string().optional(),
evidence: z.array(z.string()).optional(),
});
const ModerationResponseSchema = z.object({
results: z.array(ResultItemSchema),
});
import { sniffImageMimeType } from "./imageMimeSniffer.js";
// Internal imports for functions used locally in the facade
import { parseModerationResponse } from "./moderationResponseParser.js";
const log = createChildLogger("llmModerationClient");
@@ -112,371 +109,6 @@ async function buildCorrectedFewShotExamples(): Promise<string> {
}
}
/**
* Enhanced deferral detection pattern (R9).
*
* Only matches patterns where the model explicitly states it cannot make
* a decision and needs human review. Removed overly broad patterns that
* caused false positives:
* - "admin (perlu|harus|sebaiknya)" common in regular sentences
* - "bisa (berpotensi|mengandung)" decisive statements, not deferral
* - "maaf|sorry" opinions/apologies, not deferral
* - "saya tidak yakin|tahu|paham" expressing uncertainty, not deferral
*/
const DEFERRAL_ANALYSIS_PATTERN =
/(?:kurang (?:konteks|bukti|informasi|data) (?:untuk (?:menilai|menentukan|memutuskan)|untuk moderasi)|perlu (?:dicek|diperiksa|ditinjau|dikaji|dievaluasi) (?:oleh )?(?:admin|moderator|manusia|human review)|tidak (?:bisa|dapat|mampu) (?:menentukan|menilai|memastikan|menyimpulkan|memberi keputusan|memoderasi).*(?:karena (?:konteks tidak jelas|informasi tidak cukup|bukti kurang|konteks kurang|tidak cukup konteks)|data tidak cukup|informasi tidak lengkap)|cannot determine|insufficient (?:context|evidence|information) (?:to |for )?(?:moderate|judge|evaluate|decide|classify)|(?:sepertinya|tampaknya) (?:perlu|harus) (?:ditinjau|diperiksa|dicek) (?:oleh )?(?:admin|moderator)|tidak cukup (?:bukti|informasi|konteks) (?:untuk (?:memberikan|membuat|menentukan)|memutuskan))/i;
/**
* Exceptions: patterns that look like deferral but are actually decisive.
* Expanded to catch more variations where the model gives a clear verdict.
*/
const DEFERRAL_EXCEPTION_PATTERN =
/tidak bisa menentukan.*(?:karena|sebab|dengan alasan|sebab tidak ada).*(?:clean|tidak (?:ada|terdapat|menunjukkan).*(?:pelanggaran|masalah|indikasi|konten)|aman|bersih|normal)/i;
function hasDeferralAnalysis(analysis: string): boolean {
if (DEFERRAL_EXCEPTION_PATTERN.test(analysis)) return false;
return DEFERRAL_ANALYSIS_PATTERN.test(analysis);
}
function clampScore(value: number | undefined, fallback = 0): number {
return Math.max(
0,
Math.min(1, Number.isFinite(value) ? (value as number) : fallback),
);
}
function deriveSeverity(
status: "clean" | "warn" | "flagged",
score: number,
): z.infer<typeof SeveritySchema> {
if (status === "clean") return "none";
if (status === "warn") return score >= 0.65 ? "medium" : "low";
if (score >= 0.9) return "critical";
return score >= 0.75 ? "high" : "medium";
}
function deriveRecommendedAction(
status: "clean" | "warn" | "flagged",
severity: z.infer<typeof SeveritySchema>,
): z.infer<typeof RecommendedActionSchema> {
if (status === "clean") return "none";
if (status === "warn") return severity === "medium" ? "review" : "warn";
if (severity === "critical") return "escalate";
if (severity === "high") return "delete";
return "review";
}
/**
* Helper to extract JSON from a potentially conversational or markdown-wrapped string.
*/
export function extractJson(content: string): unknown {
const codeBlockRegex = /```(?:json)?\s*([\s\S]*?)\s*```/g;
const matches = content.matchAll(codeBlockRegex);
for (const match of matches) {
const codeContent = match[1].trim();
try {
const parsed = JSON.parse(codeContent);
if (parsed && typeof parsed === "object") {
return parsed;
}
} catch (err) {
log.debug(
{ err: err instanceof Error ? err.message : String(err) },
"Failed to parse JSON from code block — trying next block",
);
}
}
for (let start = 0; start < content.length; start++) {
const firstChar = content[start];
if (firstChar !== "{" && firstChar !== "[") continue;
const stack = [firstChar];
let inString = false;
let escaped = false;
for (let i = start + 1; i < content.length; i++) {
const char = content[i];
if (inString) {
if (escaped) {
escaped = false;
} else if (char === "\\") {
escaped = true;
} else if (char === '"') {
inString = false;
}
continue;
}
if (char === '"') {
inString = true;
continue;
}
if (char === "{" || char === "[") {
stack.push(char);
continue;
}
const last = stack[stack.length - 1];
if ((char === "}" && last === "{") || (char === "]" && last === "[")) {
stack.pop();
if (stack.length === 0) {
const candidate = content.slice(start, i + 1);
try {
const parsed = JSON.parse(candidate);
if (parsed && typeof parsed === "object") {
return parsed;
}
} catch (err) {
log.debug(
{ err: err instanceof Error ? err.message : String(err) },
"Failed to parse JSON candidate — trying next position",
);
}
break;
}
}
}
}
throw new Error("No JSON object found in response");
}
/**
* Sanitize error messages for client-facing output (R10).
* Internal details are logged but the caller gets a generic message.
*/
function sanitizeErrorMessage(internalMsg: string, messageId: string): string {
// Log the full error for debugging
log.warn(
{ messageId, internalError: internalMsg },
"Internal moderation error (sanitized for client)",
);
// Return generic message without internal details
return `Analisis gagal dan memerlukan pemeriksaan manual. Error code: MOD_${Date.now().toString(36).slice(0, 6)}`;
}
export function parseModerationResponse(
content: string,
targetIds: string[],
): AnalysisResult[] {
let parsed: any;
try {
parsed = JSON.parse(content);
} catch (e) {
parsed = extractJson(content);
}
if (Array.isArray(parsed)) {
parsed = { results: parsed };
} else if (parsed && typeof parsed === "object" && !("results" in parsed)) {
if ("message_id" in parsed) {
parsed = { results: [parsed] };
} else {
const arrayKey = Object.keys(parsed).find((key) => {
const val = parsed[key];
return (
Array.isArray(val) &&
val.length > 0 &&
val.every(
(item: unknown) =>
typeof item === "object" &&
item !== null &&
"message_id" in (item as Record<string, unknown>),
)
);
});
if (arrayKey) {
parsed.results = parsed[arrayKey];
} else {
parsed = { results: [parsed] };
}
}
}
const parseResult = ModerationResponseSchema.safeParse(parsed);
if (!parseResult.success) {
throw new Error(`Zod validation failed: ${parseResult.error.message}`);
}
const response = parseResult.data;
const foundIds = new Set<string>();
const targetIdSet = new Set(targetIds);
const results: (AnalysisResult | null)[] = response.results.map((result) => {
const {
message_id,
status,
flags,
score,
analysis,
categories,
severity,
confidence,
recommended_action,
policy_version,
evidence,
} = result;
const finalId = message_id.trim();
if (!targetIdSet.has(finalId)) {
return null;
}
if (foundIds.has(finalId)) {
throw new Error(
`Duplicate message_id in moderation response: ${finalId}`,
);
}
foundIds.add(finalId);
const coalescedAnalysis = analysis ?? "";
if (hasDeferralAnalysis(coalescedAnalysis)) {
throw new Error(
`Deferral analysis is not allowed for message ${finalId}; return a direct moderation decision`,
);
}
const normalizedScore = clampScore(score);
const normalizedConfidence = clampScore(confidence, normalizedScore);
const normalizedSeverity =
severity ?? deriveSeverity(status, normalizedScore);
return {
messageId: finalId,
status: status as "clean" | "warn" | "flagged",
flags: flags ?? [],
score: normalizedScore,
analysis: coalescedAnalysis,
categories: categories ?? flags ?? [],
severity: normalizedSeverity,
confidence: normalizedConfidence,
recommendedAction:
recommended_action ??
deriveRecommendedAction(status, normalizedSeverity),
policyVersion: policy_version ?? "default-2026-05-30",
evidence: evidence ?? [],
};
});
const filteredResults = results.filter(
(r): r is AnalysisResult => r !== null,
);
const missingIds = targetIds.filter((id) => !foundIds.has(id));
if (missingIds.length > 0) {
log.warn(
{ missingIds, foundCount: foundIds.size, totalCount: targetIds.length },
"Some target IDs missing in response - marking as incomplete",
);
for (const missingId of missingIds) {
filteredResults.push({
messageId: missingId,
status: "error",
flags: ["analysis_incomplete"],
score: 0,
analysis: sanitizeErrorMessage(
"Analysis incomplete - LLM did not process this message",
missingId,
),
categories: ["analysis_incomplete"],
severity: "none",
confidence: 0,
recommendedAction: "review",
policyVersion: "default-2026-05-30",
evidence: [],
});
}
}
return filteredResults;
}
interface ModerationInput {
targets: MessageRecord[];
contextText: string;
attachments?: AttachmentRecord[];
}
interface ModerationOutput {
results: AnalysisResult[];
raw: unknown;
}
/**
* Sniff the first bytes of a buffer to determine if it is a supported image
* format. Returns the canonical MIME type string on success, or null if the
* bytes are not a recognizable image.
*/
function sniffImageMimeType(buf: Buffer): string | null {
if (buf.length < 12) return null;
if (buf[0] === 0xff && buf[1] === 0xd8 && buf[2] === 0xff) {
return "image/jpeg";
}
if (
buf[0] === 0x89 &&
buf[1] === 0x50 &&
buf[2] === 0x4e &&
buf[3] === 0x47 &&
buf[4] === 0x0d &&
buf[5] === 0x0a &&
buf[6] === 0x1a &&
buf[7] === 0x0a
) {
return "image/png";
}
if (
buf[0] === 0x47 &&
buf[1] === 0x49 &&
buf[2] === 0x46 &&
buf[3] === 0x38
) {
return "image/gif";
}
if (
buf[0] === 0x52 &&
buf[1] === 0x49 &&
buf[2] === 0x46 &&
buf[3] === 0x46 &&
buf[8] === 0x57 &&
buf[9] === 0x45 &&
buf[10] === 0x42 &&
buf[11] === 0x50
) {
return "image/webp";
}
if (
buf.length >= 12 &&
buf[4] === 0x66 &&
buf[5] === 0x74 &&
buf[6] === 0x79 &&
buf[7] === 0x70
) {
const brand = buf.subarray(8, 12).toString("ascii");
if (brand.startsWith("avif") || brand.startsWith("avis")) {
return "image/avif";
}
if (
brand.startsWith("mif1") ||
brand.startsWith("heic") ||
brand.startsWith("heis")
) {
return "image/heic";
}
}
return null;
}
// ---------------------------------------------------------------------------
// Shared types for image resolution
// ---------------------------------------------------------------------------
@@ -1495,6 +1127,17 @@ async function runMediaBatch(
// Main entry point — splits text-only vs media, runs both paths in parallel
// ---------------------------------------------------------------------------
interface ModerationInput {
targets: MessageRecord[];
contextText: string;
attachments?: AttachmentRecord[];
}
interface ModerationOutput {
results: AnalysisResult[];
raw: unknown;
}
/**
* Runs LLM-based moderation analysis on messages.
*
@@ -0,0 +1,178 @@
import { createChildLogger } from "@bete/shared/logger";
import type { AnalysisResult } from "../message-capture/types.js";
import { extractJson } from "./jsonExtractor.js";
import { ModerationResponseSchema } from "./moderationSchemas.js";
import {
clampScore,
DEFERRAL_ANALYSIS_PATTERN,
DEFERRAL_EXCEPTION_PATTERN,
deriveRecommendedAction,
deriveSeverity,
hasDeferralAnalysis,
} from "./severityDeriver.js";
const log = createChildLogger("moderationResponseParser");
/**
* Re-export deferral patterns for backward compatibility.
* See severityDeriver.ts for the full regex definitions.
*/
export {
DEFERRAL_ANALYSIS_PATTERN,
DEFERRAL_EXCEPTION_PATTERN,
} from "./severityDeriver.js";
/**
* Sanitize error messages for client-facing output (R10).
* Internal details are logged but the caller gets a generic message.
*/
export function sanitizeErrorMessage(
internalMsg: string,
messageId: string,
): string {
// Log the full error for debugging
log.warn(
{ messageId, internalError: internalMsg },
"Internal moderation error (sanitized for client)",
);
// Return generic message without internal details
return `Analisis gagal dan memerlukan pemeriksaan manual. Error code: MOD_${Date.now().toString(36).slice(0, 6)}`;
}
export function parseModerationResponse(
content: string,
targetIds: string[],
): AnalysisResult[] {
let parsed: any;
try {
parsed = JSON.parse(content);
} catch (e) {
parsed = extractJson(content);
}
if (Array.isArray(parsed)) {
parsed = { results: parsed };
} else if (parsed && typeof parsed === "object" && !("results" in parsed)) {
if ("message_id" in parsed) {
parsed = { results: [parsed] };
} else {
const arrayKey = Object.keys(parsed).find((key) => {
const val = parsed[key];
return (
Array.isArray(val) &&
val.length > 0 &&
val.every(
(item: unknown) =>
typeof item === "object" &&
item !== null &&
"message_id" in (item as Record<string, unknown>),
)
);
});
if (arrayKey) {
parsed.results = parsed[arrayKey];
} else {
parsed = { results: [parsed] };
}
}
}
const parseResult = ModerationResponseSchema.safeParse(parsed);
if (!parseResult.success) {
throw new Error(`Zod validation failed: ${parseResult.error.message}`);
}
const response = parseResult.data;
const foundIds = new Set<string>();
const targetIdSet = new Set(targetIds);
const results: (AnalysisResult | null)[] = response.results.map((result) => {
const {
message_id,
status,
flags,
score,
analysis,
categories,
severity,
confidence,
recommended_action,
policy_version,
evidence,
} = result;
const finalId = message_id.trim();
if (!targetIdSet.has(finalId)) {
return null;
}
if (foundIds.has(finalId)) {
throw new Error(
`Duplicate message_id in moderation response: ${finalId}`,
);
}
foundIds.add(finalId);
const coalescedAnalysis = analysis ?? "";
if (hasDeferralAnalysis(coalescedAnalysis)) {
throw new Error(
`Deferral analysis is not allowed for message ${finalId}; return a direct moderation decision`,
);
}
const normalizedScore = clampScore(score);
const normalizedConfidence = clampScore(confidence, normalizedScore);
const normalizedSeverity =
severity ?? deriveSeverity(status, normalizedScore);
return {
messageId: finalId,
status: status as "clean" | "warn" | "flagged",
flags: flags ?? [],
score: normalizedScore,
analysis: coalescedAnalysis,
categories: categories ?? flags ?? [],
severity: normalizedSeverity,
confidence: normalizedConfidence,
recommendedAction:
recommended_action ??
deriveRecommendedAction(status, normalizedSeverity),
policyVersion: policy_version ?? "default-2026-05-30",
evidence: evidence ?? [],
};
});
const filteredResults = results.filter(
(r): r is AnalysisResult => r !== null,
);
const missingIds = targetIds.filter((id) => !foundIds.has(id));
if (missingIds.length > 0) {
log.warn(
{ missingIds, foundCount: foundIds.size, totalCount: targetIds.length },
"Some target IDs missing in response - marking as incomplete",
);
for (const missingId of missingIds) {
filteredResults.push({
messageId: missingId,
status: "error",
flags: ["analysis_incomplete"],
score: 0,
analysis: sanitizeErrorMessage(
"Analysis incomplete - LLM did not process this message",
missingId,
),
categories: ["analysis_incomplete"],
severity: "none",
confidence: 0,
recommendedAction: "review",
policyVersion: "default-2026-05-30",
evidence: [],
});
}
}
return filteredResults;
}
@@ -0,0 +1,41 @@
import { createChildLogger } from "@bete/shared/logger";
import { z } from "zod";
const log = createChildLogger("moderationSchemas");
export const SeveritySchema = z.enum([
"none",
"low",
"medium",
"high",
"critical",
]);
export const RecommendedActionSchema = z.enum([
"none",
"monitor",
"warn",
"review",
"delete",
"escalate",
]);
export const ResultItemSchema = z.object({
message_id: z.union([z.string(), z.number()]).transform(String),
status: z.enum(["clean", "warn", "flagged"]),
flags: z.array(z.string()).optional(),
score: z.number(),
analysis: z.string().nullable().optional(),
categories: z.array(z.string()).optional(),
severity: SeveritySchema.optional(),
confidence: z.number().optional(),
recommended_action: RecommendedActionSchema.optional(),
policy_version: z.string().optional(),
evidence: z.array(z.string()).optional(),
});
export const ModerationResponseSchema = z.object({
results: z.array(ResultItemSchema),
});
// Keep log referenced so TS does not tree-shake the logger init
log.debug("moderationSchemas loaded");
@@ -0,0 +1,64 @@
import { createChildLogger } from "@bete/shared/logger";
import type { z } from "zod";
import {
RecommendedActionSchema,
SeveritySchema,
} from "./moderationSchemas.js";
const log = createChildLogger("severityDeriver");
/**
* Enhanced deferral detection pattern (R9).
*
* Only matches patterns where the model explicitly states it cannot make
* a decision and needs human review. Removed overly broad patterns that
* caused false positives:
* - "admin (perlu|harus|sebaiknya)" common in regular sentences
* - "bisa (berpotensi|mengandung)" decisive statements, not deferral
* - "maaf|sorry" opinions/apologies, not deferral
* - "saya tidak yakin|tahu|paham" expressing uncertainty, not deferral
*/
export const DEFERRAL_ANALYSIS_PATTERN =
/(?:kurang (?:konteks|bukti|informasi|data) (?:untuk (?:menilai|menentukan|memutuskan)|untuk moderasi)|perlu (?:dicek|diperiksa|ditinjau|dikaji|dievaluasi) (?:oleh )?(?:admin|moderator|manusia|human review)|tidak (?:bisa|dapat|mampu) (?:menentukan|menilai|memastikan|menyimpulkan|memberi keputusan|memoderasi).*(?:karena (?:konteks tidak jelas|informasi tidak cukup|bukti kurang|konteks kurang|tidak cukup konteks)|data tidak cukup|informasi tidak lengkap)|cannot determine|insufficient (?:context|evidence|information) (?:to |for )?(?:moderate|judge|evaluate|decide|classify)|(?:sepertinya|tampaknya) (?:perlu|harus) (?:ditinjau|diperiksa|dicek) (?:oleh )?(?:admin|moderator)|tidak cukup (?:bukti|informasi|konteks) (?:untuk (?:memberikan|membuat|menentukan)|memutuskan))/i;
/**
* Exceptions: patterns that look like deferral but are actually decisive.
* Expanded to catch more variations where the model gives a clear verdict.
*/
export const DEFERRAL_EXCEPTION_PATTERN =
/tidak bisa menentukan.*(?:karena|sebab|dengan alasan|sebab tidak ada).*(?:clean|tidak (?:ada|terdapat|menunjukkan).*(?:pelanggaran|masalah|indikasi|konten)|aman|bersih|normal)/i;
export function hasDeferralAnalysis(analysis: string): boolean {
if (DEFERRAL_EXCEPTION_PATTERN.test(analysis)) return false;
return DEFERRAL_ANALYSIS_PATTERN.test(analysis);
}
export function clampScore(value: number | undefined, fallback = 0): number {
return Math.max(
0,
Math.min(1, Number.isFinite(value) ? (value as number) : fallback),
);
}
export function deriveSeverity(
status: "clean" | "warn" | "flagged",
score: number,
): z.infer<typeof SeveritySchema> {
if (status === "clean") return "none";
if (status === "warn") return score >= 0.65 ? "medium" : "low";
if (score >= 0.9) return "critical";
return score >= 0.75 ? "high" : "medium";
}
export function deriveRecommendedAction(
status: "clean" | "warn" | "flagged",
severity: z.infer<typeof SeveritySchema>,
): z.infer<typeof RecommendedActionSchema> {
if (status === "clean") return "none";
if (status === "warn") return severity === "medium" ? "review" : "warn";
if (severity === "critical") return "escalate";
if (severity === "high") return "delete";
return "review";
}
log.debug("severityDeriver loaded");
@@ -1,3 +1,7 @@
import { createChildLogger } from "@bete/shared/logger";
const logger = createChildLogger("stickerPrompt");
/**
* Sticker-specific prompt templates for AI moderation.
*
@@ -17,6 +21,7 @@ export function buildStickerVisionPrompt(
stickerName: string,
messageId: string,
): string {
logger.debug({ stickerName, messageId }, "Building sticker vision prompt");
return [
`Analisis sticker Discord berikut sebagai evidence moderasi.`,
`Sticker "${stickerName}" berasal dari pesan id=${messageId}.`,
@@ -49,6 +54,10 @@ export function buildStickerTextOnlyWarning(
stickerName: string,
stickerUrl: string,
): string {
logger.debug(
{ stickerName, stickerUrl },
"Building sticker text-only warning",
);
return (
`[sticker: "${stickerName}" (${stickerUrl}) — GAMBAR GAGAL DIUNDUH. ` +
`"${stickerName}" adalah sticker kartun/meme Discord. ` +
@@ -68,6 +77,7 @@ export function buildCustomEmojiVisionPrompt(
emojiName: string,
messageId: string,
): string {
logger.debug({ emojiName, messageId }, "Building custom emoji vision prompt");
return [
`Analisis custom emoji Discord berikut sebagai evidence moderasi.`,
`Emoji "${emojiName}" berasal dari pesan id=${messageId}.`,
@@ -87,6 +97,7 @@ export function buildCustomEmojiVisionPrompt(
* Fallback text for when a custom emoji image failed to download.
*/
export function buildCustomEmojiTextOnlyFallback(emojiName: string): string {
logger.debug({ emojiName }, "Building custom emoji text-only fallback");
return (
`[custom_emoji: "${emojiName}" — GAMBAR GAGAL DIUNDUH. ` +
`"${emojiName}" adalah custom emoji Discord (ikon kecil). ` +
@@ -105,6 +116,7 @@ export function buildGeneralImageVisionPrompt(
sourceLabel: string,
_messageId: string,
): string {
logger.debug({ sourceLabel }, "Building general image vision prompt");
return [
`Deskripsikan gambar ini secara objektif dan spesifik.`,
`${sourceLabel}`,
@@ -1,3 +1,4 @@
import { createChildLogger } from "@bete/shared/logger";
import { and, desc, eq } from "drizzle-orm";
import { getDatabase } from "../../shared/database/drizzle.js";
import {
@@ -6,6 +7,8 @@ import {
userReputationsTable,
} from "../../shared/database/schema.js";
const logger = createChildLogger("userReputationStore");
/**
* Ensures a user reputation record exists.
*/
@@ -21,6 +24,7 @@ export async function initializeUserReputation(
.limit(1);
if (existing.length > 0) {
logger.debug({ userId }, "Reputation record already exists");
return existing[0];
}
@@ -40,6 +44,7 @@ export async function initializeUserReputation(
if (!inserted) {
// If concurrent insert happened
logger.debug({ userId }, "Concurrent reputation insert detected, retrying");
const retry = await db
.select()
.from(userReputationsTable)
@@ -48,6 +53,10 @@ export async function initializeUserReputation(
return retry[0];
}
logger.debug(
{ userId, trustScore: inserted.trust_score },
"Initialized user reputation",
);
return inserted;
}
@@ -64,6 +73,14 @@ export async function getUserReputation(
.where(eq(userReputationsTable.user_id, userId))
.limit(1);
if (existing[0]) {
logger.debug(
{ userId, trustScore: existing[0].trust_score },
"Fetched user reputation",
);
} else {
logger.debug({ userId }, "No reputation record found, returning null");
}
return existing[0] || null;
}
@@ -93,6 +110,11 @@ export async function recordCleanMessage(
updated_at: Date.now(),
})
.where(eq(userReputationsTable.user_id, userId));
logger.debug(
{ userId, previousScore: rep.trust_score, newScore, newStreak },
"Clean message recorded, reputation updated",
);
}
/**
@@ -133,6 +155,17 @@ export async function recordInfraction(
updated_at: Date.now(),
})
.where(eq(userReputationsTable.user_id, userId));
logger.info(
{
userId,
severity,
penalty,
newScore,
totalInfractions: rep.total_infractions + 1,
},
"Infraction recorded",
);
}
/**
@@ -1,8 +1,8 @@
import { decodeCursor, encodeCursor } from "@bete/shared";
import { createChildLogger, type Logger } from "@bete/shared/logger";
import type { NodePgDatabase } from "drizzle-orm/node-postgres";
import { getDatabase } from "../../shared/database/drizzle.js";
import type * as schema from "../../shared/database/schema.js";
import { decodeCursor, encodeCursor } from "../message-capture/pagination.js";
import type {
AttachmentRecord,
MessageQuery,
@@ -18,7 +18,7 @@ import { ModerationActionsDb } from "./moderation-actions.db.js";
import { RetentionDb } from "./retention.db.js";
import { ReviewsDb } from "./reviews.db.js";
export { decodeCursor, encodeCursor } from "../message-capture/pagination.js";
export { decodeCursor, encodeCursor } from "@bete/shared";
export type { AIAnalysisUpdate } from "./messages.db.js";
// ─── Lazy singleton ────────────────────────────────────────────────────────
@@ -0,0 +1,413 @@
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 type { MessageRecord } from "../message-capture/types.js";
// ─── Helpers ──────────────────────────────────────────────────────────────────
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;
}
// ─── MessagesAnalysis Class ───────────────────────────────────────────────────
export class MessagesAnalysis {
protected logger: Logger;
constructor(
protected db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-analysis");
}
// ── AI Analysis Updates ─────────────────────────────────────────────────────
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;
}
}
// ── 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;
}
}
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;
}
}
}
@@ -0,0 +1,86 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, eq, isNull, 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 type { MessageRecord } from "../message-capture/types.js";
// ─── MessagesCleanup Class ────────────────────────────────────────────────────
export class MessagesCleanup {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-cleanup");
}
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,204 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, or, type 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 type { MessageRecord } from "../message-capture/types.js";
// ─── Shared Helpers ──────────────────────────────────────────────────────────
export function channelOrThreadCondition(channelId: string): SQL {
return or(
eq(messagesTable.channel_id, channelId),
eq(messagesTable.thread_id, channelId),
) as SQL;
}
// ─── MessagesCrud Class ──────────────────────────────────────────────────────
export class MessagesCrud {
protected logger: Logger;
constructor(
protected db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-crud");
}
// ── INSERT ──────────────────────────────────────────────────────────────────
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;
}
}
// ── UPDATE ──────────────────────────────────────────────────────────────────
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;
}
}
// ── GET ─────────────────────────────────────────────────────────────────────
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[] = [channelOrThreadCondition(channelId)];
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;
}
}
}
@@ -1,826 +1,161 @@
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";
import type { AIAnalysisUpdate } from "./messages.analysis.js";
import { MessagesAnalysis } from "./messages.analysis.js";
import { MessagesCleanup } from "./messages.cleanup.js";
import { MessagesCrud } from "./messages.crud.js";
import { MessagesPagination } from "./messages.pagination.js";
import { MessagesSearch } from "./messages.search.js";
// ─── Helpers ────────────────────────────────────────────────────────────────
// Re-export AIAnalysisUpdate for consumers (messageStore.ts imports it)
export type { AIAnalysisUpdate } from "./messages.analysis.js";
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 ──────────────────────────────────────────────────────
// ─── MessagesDb Facade ────────────────────────────────────────────────────────
// Thin facade that delegates to domain-specific sub-modules.
export class MessagesDb {
private logger: Logger;
private crud: MessagesCrud;
private analysis: MessagesAnalysis;
private search: MessagesSearch;
private pagination: MessagesPagination;
private cleanup: MessagesCleanup;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-db");
constructor(db: NodePgDatabase<typeof schema>, _parentLogger?: Logger) {
const logger = _parentLogger ?? createChildLogger("messages-db");
this.crud = new MessagesCrud(db, logger);
this.analysis = new MessagesAnalysis(db, logger);
this.search = new MessagesSearch(db, logger);
this.pagination = new MessagesPagination(db, logger);
this.cleanup = new MessagesCleanup(db, logger);
}
// ── CRUD ──────────────────────────────────────────────────────────────
// ── 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;
}
insertMessage(message: MessageRecord): Promise<void> {
return this.crud.insertMessage(message);
}
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;
}
upsertMessageForCapture(message: MessageRecord): Promise<boolean> {
return this.crud.upsertMessageForCapture(message);
}
async updateMessageAsEdited(
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;
}
return this.crud.updateMessageAsEdited(messageId, editedContent, editedAt);
}
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;
}
updateMessageAsDeleted(messageId: string, deletedAt: number): Promise<void> {
return this.crud.updateMessageAsDeleted(messageId, deletedAt);
}
async getMessagesByChannel(
getMessagesByChannel(
channelId: string,
limit: number = 50,
offset: number = 0,
limit?: number,
offset?: number,
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;
}
return this.crud.getMessagesByChannel(channelId, limit, offset, guildId);
}
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;
}
getMessageById(messageId: string): Promise<MessageRecord | null> {
return this.crud.getMessageById(messageId);
}
// ── AI Analysis ───────────────────────────────────────────────────────
// ── AI Analysis ─────────────────────────────────────────────────────────
async updateMessageAIAnalysis(
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;
}
return this.analysis.updateMessageAIAnalysis(messageId, result);
}
async updateMessagesAIAnalysisBulk(
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;
}
return this.analysis.updateMessagesAIAnalysisBulk(updates);
}
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;
}
getPendingAIAnalysisMessages(limit?: number): Promise<MessageRecord[]> {
return this.analysis.getPendingAIAnalysisMessages(limit);
}
// ── 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: {
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;
}
return this.analysis.getConversationContextBefore(input);
}
async getPendingMessagesByConversation(
getPendingMessagesByConversation(
conversationKey: string,
limit: number = 200,
limit?: number,
): Promise<MessageRecord[]> {
this.logger.debug(
{ conversationKey, limit },
"getPendingMessagesByConversation entry",
return this.analysis.getPendingMessagesByConversation(
conversationKey,
limit,
);
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;
}
getPendingConversationKeys(limit?: number): Promise<string[]> {
return this.analysis.getPendingConversationKeys(limit);
}
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;
}
getConversationKeysWithIncompleteAnalysis(limit?: number): Promise<string[]> {
return this.analysis.getConversationKeysWithIncompleteAnalysis(limit);
}
async getIncompleteMessagesByConversation(
getIncompleteMessagesByConversation(
conversationKey: string,
limit: number = 500,
limit?: number,
): Promise<MessageRecord[]> {
this.logger.debug(
{ conversationKey, limit },
"getIncompleteMessagesByConversation entry",
return this.analysis.getIncompleteMessagesByConversation(
conversationKey,
limit,
);
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 ────────────────────────────────────────────────────────────
// ── Search ──────────────────────────────────────────────────────────────
async searchMessages(input: {
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;
}
return this.search.searchMessages(input);
}
// ── Retention / Recovery ──────────────────────────────────────────────
// ── Pagination ──────────────────────────────────────────────────────────
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;
}
listMessages(query: MessageQuery): Promise<PageResult<MessageRecord>> {
return this.pagination.listMessages(query);
}
async revertStuckProcessingMessages(
timeoutMs: number = 300000,
): Promise<number> {
this.logger.debug({ timeoutMs }, "revertStuckProcessingMessages entry");
try {
const cutoffTime = Date.now() - timeoutMs;
listReviewMessages(
query: Omit<MessageQuery, "status">,
): Promise<PageResult<MessageRecord>> {
return this.pagination.listReviewMessages(query);
}
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 });
// ── Cleanup ─────────────────────────────────────────────────────────────
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",
);
}
getExpiredMessages(retentionDays: number): Promise<MessageRecord[]> {
return this.cleanup.getExpiredMessages(retentionDays);
}
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;
}
revertStuckProcessingMessages(timeoutMs?: number): Promise<number> {
return this.cleanup.revertStuckProcessingMessages(timeoutMs);
}
}
@@ -0,0 +1,100 @@
import { decodeCursor, encodeCursor, pageResult } from "@bete/shared";
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 { messagesTable } from "../../shared/database/schema.js";
import type {
MessageQuery,
MessageRecord,
PageResult,
} from "../message-capture/types.js";
import { channelOrThreadCondition } from "./messages.crud.js";
// ─── Helpers ──────────────────────────────────────────────────────────────────
export 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;
}
const pageRows = pageResult;
// ─── MessagesPagination Class ────────────────────────────────────────────────
export class MessagesPagination {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-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 pageRows<MessageRecord>(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"],
});
}
}
@@ -0,0 +1,76 @@
import { createChildLogger, type Logger } from "@bete/shared/logger";
import { and, desc, eq, 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 type { MessageRecord } from "../message-capture/types.js";
import { channelOrThreadCondition } from "./messages.crud.js";
// ─── MessagesSearch Class ─────────────────────────────────────────────────────
export class MessagesSearch {
private logger: Logger;
constructor(
private db: NodePgDatabase<typeof schema>,
_parentLogger?: Logger,
) {
this.logger = createChildLogger("messages-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;
}
}
}
@@ -1,28 +1,11 @@
import { decodeCursor, encodeCursor, pageResult } from "@bete/shared";
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 {
@@ -126,7 +109,7 @@ export class ModerationActionsDb {
)
.limit(limit + 1);
return pageRows<ModerationAction>(rows, limit);
return pageResult<ModerationAction>(rows, limit);
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
@@ -1,21 +1,36 @@
import { createChildLogger } from "@bete/shared/logger";
const logger = createChildLogger("pagination");
export interface CursorData {
created_at: number;
id: string;
}
export function encodeCursor(data: CursorData): string {
return Buffer.from(JSON.stringify(data)).toString("base64");
const encoded = Buffer.from(JSON.stringify(data)).toString("base64");
logger.debug({ id: data.id, createdAt: data.created_at }, "Encoded cursor");
return encoded;
}
export function decodeCursor(cursor?: string): CursorData | null {
if (!cursor) return null;
if (!cursor) {
logger.debug("No cursor provided to decode");
return null;
}
try {
const data = JSON.parse(Buffer.from(cursor, "base64").toString("utf-8"));
if (typeof data.created_at === "number" && typeof data.id === "string") {
logger.debug(
{ id: data.id, createdAt: data.created_at },
"Decoded cursor",
);
return data;
}
logger.warn({ cursor }, "Decoded cursor has invalid shape");
return null;
} catch {
} catch (err) {
logger.warn({ cursor, error: String(err) }, "Failed to decode cursor");
return null;
}
}
@@ -1,28 +1,11 @@
import { decodeCursor, encodeCursor, pageResult } from "@bete/shared";
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 {
@@ -128,7 +111,7 @@ export class ReviewsDb {
)
.limit(limit + 1);
return pageRows<MessageReview>(rows, limit);
return pageResult<MessageReview>(rows, limit);
} catch (error) {
this.logger.error(
{ error: error instanceof Error ? error.message : String(error) },
@@ -1,4 +1,7 @@
import { spawn } from "node:child_process";
import { createChildLogger } from "@bete/shared/logger";
const logger = createChildLogger("ffmpeg-process");
export interface MuxFfmpegArgsOptions {
inputs: string[];
@@ -42,19 +45,24 @@ export function buildMuxFfmpegArgs(options: MuxFfmpegArgsOptions): string[] {
*/
export function runFfmpeg(args: string[]): Promise<void> {
return new Promise((resolve, reject) => {
logger.debug({ args }, "Starting ffmpeg");
const proc = spawn("ffmpeg", args, {
stdio: ["ignore", "inherit", "inherit"],
});
proc.on("close", (code) => {
if (code === 0) {
logger.debug("ffmpeg completed successfully");
resolve();
} else {
logger.warn({ exitCode: code }, "ffmpeg exited with non-zero code");
reject(new Error(`ffmpeg exited with code ${code}`));
}
});
proc.on("error", (err) => {
logger.error({ error: err.message }, "ffmpeg process error");
reject(err);
});
});
@@ -1,4 +1,7 @@
import { Transform, TransformCallback } from "node:stream";
import { createChildLogger } from "@bete/shared/logger";
const logger = createChildLogger("packet-filter");
/**
* Transform stream to filter out audio packets that are too small.
@@ -31,6 +34,11 @@ export class PacketFilter extends Transform {
callback();
}
/** Returns the number of packets filtered out and processed. */
getStats(): { filtered: number; total: number } {
return { filtered: this.filteredCount, total: this.totalCount };
}
_flush(callback: TransformCallback): void {
callback();
}
@@ -1,6 +1,9 @@
import { createChildLogger } from "@bete/shared/logger";
import { EndBehaviorType, type VoiceReceiver } from "@discordjs/voice";
import { config } from "../../../shared/config/config.js";
const logger = createChildLogger("audio-stream");
export interface AudioStreamHandlers {
onPacket: (chunk: Buffer) => void;
onEnd: () => void;
@@ -12,6 +15,8 @@ export function subscribeToAudioStream(
userId: string,
handlers: AudioStreamHandlers,
): NodeJS.ReadableStream {
logger.debug({ userId }, "Subscribing to audio stream");
const audioStream = receiver.subscribe(userId, {
end: {
behavior: EndBehaviorType.AfterSilence,
@@ -20,8 +25,14 @@ export function subscribeToAudioStream(
});
audioStream.on("data", handlers.onPacket);
audioStream.on("end", handlers.onEnd);
audioStream.on("error", handlers.onError);
audioStream.on("end", () => {
logger.debug({ userId }, "Audio stream ended");
handlers.onEnd();
});
audioStream.on("error", (error: Error) => {
logger.warn({ userId, error: error.message }, "Audio stream error");
handlers.onError(error);
});
return audioStream;
}
@@ -1,6 +1,10 @@
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import { config } from "../../../shared/config/config.js";
const logger = createChildLogger("voice-metadata");
import type {
SegmentMetadata,
SegmentState,
@@ -12,12 +16,20 @@ export async function collectUserMetadata(
userId: string,
channel: VoiceChannel,
): Promise<UserMetadata> {
logger.debug({ userId }, "Collecting user metadata");
const user =
client.users.cache.get(userId) ||
(await client.users.fetch(userId).catch(() => null));
(await client.users.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch user");
return null;
}));
const member =
channel.guild.members.cache.get(userId) ||
(await channel.guild.members.fetch(userId).catch(() => null));
(await channel.guild.members.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch guild member");
return null;
}));
const username = user?.username ?? "Unknown User";
const roles =
member?.roles.cache
@@ -1,8 +1,11 @@
import fs from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import * as prism from "prism-media";
import type { SegmentState } from "../../message-capture/types.js";
const logger = createChildLogger("segment");
export function buildSegmentPaths(
userDir: string,
startTime: number,
@@ -54,6 +57,11 @@ export class SegmentManager {
oggStream,
out,
};
logger.debug(
{ index, startTime, filename, userDir: this.userDir },
"Segment opened",
);
return this.currentSegment;
}
@@ -64,6 +72,25 @@ export class SegmentManager {
oggPacketStream.unpipe(segment.oggStream);
segment.oggStream.end();
this.currentSegment = null;
// Get file size after closing
let fileSize = 0;
try {
const stat = fs.statSync(segment.filename);
fileSize = stat.size;
} catch {
// File might not exist yet
}
logger.debug(
{
index: segment.index,
filename: segment.filename,
fileSize,
durationMs: (segment.endTime ?? 0) - segment.startTime,
},
"Segment closed",
);
return segment;
}
@@ -77,6 +104,15 @@ export class SegmentManager {
)
)
return null;
logger.debug(
{
index: this.currentSegment.index,
filename: this.currentSegment.filename,
durationMs: Date.now() - this.currentSegment.startTime,
},
"Segment rotating",
);
this.close(oggPacketStream);
return this.open(oggPacketStream);
}
@@ -1,11 +1,14 @@
import fs, { promises as fsPromises } from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { UserMetadata } from "../../message-capture/types.js";
import {
buildMuxFfmpegArgs,
runFfmpeg as defaultRunFfmpeg,
} from "../ffmpegProcess.js";
const logger = createChildLogger("recording-session");
export type SessionRecordingStatus =
| "pending"
| "completed"
@@ -86,6 +89,16 @@ export function createRecordingSession(
const participants = new Map<string, SessionParticipant>();
const segments: SessionSegmentRef[] = [];
logger.info(
{
sessionId,
guildId: options.guildId,
channelId: options.channelId,
channelName: options.channelName,
},
"Recording session created",
);
return {
sessionId,
recordingsDir: options.recordingsDir,
@@ -108,6 +121,10 @@ export function createRecordingSession(
durationMs: input.endTime - input.startTime,
offsetMs: input.startTime - options.startTime,
});
logger.debug(
{ sessionId, userId: input.user.userId, segmentCount: segments.length },
"Segment registered in session",
);
},
snapshot(endTime: number): SessionRecordingMetadata {
@@ -132,6 +149,11 @@ export function buildSessionMuxFilter(
segments: Array<{ startTime: number }>,
sessionStartTime: number,
): string {
if (segments.length === 0) {
logger.debug("Building mux filter with no segments");
return "";
}
const filters = segments.map((segment, index) => {
const delayMs = Math.max(0, segment.startTime - sessionStartTime);
return `[${index}:a]adelay=${delayMs}|${delayMs}[pad${index}]`;
@@ -140,6 +162,11 @@ export function buildSessionMuxFilter(
filters.push(
`${inputs}amix=inputs=${segments.length}:dropout_transition=0[out]`,
);
logger.debug(
{ segmentCount: segments.length, filter: filters.join(";") },
"Built mux filter",
);
return filters.join(";");
}
@@ -166,26 +193,66 @@ export async function finalizeRecordingSession(
await mkdir(sessionDir);
const metadata = session.snapshot(endTime);
logger.info(
{
sessionId: session.sessionId,
segmentCount: metadata.segments.length,
outputFile,
},
"Finalizing recording session",
);
if (metadata.segments.length === 0) {
await writeJson(metadataFile, { ...metadata, status: "empty" });
logger.info(
{ sessionId: session.sessionId },
"Recording session finalized with no segments",
);
return;
}
try {
await runFfmpeg(
buildMuxFfmpegArgs({
inputs: metadata.segments.map((segment) => segment.oggPath),
filter: buildSessionMuxFilter(metadata.segments, metadata.startTime),
output: outputFile,
codec: "libopus",
}),
const ffmpegArgs = buildMuxFfmpegArgs({
inputs: metadata.segments.map((segment) => segment.oggPath),
filter: buildSessionMuxFilter(metadata.segments, metadata.startTime),
output: outputFile,
codec: "libopus",
});
logger.debug(
{ sessionId: session.sessionId, ffmpegArgs },
"Running FFmpeg mux for session",
);
await runFfmpeg(ffmpegArgs);
// Get output file size
let outputSize = 0;
try {
const outStat = await fsPromises.stat(outputFile);
outputSize = outStat.size;
} catch {
// File might not exist yet, ignore
}
await writeJson(metadataFile, {
...metadata,
status: "completed",
outputFile,
});
logger.info(
{ sessionId: session.sessionId, outputFile, outputSize },
"Recording session finalized successfully",
);
} catch (error) {
logger.error(
{
sessionId: session.sessionId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to finalize recording session via FFmpeg",
);
await writeJson(metadataFile, {
...metadata,
status: "failed",
@@ -1,5 +1,8 @@
import { createChildLogger } from "@bete/shared/logger";
import { retryWithBackoff } from "@bete/shared/utils";
const logger = createChildLogger("tele-upload");
export interface TeleUploadResponse {
download_url: string;
public_id?: string;
@@ -40,6 +43,8 @@ export async function uploadToTele(input: {
const { buffer, filename, contentType, uploadUrl, timeoutMs, retries } =
input;
logger.debug({ filename, uploadUrl }, "Starting tele upload");
const response = await retryWithBackoff(
async () => {
const fileBlob = new Blob([new Uint8Array(buffer)], {