Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
588e750ede | ||
|
|
31e303c187 | ||
|
|
796c6390ac | ||
|
|
ecbf2617e4 | ||
|
|
9ef7d005fb | ||
|
|
842610b1af | ||
|
|
fca96396b9 | ||
|
|
ccf3fa260e |
No files matched your search
+3
-3
@@ -84,9 +84,9 @@ BACKLOG_SYNC_BATCH_SIZE=100 # Messages per backlog batch, max 100 (d
|
||||
|
||||
# === AI Analysis ===
|
||||
AI_ANALYSIS_ENABLED=false # Enable AI content moderation (default: false)
|
||||
# AI_LLM_API_KEY= # REQUIRED if AI_ANALYSIS_ENABLED=true. LLM API key
|
||||
AI_LLM_BASE_URL=http://100.121.180.82:20128/api/v1 # LLM API base URL (omniroute on imrnes; /api/v1 exposes OpenAI-compatible chat+embeddings)
|
||||
AI_LLM_MODEL=text # LLM text model name (default: text)
|
||||
AI_LLM_API_KEY= # REQUIRED if AI_ANALYSIS_ENABLED=true. LLM API key
|
||||
AI_LLM_BASE_URL=https://9router.asepharyana.my.id/v1 # LLM API base URL (9router — OpenAI-compatible router, replaces omniroute)
|
||||
AI_LLM_MODEL=claude-opus-5 # LLM text model name (default: claude-opus-5)
|
||||
# AI_LLM_VISION_MODEL= # Vision model for image analysis (falls back to AI_LLM_MODEL)
|
||||
# AI_LLM_EMBEDDING_MODEL= # Embedding model for semantic moderation cache (optional; enables near-duplicate text reuse to save LLM calls)
|
||||
# AI_LLM_EMBEDDING_MIN_SIMILARITY=0.97 # Min cosine similarity to reuse a cached verdict (default: 0.97)
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
# Spec: Perbaiki Delay Attachment 162s→<20s (GMW AI Analysis)
|
||||
|
||||
Tanggal: 2026-08-24 · Repo `~/GMW` · Service discord-gateway
|
||||
|
||||
## Evidence (audit produksi)
|
||||
|
||||
Klaster pesan attachment delay ~330–400 detik. Trace pesan `1541417073245290638` (.gif):
|
||||
19:01:08 dibuat → 19:01:09 batch incomplete → fan-out individual → **guard upload-pending
|
||||
mengembalikan `results:[]`** → diperalakukan sukses (`complete ... (undefined)`) → row
|
||||
tertahan `ai_status='processing'` **tanpa penanggung jawab** → 19:06:12 cleanup mengembalikan
|
||||
ke `pending` (tepat 300s) → baru dianalisis. Plus vision gagal 3× utk GIF besar
|
||||
("Stream ended before producing a non-ping SSE event") → degradasi teks.
|
||||
|
||||
## Root causes
|
||||
|
||||
- **A (fatal)**: `individualFallbackProcessor.processIndividualFallback` memperlakukan
|
||||
`ok:true + results:[]` sebagai sukses. Race-guard upload di `ai-analysis-worker.processIndividual`
|
||||
sengaja balik `results:[]` (desain lama) → pesan yatim `processing` sampai cleanup 300s.
|
||||
- **B**: `llmVision` hanya mencoba `stream:true`; kegagalan SSE truncation pada gambar besar
|
||||
= 3 retry sia-sia (semua jalur sama) → bukti media hilang.
|
||||
- **C**: safety-net cleanup 300s terlalu lambat sbg satu-satunya pemulih `processing`.
|
||||
|
||||
## Fix
|
||||
|
||||
1. **F1 — sinyal eksplisit upload-pending**: `IndividualOkResponse` + field opsional
|
||||
`uploadPending?: boolean`. Worker set `uploadPending:true` saat race guard kena.
|
||||
2. **F2 — processor menangani 3 kondisi** via helper murni baru
|
||||
`classifyIndividualWorkerResult(result): "success" | "upload_pending" | "incomplete" | "error"`
|
||||
(modul baru `fallbackResultClassifier.ts`, zero-dep agar mudah dites):
|
||||
- `upload_pending` → tulis ulang row ke `pending` (pola sama dgn revert apiFailed di
|
||||
batchProcessor) + broadcast + **re-schedule analisis percakapan segera**
|
||||
(dynamic import batchScheduler, pola anti-siklus yg sudah ada) → retry dalam ~250ms
|
||||
begitu upload beres. Bukan error, tidak naikkan CB counter.
|
||||
- `incomplete` (flags analysis_incomplete) → perilaku lama (exhausted path).
|
||||
- `error` / `results kosong tanpa penjelasan` → throw transien (retry oleh recovery),
|
||||
BUKAN sukses palsu. Log "(undefined)" hilang.
|
||||
3. **F3 — vision non-stream fallback**: di `llmVision`, jika error match
|
||||
`/Stream ended before producing a non-ping SSE|stream ended/i` → coba SEKALI lagi dengan
|
||||
`stream:false` (router agregasi penuh; timeout tetap 60s). Konversi hard-fail jadi sukses.
|
||||
4. **F4 — turunkan safety net**: default `revertStuckProcessingMessages` 300000 → 120000 ms.
|
||||
|
||||
## File disentuh
|
||||
|
||||
- `src/modules/ai-moderation/fallbackResultClassifier.ts` (BARU, pure)
|
||||
- `src/modules/ai-moderation/ai-analysis-worker.ts` (tipe + set flag uploadPending)
|
||||
- `src/modules/ai-moderation/individualFallbackProcessor.ts` (konsumsi classifier + reschedule)
|
||||
- `src/modules/ai-moderation/llmClient.ts` (fallback non-stream di llmVision)
|
||||
- `src/modules/message-capture/messagesCleanup.ts` (default 120s)
|
||||
|
||||
## Verifikasi
|
||||
|
||||
- Test baru `tests/fallbackResultClassifier.test.ts` (4 klasifikasi + edge kosong).
|
||||
- Gate: tsc --noEmit, biome error-level, vitest run semua hijau.
|
||||
- Deploy GHA sukses; pasca-deploy: pesan attachment baru p50 < 20s
|
||||
(`SELECT percentile_cont(0.5) ... WHERE metadata attachments>0 AND created_at > deploy`),
|
||||
tidak ada lagi "complete ... (undefined)".
|
||||
@@ -93,6 +93,12 @@ type BatchOkResponse = {
|
||||
ok: true;
|
||||
conversationKey: string;
|
||||
rows: MessageRecord[];
|
||||
/**
|
||||
* Race-guard signal (2026-08-25): target ids whose attachment upload is
|
||||
* still in-flight — NO analysis ran for them. The processor must defer
|
||||
* these (requeue + poll), never fan them out as failures.
|
||||
*/
|
||||
uploadPendingIds?: string[];
|
||||
};
|
||||
type BatchErrorResponse = {
|
||||
ok: false;
|
||||
@@ -100,7 +106,16 @@ type BatchErrorResponse = {
|
||||
rows: MessageRecord[];
|
||||
error: string;
|
||||
};
|
||||
type IndividualOkResponse = { ok: true; results: AnalysisResult[] };
|
||||
type IndividualOkResponse = {
|
||||
ok: true;
|
||||
results: AnalysisResult[];
|
||||
/**
|
||||
* Race-guard signal (2026-08-24): the message's attachment upload is still
|
||||
* in-flight — NO analysis ran. The processor must re-queue the message as
|
||||
* `pending` and re-schedule, never treat this as a completed moderation.
|
||||
*/
|
||||
uploadPending?: boolean;
|
||||
};
|
||||
type IndividualErrorResponse = {
|
||||
ok: false;
|
||||
results: AnalysisResult[];
|
||||
@@ -310,7 +325,16 @@ async function processBatch(job: {
|
||||
? messages
|
||||
: messages.filter((m) => !pendingUploadTargetIds.has(m.id));
|
||||
if (readyMessages.length === 0) {
|
||||
return { ok: true, conversationKey, rows: [] };
|
||||
// Explicit signal (2026-08-25): every target is still upload-pending.
|
||||
// Returning bare {ok:true, rows:[]} made the processor classify all of
|
||||
// them "incomplete" and fan out to the individual queue — a hot ~300ms
|
||||
// requeue loop for the whole upload duration.
|
||||
return {
|
||||
ok: true,
|
||||
conversationKey,
|
||||
rows: [],
|
||||
uploadPendingIds: messages.map((m) => m.id),
|
||||
};
|
||||
}
|
||||
|
||||
// The orchestrator handles text/media split + caching + parallel paths
|
||||
@@ -415,7 +439,7 @@ async function processIndividual(job: {
|
||||
(a) => a.message_id === message.id && a.upload_status === "pending",
|
||||
);
|
||||
if (uploadStillPending) {
|
||||
return { ok: true, results: [] };
|
||||
return { ok: true, results: [], uploadPending: true };
|
||||
}
|
||||
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
/**
|
||||
* batchOutcomeClassifier.ts
|
||||
*
|
||||
* Pure partitioner of the batch worker response (2026-08-25).
|
||||
*
|
||||
* Bug history: the batch race guard returned `{ok:true, rows:[]}` when every
|
||||
* target's attachment upload was still in-flight. The processor classified all
|
||||
* of them as "incomplete" and fanned out to the individual queue, where the
|
||||
* guard there requeued + rescheduled at the 250ms debounce — a hot ~300ms loop
|
||||
* for the entire upload duration (~10 cycles in 3s in prod logs). Root fix:
|
||||
* the worker now reports `uploadPendingIds` explicitly and this pure function
|
||||
* partitions the outcome so upload-pending targets NEVER enter the fanout.
|
||||
*/
|
||||
|
||||
export interface BatchRowLike {
|
||||
id?: string;
|
||||
ai_status?: string | null;
|
||||
ai_moderation_flags?: string | null;
|
||||
}
|
||||
|
||||
export interface BatchWorkerResponseLike {
|
||||
ok?: boolean;
|
||||
rows?: BatchRowLike[];
|
||||
/** Explicit race-guard signal from the worker (2026-08-25). */
|
||||
uploadPendingIds?: string[];
|
||||
error?: string;
|
||||
}
|
||||
|
||||
/** One target's per-message disposition after a batch attempt. */
|
||||
export type BatchTargetKind =
|
||||
| "completed"
|
||||
| "upload_pending"
|
||||
| "incomplete"
|
||||
| "parse_failed"
|
||||
| "api_failed";
|
||||
|
||||
function flagsOf(row: { ai_moderation_flags?: string | null }): string[] {
|
||||
if (!row.ai_moderation_flags) return [];
|
||||
try {
|
||||
const parsed = JSON.parse(row.ai_moderation_flags) as unknown;
|
||||
return Array.isArray(parsed) ? (parsed as string[]) : [];
|
||||
} catch {
|
||||
return [] as string[];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Partition the input message ids into per-message dispositions for one batch
|
||||
* worker response. Pure: no DB/Piscina/logger — unit-testable directly.
|
||||
*
|
||||
* Priority per id: explicit uploadPendingIds → completed row → flag-based
|
||||
* failure kinds → unexplained missing (treated like incomplete).
|
||||
*/
|
||||
export function partitionBatchOutcome(
|
||||
messages: ReadonlyArray<{ id: string }>,
|
||||
response: BatchWorkerResponseLike,
|
||||
): Map<string, BatchTargetKind> {
|
||||
const pendingSet = new Set(response.uploadPendingIds ?? []);
|
||||
const rowsById = new Map(
|
||||
(response.rows ?? [])
|
||||
.filter((r): r is BatchRowLike & { id: string } => Boolean(r?.id))
|
||||
.map((r) => [r.id, r]),
|
||||
);
|
||||
|
||||
const out = new Map<string, BatchTargetKind>();
|
||||
for (const msg of messages) {
|
||||
if (pendingSet.has(msg.id)) {
|
||||
out.set(msg.id, "upload_pending");
|
||||
continue;
|
||||
}
|
||||
const row = rowsById.get(msg.id);
|
||||
if (!row) {
|
||||
// Unexplained drop: LLM silently omitted it. Same retryable bucket as
|
||||
// analysis_incomplete — never a silent success.
|
||||
out.set(msg.id, "incomplete");
|
||||
continue;
|
||||
}
|
||||
if (row.ai_status !== "error") {
|
||||
out.set(msg.id, "completed");
|
||||
continue;
|
||||
}
|
||||
const flags = flagsOf(row);
|
||||
if (flags.includes("analysis_incomplete")) {
|
||||
out.set(msg.id, "incomplete");
|
||||
} else if (flags.includes("analysis_parse_failed")) {
|
||||
out.set(msg.id, "parse_failed");
|
||||
} else if (flags.includes("analysis_api_failed")) {
|
||||
out.set(msg.id, "api_failed");
|
||||
} else {
|
||||
out.set(msg.id, "incomplete");
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Linear backoff ramp for consecutive upload-pending polls:
|
||||
* poll N (1-based) waits min(base × N, cap). Keeps latency low for fast
|
||||
* uploads while bounding total polling cost for long uploads.
|
||||
*/
|
||||
export function computeUploadPollDelayMs(
|
||||
consecutivePolls: number,
|
||||
baseMs: number,
|
||||
capMs: number,
|
||||
): number {
|
||||
const n = Math.max(1, Math.floor(consecutivePolls));
|
||||
return Math.min(Math.round(baseMs * n), Math.round(capMs));
|
||||
}
|
||||
@@ -4,6 +4,10 @@ import { isAgeRestrictedMetadata } from "../message-capture/messageMetadata.js";
|
||||
import { messageStore } from "../message-capture/messageStore.js";
|
||||
import type { MessageRecord } from "../message-capture/types.js";
|
||||
import { pickBatchWithinBudget as pickBatchWithinBudgetPure } from "./batchBudget.js";
|
||||
import {
|
||||
computeUploadPollDelayMs,
|
||||
partitionBatchOutcome,
|
||||
} from "./batchOutcomeClassifier.js";
|
||||
import { workerPool } from "./circuitBreaker.js";
|
||||
import { estimateTokens } from "./conversationContext.js";
|
||||
import {
|
||||
@@ -21,11 +25,20 @@ import {
|
||||
|
||||
const logger = createChildLogger("batch-processor");
|
||||
|
||||
/**
|
||||
* Consecutive upload-pending poll counter per conversation (2026-08-25).
|
||||
* Drives the linear backoff ramp while attachments are still uploading;
|
||||
* cleared as soon as a batch comes back with no upload-pending targets.
|
||||
*/
|
||||
const conversationUploadPolls = new Map<string, number>();
|
||||
|
||||
export interface AnalysisWorkerResponse {
|
||||
ok: boolean;
|
||||
conversationKey: string;
|
||||
rows: MessageRecord[];
|
||||
error?: string;
|
||||
/** Explicit upload-in-flight signal from the batch race guard (2026-08-25). */
|
||||
uploadPendingIds?: string[];
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -141,6 +154,8 @@ export async function processBatch(
|
||||
|
||||
activeRequests++;
|
||||
let shouldScheduleNext = false;
|
||||
/** Set when upload-pending targets defer the next cycle by this many ms. */
|
||||
let deferredUploadRescheduleMs: number | null = null;
|
||||
try {
|
||||
const result = (await workerPool.run({
|
||||
type: "batch",
|
||||
@@ -199,37 +214,85 @@ export async function processBatch(
|
||||
return;
|
||||
}
|
||||
|
||||
// Batch succeeded -- check for messages the LLM silently dropped or failed
|
||||
const incompleteMessages: MessageRecord[] = [];
|
||||
const parseFailedMessages: MessageRecord[] = [];
|
||||
// Batch succeeded -- partition per-message outcome explicitly (2026-08-25).
|
||||
// upload_pending targets are DEFERRED (never fanned out): the old code
|
||||
// treated them as incomplete -> individual queue -> requeue+250ms
|
||||
// reschedule -> hot ~300ms loop for the whole upload duration.
|
||||
const outcomeById = partitionBatchOutcome(messages, result);
|
||||
const messagesForIndividualQueue: MessageRecord[] = [];
|
||||
const apiFailedMessages: MessageRecord[] = [];
|
||||
const uploadPendingMessages: 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")) {
|
||||
switch (outcomeById.get(msg.id)) {
|
||||
case "upload_pending":
|
||||
uploadPendingMessages.push(msg);
|
||||
break;
|
||||
case "api_failed":
|
||||
// Preserve the dedicated api-failure semantics below: revert +
|
||||
// conversation cooldown instead of an immediate individual retry.
|
||||
apiFailedMessages.push(msg);
|
||||
}
|
||||
break;
|
||||
default:
|
||||
// incomplete / parse_failed / unexplained drops stay retryable via
|
||||
// the individual fallback queue (same semantics as before).
|
||||
messagesForIndividualQueue.push(msg);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const messagesForIndividualQueue = [
|
||||
...incompleteMessages,
|
||||
...parseFailedMessages,
|
||||
];
|
||||
if (uploadPendingMessages.length > 0) {
|
||||
const polls = (conversationUploadPolls.get(conversationKey) ?? 0) + 1;
|
||||
conversationUploadPolls.set(conversationKey, polls);
|
||||
const delayMs = computeUploadPollDelayMs(
|
||||
polls,
|
||||
config.AI_ANALYSIS_UPLOAD_POLL_MS,
|
||||
config.AI_ANALYSIS_MAX_UPLOAD_POLL_MS,
|
||||
);
|
||||
logger.debug(
|
||||
{
|
||||
conversationKey,
|
||||
count: uploadPendingMessages.length,
|
||||
ids: uploadPendingMessages.map((m) => m.id),
|
||||
pollAttempt: polls,
|
||||
delayMs,
|
||||
},
|
||||
"Attachment upload in-flight for batch targets — deferring with poll backoff",
|
||||
);
|
||||
|
||||
// Put the rows back to `pending` so the scheduler owns them again.
|
||||
await messageStore
|
||||
.updateMessagesAIAnalysisBulk(
|
||||
uploadPendingMessages.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: unknown) => {
|
||||
logger.error(
|
||||
{ error: String(err), ids: uploadPendingMessages.map((m) => m.id) },
|
||||
"Failed to revert upload-pending batch targets to pending",
|
||||
);
|
||||
return [] as MessageRecord[];
|
||||
});
|
||||
|
||||
// Poll backoff instead of the 250ms debounce: the finally-block
|
||||
// schedules the next cycle after this delay instead of immediately.
|
||||
deferredUploadRescheduleMs = delayMs;
|
||||
} else {
|
||||
conversationUploadPolls.delete(conversationKey);
|
||||
}
|
||||
|
||||
if (messagesForIndividualQueue.length > 0) {
|
||||
logger.warn(
|
||||
@@ -307,7 +370,11 @@ export async function processBatch(
|
||||
resetConversationBatchFailures(conversationKey);
|
||||
conversationErrorCooldown.delete(conversationKey);
|
||||
}
|
||||
shouldScheduleNext = true;
|
||||
// Upload-pending defer owns the next-cycle timing; don't let the default
|
||||
// immediate schedule override it.
|
||||
if (deferredUploadRescheduleMs === null) {
|
||||
shouldScheduleNext = true;
|
||||
}
|
||||
} catch (error) {
|
||||
recordConversationBatchFailure(conversationKey);
|
||||
|
||||
@@ -344,7 +411,17 @@ export async function processBatch(
|
||||
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
|
||||
conversationProcessing.delete(conversationKey);
|
||||
}
|
||||
if (shouldScheduleNext) {
|
||||
if (deferredUploadRescheduleMs !== null) {
|
||||
// Upload still in-flight: re-schedule after the backoff delay instead of
|
||||
// immediately (the old path hot-looped at ~250-300ms per cycle).
|
||||
const delayMs = deferredUploadRescheduleMs;
|
||||
setTimeout(() => {
|
||||
// Dynamic import to avoid circular dependency at module scope
|
||||
import("./batchScheduler.js").then((m) =>
|
||||
m.scheduleConversationAnalysis(conversationKey),
|
||||
);
|
||||
}, delayMs).unref();
|
||||
} else if (shouldScheduleNext) {
|
||||
setImmediate(() => {
|
||||
// Dynamic import to avoid circular dependency at module scope
|
||||
import("./batchScheduler.js").then((m) =>
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
/**
|
||||
* fallbackResultClassifier.ts
|
||||
*
|
||||
* Pure classifier for the individual-fallback worker response.
|
||||
*
|
||||
* Bug history (2026-08-24): the worker's upload-pending race guard returned
|
||||
* `{ ok: true, results: [] }` (a legacy "no results yet" signal), but the
|
||||
* processor treated ANY `ok:true` as a successful moderation. Empty results
|
||||
* meant nothing was written to the DB — the message stayed stuck in
|
||||
* `ai_status='processing'` with nobody watching it until the 300s cleanup
|
||||
* reverted it. That single gap produced the ~330-400s attachment delay
|
||||
* cluster. Classification now happens in ONE pure function so every outcome
|
||||
* has an explicit, testable owner.
|
||||
*/
|
||||
|
||||
export type WorkerResultKind =
|
||||
| "success"
|
||||
| "upload_pending"
|
||||
| "incomplete"
|
||||
| "error";
|
||||
|
||||
export interface ClassifiableWorkerResult {
|
||||
ok?: boolean;
|
||||
/** Upload-pending marker set by ai-analysis-worker's race guard. */
|
||||
uploadPending?: boolean;
|
||||
results?: Array<{ status?: string; flags?: string[] | string } | undefined>;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
function flagsOf(r: { flags?: string[] | string }): string[] {
|
||||
if (!r.flags) return [];
|
||||
if (Array.isArray(r.flags)) return r.flags;
|
||||
try {
|
||||
const parsed = JSON.parse(r.flags) as unknown;
|
||||
return Array.isArray(parsed) ? (parsed as string[]) : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify an individual-fallback worker response:
|
||||
* - "upload_pending": explicit race-guard signal — retry shortly, NOT an error.
|
||||
* - "success": at least one result and none is analysis_incomplete.
|
||||
* - "incomplete": LLM ran but dropped/failed this message after retries
|
||||
* (analysis_incomplete flag) — terminal exhausted path.
|
||||
* - "error": anything else (ok:false, or ok:true with NO explainable
|
||||
* results). The old code silently succeeded here — never again.
|
||||
*/
|
||||
export function classifyIndividualWorkerResult(
|
||||
result: ClassifiableWorkerResult,
|
||||
): WorkerResultKind {
|
||||
if (result.uploadPending === true) return "upload_pending";
|
||||
const results = (result.results ?? []).filter(
|
||||
(r): r is NonNullable<typeof r> => Boolean(r),
|
||||
);
|
||||
if (results.length === 0) return "error";
|
||||
if (result.ok !== true) return "error";
|
||||
for (const r of results) {
|
||||
const flags = flagsOf(r);
|
||||
if (flags.includes("analysis_incomplete")) return "incomplete";
|
||||
if ((r.status ?? "") === "") return "error";
|
||||
}
|
||||
return "success";
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import type {
|
||||
} from "../message-capture/types.js";
|
||||
import { getConversationKey, workerPool } from "./circuitBreaker.js";
|
||||
import { fireAlert } from "./conversationState.js";
|
||||
import { classifyIndividualWorkerResult } from "./fallbackResultClassifier.js";
|
||||
import {
|
||||
broadcastAnalysisCompleted,
|
||||
LAST_ERROR,
|
||||
@@ -78,26 +79,82 @@ async function processIndividualFallback(
|
||||
message,
|
||||
skipNormalAnalysis: false,
|
||||
} as unknown)) as
|
||||
| { ok: true; results: AnalysisResult[] }
|
||||
| { ok: true; results: AnalysisResult[]; uploadPending?: boolean }
|
||||
| { ok: false; results: AnalysisResult[]; error: string };
|
||||
|
||||
// Explicit outcome classification (2026-08-24): the old code treated any
|
||||
// ok:true as a completed moderation, so the upload-pending race guard's
|
||||
// empty results left messages stuck in `processing` until the 300s
|
||||
// cleanup reverted them — the root cause of the ~330s attachment delays.
|
||||
const kind = classifyIndividualWorkerResult(workerResult);
|
||||
|
||||
if (kind === "upload_pending") {
|
||||
// Attachment still uploading — put the row back to `pending` and
|
||||
// re-schedule this conversation immediately. The next scheduler cycle
|
||||
// (~debounce 250ms) re-fetches; once upload_status flips to done the
|
||||
// race guard passes and analysis proceeds. NOT an error: never touches
|
||||
// the circuit breaker counters.
|
||||
const revertedRows = await messageStore
|
||||
.updateMessagesAIAnalysisBulk([
|
||||
{
|
||||
messageId,
|
||||
result: {
|
||||
status: "pending",
|
||||
flags: null,
|
||||
score: null,
|
||||
analysis: null,
|
||||
categories: null,
|
||||
severity: null,
|
||||
confidence: null,
|
||||
recommendedAction: null,
|
||||
analyzedAt: null,
|
||||
error: null,
|
||||
},
|
||||
},
|
||||
])
|
||||
.catch((dbErr: unknown) => {
|
||||
logger.error(
|
||||
{ messageId, error: String(dbErr) },
|
||||
"Failed to revert upload-pending message to pending",
|
||||
);
|
||||
return [] as MessageRecord[];
|
||||
});
|
||||
for (const row of revertedRows) {
|
||||
broadcastAnalysisCompleted(row);
|
||||
}
|
||||
logger.debug(
|
||||
{ messageId, conversationKey },
|
||||
"Individual fallback: attachment upload in-flight — requeued as pending + rescheduled",
|
||||
);
|
||||
setImmediate(() => {
|
||||
import("./batchScheduler.js")
|
||||
.then((m) => m.scheduleConversationAnalysis(conversationKey))
|
||||
.catch(() => {});
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
let analysisResult: { results: AnalysisResult[] } | null = null;
|
||||
|
||||
if (workerResult.ok) {
|
||||
const stillIncomplete = workerResult.results.some((r) =>
|
||||
r.flags.includes("analysis_incomplete"),
|
||||
if (kind === "success") {
|
||||
analysisResult = workerResult;
|
||||
} else if (kind === "incomplete") {
|
||||
exhaustedOnIncomplete = true;
|
||||
analysisResult = null;
|
||||
} else {
|
||||
// "error" — includes ok:true with unexplainable empty results (the old
|
||||
// silent-success bug). Throw so it is treated as a transient failure.
|
||||
throw new Error(
|
||||
(workerResult as { error?: string }).error ??
|
||||
"Individual worker returned no explainable results",
|
||||
);
|
||||
if (stillIncomplete) {
|
||||
exhaustedOnIncomplete = true;
|
||||
analysisResult = null;
|
||||
} else {
|
||||
analysisResult = workerResult;
|
||||
}
|
||||
}
|
||||
|
||||
// No heuristic fallback: an incomplete/errored LLM result stays a
|
||||
// retryable error — the recovery worker picks it up later. Producing a
|
||||
// regex/wordlist verdict here would reintroduce false positives.
|
||||
// (incomplete keeps its exhausted flag so the catch writes the terminal
|
||||
// individual_analysis_exhausted status.)
|
||||
if (!analysisResult) {
|
||||
throw new Error(`LLM analysis failed for message ${messageId}`);
|
||||
}
|
||||
|
||||
@@ -83,7 +83,7 @@ export async function callModerationLLM(
|
||||
jsonResponse: { type: "json_object" },
|
||||
retries: 0,
|
||||
signal,
|
||||
// Router (9router/omniroute) always streams SSE even when the
|
||||
// Router (9router / formerly omniroute) always streams SSE even when the
|
||||
// request omits `stream`. In non-stream mode the OpenAI SDK waits
|
||||
// for the FULL body before parsing, so slow/long upstream streams
|
||||
// hit the 30s/60s timeout and abort mid-generation. Streaming mode
|
||||
|
||||
@@ -356,10 +356,10 @@ export async function llmVision(
|
||||
promptText: string,
|
||||
imageUrl: { url: string },
|
||||
): Promise<string | null> {
|
||||
const completion = await llmChat({
|
||||
const params = {
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
role: "user" as const,
|
||||
content: [
|
||||
{ type: "text" as const, text: promptText },
|
||||
{ type: "image_url" as const, image_url: imageUrl },
|
||||
@@ -371,9 +371,26 @@ export async function llmVision(
|
||||
temperature: 0.1,
|
||||
top_p: 0.9,
|
||||
retries: 0,
|
||||
stream: true, // router always streams SSE; non-stream waits for full body and times out
|
||||
timeout: config.AI_LLM_VISION_ANALYSIS_TIMEOUT_MS ?? 60_000,
|
||||
});
|
||||
};
|
||||
|
||||
// Streaming first (the router always streams SSE; a non-stream request
|
||||
// waits for the full body and times out on slow models). Fallback (2026-08-24):
|
||||
// large GIFs/images sometimes get their SSE stream truncated mid-flight by
|
||||
// the upstream ("Stream ended before producing a non-ping SSE event") — all
|
||||
// streaming retries fail identically, so retry ONCE with stream:false where
|
||||
// the router assembles the complete response server-side.
|
||||
let completion: Awaited<ReturnType<typeof llmChat>>;
|
||||
try {
|
||||
completion = await llmChat({ ...params, stream: true });
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
if (/stream ended before producing a non-ping sse/i.test(msg)) {
|
||||
completion = await llmChat({ ...params, stream: false });
|
||||
} else {
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
if (!completion) return null;
|
||||
return completion.choices[0]?.message?.content?.trim() ?? null;
|
||||
|
||||
@@ -4,6 +4,8 @@
|
||||
* Orchestrates LLM-based moderation analysis — manages batch splitting,
|
||||
* parallel text+media analysis, LLM calls with retry, and cache handling.
|
||||
*/
|
||||
|
||||
import { LRUCache } from "lru-cache";
|
||||
import { createChildLogger } from "@/shared/logger/index";
|
||||
import { config } from "../../shared/config/config.js";
|
||||
import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js";
|
||||
@@ -31,10 +33,18 @@ import {
|
||||
parseQdrantVerdict,
|
||||
type StoredModerationVerdict,
|
||||
setCachedTextModeration,
|
||||
upsertBareKeyToQdrant,
|
||||
} from "./textCacheStore.js";
|
||||
|
||||
const log = createChildLogger("moderationOrchestrator");
|
||||
|
||||
/**
|
||||
* Bare keys already written this process (dual-key write-back dedupe).
|
||||
* LRU-bounded so a long-lived gateway can't grow it without limit; the DB
|
||||
* upsert underneath is idempotent anyway — this just avoids redundant writes.
|
||||
*/
|
||||
const globalBareKeysWritten = new LRUCache<string, true>({ max: 5000 });
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -309,6 +319,7 @@ export async function runModerationAnalysis(
|
||||
};
|
||||
cacheHits.push(hit);
|
||||
hitByKey.set(cacheKey, hit);
|
||||
servedCacheKeys.add(cacheKey); // bump hit_count for metrics
|
||||
logCacheEvent("hit", cacheKey, "text");
|
||||
}
|
||||
} else {
|
||||
@@ -347,6 +358,7 @@ export async function runModerationAnalysis(
|
||||
};
|
||||
cacheHits.push(hit);
|
||||
hitByKey.set(cacheKey, hit);
|
||||
servedCacheKeys.add(cacheKey); // bump hit_count for metrics
|
||||
logCacheEvent("hit", cacheKey, "text");
|
||||
}
|
||||
}
|
||||
@@ -374,6 +386,7 @@ export async function runModerationAnalysis(
|
||||
cacheHits: cacheHits.length,
|
||||
uncached: uncachedTargets.length,
|
||||
total: targets.length,
|
||||
servedKeys: servedCacheKeys.size,
|
||||
},
|
||||
"User moderation cache applied",
|
||||
);
|
||||
@@ -434,20 +447,59 @@ export async function runModerationAnalysis(
|
||||
rawContent,
|
||||
makeModerationContextKey(target),
|
||||
);
|
||||
const stored = {
|
||||
flags: result.flags ?? [],
|
||||
score: result.score ?? 0,
|
||||
analysis: result.analysis ?? "",
|
||||
categories: result.categories ?? result.flags ?? [],
|
||||
severity: result.severity ?? "none",
|
||||
confidence: result.confidence ?? result.score ?? 0,
|
||||
recommendedAction: result.recommendedAction ?? "none",
|
||||
status: result.status,
|
||||
};
|
||||
setCachedTextModeration(
|
||||
cacheKey,
|
||||
{
|
||||
flags: result.flags ?? [],
|
||||
score: result.score ?? 0,
|
||||
analysis: result.analysis ?? "",
|
||||
categories: result.categories ?? result.flags ?? [],
|
||||
severity: result.severity ?? "none",
|
||||
confidence: result.confidence ?? result.score ?? 0,
|
||||
recommendedAction: result.recommendedAction ?? "none",
|
||||
status: result.status,
|
||||
},
|
||||
stored,
|
||||
embeddingsByKey.get(cacheKey),
|
||||
).catch(() => {});
|
||||
|
||||
// Dual-key write-back (2026-08-24): the FIRST analysis of a message runs
|
||||
// WITH conversation context (accurate), but its verdict is also stored
|
||||
// under the context-free bare key so repeats in OTHER channels hit the
|
||||
// exact cache instead of paying a new LLM call. Same guard as the read
|
||||
// path — only non-actionable clean verdicts may cross channels.
|
||||
//
|
||||
// 2026-08-25 cache-hit fix: the bare key is ALSO upserted to Qdrant
|
||||
// (via upsertBareKeyToQdrant) with the SAME embedding already computed
|
||||
// at lookup time. Previously the bare key was only PG-written with
|
||||
// embedding=null — bare clean verdicts were DB-only and invisible to
|
||||
// searchQdrantBatch, capping the semantic hit-rate below the exact-cache
|
||||
// hit-rate for cross-channel repeats.
|
||||
const bareKey = makeTextModerationCacheKey(rawContent);
|
||||
if (
|
||||
bareKey !== cacheKey &&
|
||||
!globalBareKeysWritten.has(bareKey) &&
|
||||
isGloballyReusableCleanVerdict(
|
||||
{
|
||||
status: stored.status,
|
||||
flags: stored.flags,
|
||||
score: stored.score,
|
||||
analysis: stored.analysis,
|
||||
categories: stored.categories,
|
||||
severity: stored.severity,
|
||||
confidence: stored.confidence,
|
||||
recommendedAction: stored.recommendedAction,
|
||||
},
|
||||
undefined,
|
||||
)
|
||||
) {
|
||||
globalBareKeysWritten.set(bareKey, true);
|
||||
setCachedTextModeration(bareKey, stored, null).catch(() => {});
|
||||
const bareEmbedding = embeddingsByKey.get(cacheKey);
|
||||
if (bareEmbedding && bareEmbedding.length > 0) {
|
||||
upsertBareKeyToQdrant(bareKey, stored, bareEmbedding).catch(() => {});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const allResults = [
|
||||
|
||||
@@ -524,7 +524,7 @@ export function isSemanticBandAccepted(
|
||||
* globally reused — enforcement is context-sensitive by design.
|
||||
*/
|
||||
export function isGloballyReusableCleanVerdict(
|
||||
verdict: StoredModerationVerdict,
|
||||
verdict: Omit<StoredModerationVerdict, "status"> & { status: string },
|
||||
analyzedAtMs: number | undefined,
|
||||
): boolean {
|
||||
if (verdict.status !== "clean") return false;
|
||||
@@ -664,6 +664,55 @@ export async function findSimilarTextModeration(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Upsert a bare (context-free) clean verdict to the Qdrant vector store,
|
||||
* making global-reuse clean verdicts discoverable by semantic search.
|
||||
*
|
||||
* Why: the main `setCachedTextModeration` writes bare-key rows to Postgres
|
||||
* with embedding=null (deliberate — no duplicate PG embedding column), but
|
||||
* a bare clean verdict that never reaches Qdrant is invisible to
|
||||
* searchQdrantBatch. So two messages with identical clean content in
|
||||
* DIFFERENT channels never match semantically — the semantic hit-rate is
|
||||
* capped below the exact-cache hit-rate. This helper shares the embedding
|
||||
* already computed at lookup time so the bare point is semantically
|
||||
* findable.
|
||||
*
|
||||
* Guard: only non-actionable clean verdicts qualify (same guard as the
|
||||
* read path and as the orchestrator's bare-key write-back). No-op when
|
||||
* Qdrant is disabled or no embedding is available.
|
||||
*/
|
||||
export async function upsertBareKeyToQdrant(
|
||||
bareKey: string,
|
||||
result: {
|
||||
status: string;
|
||||
flags: string[];
|
||||
score: number;
|
||||
analysis: string;
|
||||
categories: string[];
|
||||
severity: string;
|
||||
confidence: number;
|
||||
recommendedAction: string;
|
||||
},
|
||||
embedding: number[] | null | undefined,
|
||||
): Promise<void> {
|
||||
if (!isQdrantConfigured() || !embedding || embedding.length === 0) return;
|
||||
if (!isGloballyReusableCleanVerdict(result, undefined)) return;
|
||||
const now = Date.now();
|
||||
const USER_MOD_CACHE_TTL_MS = 24 * 60 * 60 * 1000;
|
||||
await upsertQdrantPoint(bareKey, embedding, {
|
||||
text: bareKey,
|
||||
flags: JSON.stringify(result),
|
||||
analyzed_at: now,
|
||||
expires_at: now + USER_MOD_CACHE_TTL_MS,
|
||||
content_hash: bareKey.split(":").pop() ?? "",
|
||||
}).catch((err: unknown) => {
|
||||
logger.error(
|
||||
{ error: err instanceof Error ? err.message : String(err), bareKey },
|
||||
"Failed to upsert bare-key clean verdict to Qdrant",
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a moderation result for a (user, content) pair.
|
||||
* The `flags` field stores the full result object as JSON.
|
||||
|
||||
@@ -48,7 +48,13 @@ export class MessagesCleanup {
|
||||
}
|
||||
|
||||
async revertStuckProcessingMessages(
|
||||
timeoutMs: number = 300000,
|
||||
// 2026-08-24: lowered from 300000 — this cleanup is the last-resort
|
||||
// recoverer for rows stuck in `processing`. With the upload-pending
|
||||
// race-guard now requeueing properly (fallbackResultClassifier), any row
|
||||
// that still sits here for >2min is a genuine leak; reverting sooner
|
||||
// bounds the worst-case delay without racing legitimate in-flight work
|
||||
// (media batches can legitimately take ~60s+).
|
||||
timeoutMs: number = 120000,
|
||||
): Promise<number> {
|
||||
this.logger.debug({ timeoutMs }, "revertStuckProcessingMessages entry");
|
||||
try {
|
||||
|
||||
@@ -149,8 +149,8 @@ export const configSchema = z
|
||||
AI_LLM_BASE_URL: z
|
||||
.string()
|
||||
.url()
|
||||
.default("https://omniroute.imrnes.team/v1"),
|
||||
AI_LLM_MODEL: z.string().default("text"),
|
||||
.default("https://9router.asepharyana.my.id/v1"),
|
||||
AI_LLM_MODEL: z.string().default("claude-opus-5"),
|
||||
// Vision uses the SAME router/base URL as text moderation
|
||||
// (AI_LLM_BASE_URL) but a different model alias. The dedicated NVIDIA
|
||||
// multimodal endpoint was removed.
|
||||
@@ -182,7 +182,7 @@ export const configSchema = z
|
||||
.number()
|
||||
.int()
|
||||
.positive()
|
||||
.default(30),
|
||||
.default(50),
|
||||
// Qdrant vector store for the semantic moderation cache. When
|
||||
// QDRANT_URL is set, embeddings are stored/searched there (Postgres
|
||||
// embedding column remains as a legacy fallback).
|
||||
@@ -254,6 +254,12 @@ export const configSchema = z
|
||||
.positive()
|
||||
.default(10000),
|
||||
AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000),
|
||||
// Upload-pending batch poll (2026-08-25): when a batch is deferred because
|
||||
// attachments are still uploading, the processor re-schedules with this
|
||||
// base delay (linear ramp per consecutive poll, capped) instead of the
|
||||
// 250ms debounce — the old path hot-looped ~300ms for the whole upload.
|
||||
AI_ANALYSIS_UPLOAD_POLL_MS: z.coerce.number().positive().default(1500),
|
||||
AI_ANALYSIS_MAX_UPLOAD_POLL_MS: z.coerce.number().positive().default(8000),
|
||||
|
||||
// ── AI Analysis Batch ───────────────────────────────────────────────
|
||||
AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(200),
|
||||
@@ -270,7 +276,7 @@ export const configSchema = z
|
||||
.number()
|
||||
.int()
|
||||
.positive()
|
||||
.default(72),
|
||||
.default(120),
|
||||
AI_ANALYSIS_MAX_CONTEXT_TOKENS: z.coerce.number().positive().default(8000),
|
||||
AI_ANALYSIS_MAX_TARGET_TOKENS: z.coerce.number().positive().default(14000),
|
||||
AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT: z.coerce
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// partitionBatchOutcome — upload-pending defer vs fanout (2026-08-25)
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// Bug history: the batch worker's race guard returned {ok:true, rows:[]} while
|
||||
// attachments were still uploading; every target was classified "incomplete",
|
||||
// fanned out to the individual queue, requeued there, rescheduled at 250ms —
|
||||
// a hot ~300ms loop for the whole upload duration (~10 cycles in 3s in prod).
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
computeUploadPollDelayMs,
|
||||
partitionBatchOutcome,
|
||||
} from "../src/modules/ai-moderation/batchOutcomeClassifier.js";
|
||||
|
||||
const msgs = (...ids: string[]) => ids.map((id) => ({ id }));
|
||||
|
||||
describe("partitionBatchOutcome", () => {
|
||||
it("marks ALL targets upload_pending when the full-batch guard fires", () => {
|
||||
const out = partitionBatchOutcome(msgs("a", "b"), {
|
||||
ok: true,
|
||||
rows: [],
|
||||
uploadPendingIds: ["a", "b"],
|
||||
});
|
||||
expect(out.get("a")).toBe("upload_pending");
|
||||
expect(out.get("b")).toBe("upload_pending");
|
||||
});
|
||||
|
||||
it("never classifies an explicit upload_pending id as incomplete", () => {
|
||||
// The regression this file exists for: uploadPendingIds must win over the
|
||||
// missing-row heuristic.
|
||||
const out = partitionBatchOutcome(msgs("a"), {
|
||||
ok: true,
|
||||
rows: [],
|
||||
uploadPendingIds: ["a"],
|
||||
});
|
||||
expect(out.get("a")).not.toBe("incomplete");
|
||||
expect(out.get("a")).toBe("upload_pending");
|
||||
});
|
||||
|
||||
it("partitions a mixed batch: completed + upload-pending + missing", () => {
|
||||
const out = partitionBatchOutcome(msgs("ok1", "up1", "gone1"), {
|
||||
ok: true,
|
||||
rows: [
|
||||
{ id: "ok1", ai_status: "clean" },
|
||||
// up1 has NO row but IS in uploadPendingIds -> deferred, not failed
|
||||
],
|
||||
uploadPendingIds: ["up1"],
|
||||
});
|
||||
expect(out.get("ok1")).toBe("completed");
|
||||
expect(out.get("up1")).toBe("upload_pending");
|
||||
expect(out.get("gone1")).toBe("incomplete"); // unexplained drop stays retryable
|
||||
});
|
||||
|
||||
it("routes flag-based failures to their buckets", () => {
|
||||
const out = partitionBatchOutcome(msgs("i", "p", "f"), {
|
||||
ok: true,
|
||||
rows: [
|
||||
{
|
||||
id: "i",
|
||||
ai_status: "error",
|
||||
ai_moderation_flags: JSON.stringify(["analysis_incomplete"]),
|
||||
},
|
||||
{
|
||||
id: "p",
|
||||
ai_status: "error",
|
||||
ai_moderation_flags: JSON.stringify(["analysis_parse_failed"]),
|
||||
},
|
||||
{
|
||||
id: "f",
|
||||
ai_status: "error",
|
||||
ai_moderation_flags: JSON.stringify(["analysis_api_failed"]),
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(out.get("i")).toBe("incomplete");
|
||||
expect(out.get("p")).toBe("parse_failed");
|
||||
expect(out.get("f")).toBe("api_failed");
|
||||
});
|
||||
|
||||
it("treats error rows with no known flag as retryable incomplete", () => {
|
||||
const out = partitionBatchOutcome(msgs("x"), {
|
||||
ok: true,
|
||||
rows: [
|
||||
{ id: "x", ai_status: "error", ai_moderation_flags: '["weird_flag"]' },
|
||||
],
|
||||
});
|
||||
expect(out.get("x")).toBe("incomplete");
|
||||
});
|
||||
|
||||
it("accepts DB-round-trip MessageRecord shape (null ai_status)", () => {
|
||||
const out = partitionBatchOutcome([{ id: "r", ai_status: null }] as never, {
|
||||
ok: true,
|
||||
rows: [{ id: "r", ai_status: null }],
|
||||
});
|
||||
expect(out.get("r")).toBe("completed");
|
||||
});
|
||||
});
|
||||
|
||||
describe("computeUploadPollDelayMs", () => {
|
||||
it("ramps linearly and respects the cap", () => {
|
||||
expect(computeUploadPollDelayMs(1, 1500, 8000)).toBe(1500);
|
||||
expect(computeUploadPollDelayMs(2, 1500, 8000)).toBe(3000);
|
||||
expect(computeUploadPollDelayMs(3, 1500, 8000)).toBe(4500);
|
||||
expect(computeUploadPollDelayMs(9, 1500, 8000)).toBe(8000); // capped
|
||||
});
|
||||
|
||||
it("is safe on degenerate input", () => {
|
||||
expect(computeUploadPollDelayMs(0, 1500, 8000)).toBe(1500);
|
||||
expect(computeUploadPollDelayMs(-5, 1000, 4000)).toBe(1000);
|
||||
});
|
||||
});
|
||||
@@ -145,10 +145,10 @@ describe("isGloballyReusableCleanVerdict", () => {
|
||||
});
|
||||
|
||||
it("rejects entries older than the freshness window", () => {
|
||||
// Default AI_CACHE_GLOBAL_REUSE_MAX_AGE_H = 72h.
|
||||
const tooOld = Date.now() - 73 * 60 * 60 * 1000;
|
||||
// Default AI_CACHE_GLOBAL_REUSE_MAX_AGE_H = 120h.
|
||||
const tooOld = Date.now() - 121 * 60 * 60 * 1000;
|
||||
expect(isGloballyReusableCleanVerdict(makeVerdict(), tooOld)).toBe(false);
|
||||
const freshEnough = Date.now() - 71 * 60 * 60 * 1000;
|
||||
const freshEnough = Date.now() - 119 * 60 * 60 * 1000;
|
||||
expect(isGloballyReusableCleanVerdict(makeVerdict(), freshEnough)).toBe(
|
||||
true,
|
||||
);
|
||||
@@ -157,4 +157,13 @@ describe("isGloballyReusableCleanVerdict", () => {
|
||||
it("skips the age check when analyzedAt is unknown", () => {
|
||||
expect(isGloballyReusableCleanVerdict(makeVerdict(), undefined)).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects non-standard statuses such as processing write-backs", () => {
|
||||
expect(
|
||||
isGloballyReusableCleanVerdict(
|
||||
{ ...makeVerdict(), status: "processing" },
|
||||
undefined,
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,79 @@
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// classifyIndividualWorkerResult — upload-pending vs success vs incomplete vs error
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
// Bug (2026-08-24): the worker's upload-pending race guard returned
|
||||
// {ok:true, results:[]}; the processor treated it as a completed moderation,
|
||||
// leaving messages stuck in `processing` until the 300s cleanup reverted them.
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { classifyIndividualWorkerResult } from "../src/modules/ai-moderation/fallbackResultClassifier.js";
|
||||
|
||||
describe("classifyIndividualWorkerResult", () => {
|
||||
it("classifies the upload-pending race guard signal FIRST", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: true,
|
||||
results: [],
|
||||
uploadPending: true,
|
||||
}),
|
||||
).toBe("upload_pending");
|
||||
});
|
||||
|
||||
it("classifies a normal verdict as success", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: true,
|
||||
results: [{ status: "clean", flags: [] }],
|
||||
}),
|
||||
).toBe("success");
|
||||
});
|
||||
|
||||
it("classifies analysis_incomplete as incomplete", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: true,
|
||||
results: [{ status: "error", flags: ["analysis_incomplete"] }],
|
||||
}),
|
||||
).toBe("incomplete");
|
||||
});
|
||||
|
||||
it("accepts flags as JSON string (DB round-trip shape)", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: true,
|
||||
results: [
|
||||
{ status: "error", flags: JSON.stringify(["analysis_incomplete"]) },
|
||||
],
|
||||
}),
|
||||
).toBe("incomplete");
|
||||
});
|
||||
|
||||
it("classifies ok:false as error", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: false,
|
||||
results: [],
|
||||
error: "boom",
|
||||
}),
|
||||
).toBe("error");
|
||||
});
|
||||
|
||||
it("THE BUG: empty results with ok:true is an ERROR, not a success", () => {
|
||||
// Old code silently succeeded here → stuck `processing` rows.
|
||||
expect(classifyIndividualWorkerResult({ ok: true, results: [] })).toBe(
|
||||
"error",
|
||||
);
|
||||
expect(
|
||||
classifyIndividualWorkerResult({ ok: true, results: [undefined] }),
|
||||
).toBe("error");
|
||||
expect(classifyIndividualWorkerResult({})).toBe("error");
|
||||
});
|
||||
|
||||
it("treats a result with no status as unexplainable (error)", () => {
|
||||
expect(
|
||||
classifyIndividualWorkerResult({
|
||||
ok: true,
|
||||
results: [{ flags: [] }],
|
||||
}),
|
||||
).toBe("error");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,91 @@
|
||||
# FE Optimasi: Error Handling, State Refactor, Polish
|
||||
|
||||
**Scope:** `services/frontend/src` — Next.js 16, React 19, SWR, oRPC-over-WS.
|
||||
**Branch:** `feat/fe-error-handling-state` (buat baru dari `main`).
|
||||
**User:** MythEclipse — bahasa Indonesia/Inggris campur, "gas" = eksekusi langsung.
|
||||
|
||||
## 1. Masalah yang teridentifikasi
|
||||
|
||||
### 1.1 Error handling tidak seragam
|
||||
- `ErrorState` (states.tsx:126) sudah ada dengan `onRetry?`, tapi **80% pemanggilan tidak pass `onRetry`** → user stuck, harus refresh manual.
|
||||
- Pola `if (error && !data) return <ErrorState />` berulang di 5+ view, setiap satu hardcode logic-nya.
|
||||
- `useAction` (use-action.ts) set `error` di state tapi komponen pakai `try/catch` lokal juga — **dua error states** (hook `action.error` + lokal `catch (e)`).
|
||||
|
||||
### 1.2 State duplication / double source of truth
|
||||
- `useReview` (use-messages.ts:129) pakai `refreshInterval: 15_000` (polling) padahal WS sudah broadcast `moderation_action` real-time. **Polling + WS = double-fetch + race condition.**
|
||||
- `page.tsx` dan `view.tsx` duplikat state `guildId`/`channelId`/`query` — logic selection ada di 2 tempat.
|
||||
- Chatbot pakai local `failed` state + `setFailed(text)` untuk retry, tapi `useAction` sudah ada `error` field yang tidak dipakai.
|
||||
- `useSpeakers` (use-voice.ts:50) pakai local `useState` untuk speakers, bukan SWR → tidak konsisten dengan pola SWR yang dipakai semua hook lain.
|
||||
|
||||
### 1.3 WebSocket error/user feedback lemah
|
||||
- `WsConnection.onerror` (connection.ts:78) cuma `setStatus("error")` — tidak ada info ke user "reconnecting… (attempt N)".
|
||||
- Tidak ada broadcast error ke komponen (tidak ada `ws_error` handler).
|
||||
|
||||
## 2. Rencana perubahan
|
||||
|
||||
### 2.1 Standardisasi ErrorBoundary + Retry (FE-only)
|
||||
**File:** `src/components/shared/states.tsx`
|
||||
- Tambahkan komponen `ErrorBoundary` (React error boundary untuk crash React, bukan fetch error).
|
||||
- Refactor semua `<ErrorState error={error} />` → `<ErrorState error={error} onRetry={retryFn} />`.
|
||||
- Tambahkan helper `withSWR` atau pattern: tiap view pakai `error` + `mutate`/`refetch` dari SWR dan pass ke ErrorState.
|
||||
|
||||
**File:** `src/app/(dashboard)/*/view.tsx` (6 files)
|
||||
- Setiap `ErrorState` dapat `onRetry` yang memanggil `mutate`/`refetch`.
|
||||
- `DashboardView`: `onRetry={() => void mutate(["dashboard-stats"])}` — tapi SWR keys tersebar. Solusi: export `mutate` via custom hook atau pakai `useSWR` config `onErrorRetry` global.
|
||||
- **Keputusan:** gunakan pola **global SWR config** (`src/lib/swr-config.ts`) dengan `onErrorRetry` backoff, dan tiap view pass `onRetry` explicit ke ErrorState.
|
||||
|
||||
### 2.2 Hapus polling `useReview`, ganti WS-driven
|
||||
**File:** `src/hooks/use-messages.ts`
|
||||
- Hapus `refreshInterval: 15_000` dari `useReview`.
|
||||
- Tambahkan `useReviewWsSync(ws)` — subscribe ke `moderation_action` WS event, mutate key `["messages-review", channelId]`.
|
||||
- Backend sudah broadcast `moderation_action` via WS (redis-channels.ts:130). Review messages yang di-flag akan dapat `moderation_action` event. **Bisa pakai ini.**
|
||||
|
||||
**File:** `src/app/(dashboard)/messages/view.tsx`
|
||||
- Tambahkan `useReviewWsSync(ws)` call.
|
||||
|
||||
### 2.3 Konsistensi state: `useSpeakers` → SWR
|
||||
**File:** `src/hooks/use-voice.ts`
|
||||
- Refactor `useSpeakers` agar pakai SWR key `["voice-speakers"]` + initialData dari `useVoiceStatus`. Ini memungkinkan revalidate + cache sharing.
|
||||
- Tapi `voice_state` dan `voice_active_user` adalah WS events — perlu persist ke SWR cache via `mutate`. Refactor: `useSpeakers` subscribe WS + mutate SWR key.
|
||||
|
||||
### 2.4 Chatbot: gunakan `useAction` error state, hapus duplikat `failed`
|
||||
**File:** `src/components/chatbot/chatbot.tsx`
|
||||
- Hapus `const [failed, setFailed]` — sebalihoikan ke `useAction` return `error`.
|
||||
- Tapi `useAction` `mutateAsync` throw — perlu catch. Refactor: pakai `mutate` (fire-and-forget) + `isPending` + `error`.
|
||||
- **Note:** chatbot pakai `send` yang butuh pemuatan history — tetap pakai local state untuk msgs tapi gunakan `action.error` untuk display.
|
||||
|
||||
### 2.5 WS error feedback
|
||||
**File:** `src/lib/ws/connection.ts`
|
||||
- `onerror` emit kode/status ke status listeners.
|
||||
- Tambahkan method `getReconnectAttempt()` atau expose via status change.
|
||||
|
||||
**File:** `src/lib/ws/context.tsx`
|
||||
- Subscribe `onStatusChange` di `WsProvider`, toast "Reconnecting… (attempt N)" ketika status `error`/`connecting`.
|
||||
|
||||
## 3. Verification
|
||||
- `npx tsc --noEmit` — compile OK
|
||||
- `npx biome check src/` — lint OK
|
||||
- `npm run build` — build OK (Next 16 SSG/SSR)
|
||||
- Manual: refresh halaman, pastikan ErrorState muncul dengan tombol Retry yang bisa diklik.
|
||||
|
||||
## 4. Files yang disentuh
|
||||
```
|
||||
src/components/shared/states.tsx # ErrorBoundary + helper
|
||||
src/lib/swr-config.ts # global SWR config (baru)
|
||||
src/hooks/use-messages.ts # useReviewWsSync, hapus polling
|
||||
src/hooks/use-voice.ts # useSpeakers → SWR
|
||||
src/hooks/use-action.ts # expose resetError
|
||||
src/lib/ws/connection.ts # reconnect info
|
||||
src/lib/ws/context.tsx # WS error toast
|
||||
src/app/(dashboard)/messages/view.tsx # useReviewWsSync
|
||||
src/app/(dashboard)/messages/page.tsx # SSR error passthrough
|
||||
src/app/(dashboard)/moderation/view.tsx # onRetry
|
||||
src/app/(dashboard)/dashboard/view.tsx # onRetry
|
||||
src/app/(dashboard)/voice/view.tsx # onRetry
|
||||
src/app/(dashboard)/media/view.tsx # onRetry
|
||||
src/app/(dashboard)/recordings/view.tsx # onRetry
|
||||
src/components/chatbot/chatbot.tsx # pakai useAction error
|
||||
```
|
||||
|
||||
## 5. Out of scope (BE)
|
||||
- Regex heuristic di moderation.repository.ts (scam domain extraction) — ini BE task, catat tapi jangan sentuh kecuali diminta.
|
||||
@@ -50,7 +50,12 @@ export function DashboardView({
|
||||
initialStats?: DashboardStats;
|
||||
initialActivity?: Awaited<ReturnType<typeof useActivity>>["data"];
|
||||
}) {
|
||||
const { data: stats, isLoading, error } = useStats(initialStats);
|
||||
const {
|
||||
data: stats,
|
||||
isLoading,
|
||||
error,
|
||||
mutate: mutateStats,
|
||||
} = useStats(initialStats);
|
||||
const { data: activity } = useActivity(14, initialActivity as never);
|
||||
const { data: reactors } = useTopReactors();
|
||||
const { data: reactions } = useTopReactions();
|
||||
@@ -65,7 +70,8 @@ export function DashboardView({
|
||||
);
|
||||
}, [stats, ambient]);
|
||||
|
||||
if (error && !stats) return <ErrorState error={error} />;
|
||||
if (error && !stats)
|
||||
return <ErrorState error={error} onRetry={() => void mutateStats()} />;
|
||||
if (!stats && isLoading)
|
||||
return (
|
||||
<div className="space-y-5">
|
||||
@@ -78,7 +84,13 @@ export function DashboardView({
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
if (!stats) return <ErrorState error={error ?? new Error("No data")} />;
|
||||
if (!stats)
|
||||
return (
|
||||
<ErrorState
|
||||
error={error ?? new Error("No data")}
|
||||
onRetry={() => void mutateStats()}
|
||||
/>
|
||||
);
|
||||
|
||||
const s = stats;
|
||||
const total = s.total_flagged + s.total_clean || 1;
|
||||
|
||||
@@ -33,7 +33,12 @@ import { useWebSocket } from "@/lib/ws/context";
|
||||
|
||||
export function MediaView({ initialStatus }: { initialStatus?: MediaState }) {
|
||||
const ws = useWebSocket();
|
||||
const { data: media, isLoading, error } = useMediaState(initialStatus);
|
||||
const {
|
||||
data: media,
|
||||
isLoading,
|
||||
error,
|
||||
mutate,
|
||||
} = useMediaState(initialStatus);
|
||||
const queue = useMediaQueue();
|
||||
const skip = useMediaSkip();
|
||||
const stop = useMediaStop();
|
||||
@@ -79,7 +84,8 @@ export function MediaView({ initialStatus }: { initialStatus?: MediaState }) {
|
||||
}
|
||||
};
|
||||
|
||||
if (error && !media) return <ErrorState error={error} />;
|
||||
if (error && !media)
|
||||
return <ErrorState error={error} onRetry={() => void mutate()} />;
|
||||
if (!media && isLoading)
|
||||
return (
|
||||
<div className="space-y-5">
|
||||
|
||||
@@ -39,6 +39,7 @@ import {
|
||||
useMessagesStream,
|
||||
useMessagesWsSync,
|
||||
useRecentEdits,
|
||||
useReviewWsSync,
|
||||
useSemanticSearch,
|
||||
} from "@/hooks";
|
||||
import { aiTone } from "@/lib/ai-status";
|
||||
@@ -94,6 +95,7 @@ export function MessagesView({
|
||||
data: messages,
|
||||
isLoading,
|
||||
error,
|
||||
refetch,
|
||||
} = useMessages(
|
||||
guildId ?? "",
|
||||
channelId ?? undefined,
|
||||
@@ -112,6 +114,7 @@ export function MessagesView({
|
||||
const hasMore = pageInfo?.hasMore ?? false;
|
||||
const loadMore = useLoadMore();
|
||||
useMessagesWsSync(ws, guildId ?? "");
|
||||
useReviewWsSync(ws);
|
||||
const search = useMessageSearch(
|
||||
query,
|
||||
query.trim().length >= 2 && !semanticMode,
|
||||
@@ -326,7 +329,7 @@ export function MessagesView({
|
||||
}
|
||||
/>
|
||||
{error && !messages ? (
|
||||
<ErrorState error={error} />
|
||||
<ErrorState error={error} onRetry={() => refetch()} />
|
||||
) : isLoading && !messages ? (
|
||||
<SkeletonRows rows={8} />
|
||||
) : list.length === 0 ? (
|
||||
|
||||
@@ -81,7 +81,12 @@ export function ModerationView({
|
||||
initialStats?: ModerationStats;
|
||||
initialActions?: ModerationAction[];
|
||||
}) {
|
||||
const { data: stats, isLoading, error } = useModerationStats(initialStats);
|
||||
const {
|
||||
data: stats,
|
||||
isLoading,
|
||||
error,
|
||||
mutate: mutateStats,
|
||||
} = useModerationStats(initialStats);
|
||||
const [statusFilter, setStatusFilter] = useState<string>("");
|
||||
const [typeFilter, setTypeFilter] = useState<string>("");
|
||||
const { data: actions } = useModerationActions(
|
||||
@@ -122,7 +127,8 @@ export function ModerationView({
|
||||
);
|
||||
}, [failedRate, ambient]);
|
||||
|
||||
if (error && !stats) return <ErrorState error={error} />;
|
||||
if (error && !stats)
|
||||
return <ErrorState error={error} onRetry={() => void mutateStats()} />;
|
||||
if (!stats && isLoading)
|
||||
return (
|
||||
<div className="space-y-5">
|
||||
@@ -131,7 +137,13 @@ export function ModerationView({
|
||||
<SkeletonRows rows={6} />
|
||||
</div>
|
||||
);
|
||||
if (!stats) return <ErrorState error={error ?? new Error("No data")} />;
|
||||
if (!stats)
|
||||
return (
|
||||
<ErrorState
|
||||
error={error ?? new Error("No data")}
|
||||
onRetry={() => void mutateStats()}
|
||||
/>
|
||||
);
|
||||
|
||||
const statusOpts: SelectOption[] = [
|
||||
{ value: "", label: "All statuses" },
|
||||
|
||||
@@ -33,7 +33,7 @@ export function RecordingsView({
|
||||
initialItems?: VoiceRecording[];
|
||||
}) {
|
||||
const ws = useWebSocket();
|
||||
const { data: items, isLoading, error } = useRecordings(initialItems);
|
||||
const { data: items, isLoading, error, mutate } = useRecordings(initialItems);
|
||||
const del = useDeleteRecording();
|
||||
useRecordingsWsSync(ws);
|
||||
const ambient = useAmbient();
|
||||
@@ -56,7 +56,8 @@ export function RecordingsView({
|
||||
}
|
||||
};
|
||||
|
||||
if (error && !items) return <ErrorState error={error} />;
|
||||
if (error && !items)
|
||||
return <ErrorState error={error} onRetry={() => void mutate()} />;
|
||||
if (!items && isLoading)
|
||||
return (
|
||||
<GlassPanel>
|
||||
|
||||
@@ -40,7 +40,12 @@ export function VoiceView({
|
||||
initialGuilds?: Guild[];
|
||||
}) {
|
||||
const ws = useWebSocket();
|
||||
const { data: status, isLoading, error } = useVoiceStatus(initialStatus);
|
||||
const {
|
||||
data: status,
|
||||
isLoading,
|
||||
error,
|
||||
mutate,
|
||||
} = useVoiceStatus(initialStatus);
|
||||
const connect = useVoiceConnect();
|
||||
const disconnect = useVoiceDisconnect();
|
||||
const mic = useMicTransmit(ws);
|
||||
@@ -68,7 +73,8 @@ export function VoiceView({
|
||||
else ambient.set("vermilion", 0.35, "voice idle");
|
||||
}, [status?.connected, ambient]);
|
||||
|
||||
if (error && !status) return <ErrorState error={error} />;
|
||||
if (error && !status)
|
||||
return <ErrorState error={error} onRetry={() => void mutate()} />;
|
||||
if (!status && isLoading)
|
||||
return (
|
||||
<div className="space-y-5">
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
"use client";
|
||||
|
||||
import { Component, type ReactNode } from "react";
|
||||
import { ErrorState } from "./states";
|
||||
|
||||
/**
|
||||
* React error boundary — catches render/exception crashes in child trees
|
||||
* (e.g. third-party lib throwing on unexpected payload shape) and surfaces
|
||||
* a consistent ErrorState instead of unmounting the whole app shell.
|
||||
*
|
||||
* Usage: wrap leaf views in <ErrorBoundary><SomeView /></ErrorBoundary>.
|
||||
*/
|
||||
interface ErrorBoundaryProps {
|
||||
children: ReactNode;
|
||||
fallback?: (error: Error, reset: () => void) => ReactNode;
|
||||
onReset?: () => void;
|
||||
}
|
||||
|
||||
interface ErrorBoundaryState {
|
||||
error: Error | null;
|
||||
}
|
||||
|
||||
export class ErrorBoundary extends Component<
|
||||
ErrorBoundaryProps,
|
||||
ErrorBoundaryState
|
||||
> {
|
||||
constructor(props: ErrorBoundaryProps) {
|
||||
super(props);
|
||||
this.state = { error: null };
|
||||
}
|
||||
|
||||
static getDerivedStateFromError(error: Error): ErrorBoundaryState {
|
||||
return { error };
|
||||
}
|
||||
|
||||
reset = () => {
|
||||
this.setState({ error: null });
|
||||
this.props.onReset?.();
|
||||
};
|
||||
|
||||
override render() {
|
||||
if (!this.state.error) return this.props.children;
|
||||
|
||||
if (this.props.fallback) {
|
||||
return this.props.fallback(this.state.error, this.reset);
|
||||
}
|
||||
|
||||
return (
|
||||
<ErrorState
|
||||
title="Something went wrong"
|
||||
error={this.state.error}
|
||||
onRetry={this.reset}
|
||||
/>
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ export { GuildChannelPicker } from "./guild-picker";
|
||||
export { MarkdownLite } from "./markdown";
|
||||
export { PageTransition } from "./page-transition";
|
||||
export { MetricTile, SectionHeader } from "./section";
|
||||
export { ErrorBoundary } from "./error-boundary";
|
||||
export {
|
||||
EmptyState,
|
||||
ErrorState,
|
||||
|
||||
@@ -31,6 +31,7 @@ export {
|
||||
useMessagesWsSync,
|
||||
useRecentEdits,
|
||||
useReview,
|
||||
useReviewWsSync,
|
||||
useSemanticSearch,
|
||||
useTextChannels,
|
||||
} from "./use-messages";
|
||||
|
||||
@@ -7,17 +7,23 @@ export interface UseActionState {
|
||||
|
||||
/**
|
||||
* A lightweight mutation hook with a TanStack-compatible surface
|
||||
* ({ mutate, mutateAsync, isPending, error }) built on plain state —
|
||||
* ({ mutate, mutateAsync, isPending, error, resetError }) built on plain state —
|
||||
* the SWR replacement for useMutation. Fire-and-forget via `mutate`,
|
||||
* await the result via `mutateAsync`.
|
||||
*
|
||||
* `onSuccess` receives (data, args) and may perform SWR cache updates
|
||||
* (e.g. `mutate(key, data, { revalidate: false })`).
|
||||
*
|
||||
* `onError` receives the caught error and args — use it for side-effect
|
||||
* logging without throwing (the error is also surfaced via `state.error`).
|
||||
*
|
||||
* `resetError` clears the error state without re-running the action.
|
||||
*/
|
||||
export function useAction<TArgs = void, TResult = unknown>(
|
||||
fn: (args: TArgs) => Promise<TResult>,
|
||||
options?: {
|
||||
onSuccess?: (data: TResult, args: TArgs) => void | Promise<void>;
|
||||
onError?: (error: Error, args: TArgs) => void | Promise<void>;
|
||||
},
|
||||
) {
|
||||
const [state, setState] = useState<UseActionState>({
|
||||
@@ -29,6 +35,8 @@ export function useAction<TArgs = void, TResult = unknown>(
|
||||
fnRef.current = fn;
|
||||
const onSuccessRef = useRef(options?.onSuccess);
|
||||
onSuccessRef.current = options?.onSuccess;
|
||||
const onErrorRef = useRef(options?.onError);
|
||||
onErrorRef.current = options?.onError;
|
||||
|
||||
const run = useCallback(async (args?: TArgs): Promise<TResult> => {
|
||||
setState({ isPending: true, error: null });
|
||||
@@ -38,8 +46,10 @@ export function useAction<TArgs = void, TResult = unknown>(
|
||||
setState({ isPending: false, error: null });
|
||||
return data;
|
||||
} catch (err) {
|
||||
setState({ isPending: false, error: err as Error });
|
||||
throw err;
|
||||
const error = err as Error;
|
||||
setState({ isPending: false, error });
|
||||
await onErrorRef.current?.(error, args as TArgs);
|
||||
throw error;
|
||||
}
|
||||
}, []);
|
||||
|
||||
@@ -50,6 +60,6 @@ export function useAction<TArgs = void, TResult = unknown>(
|
||||
mutateAsync: run,
|
||||
isPending: state.isPending,
|
||||
error: state.error,
|
||||
reset: () => setState({ isPending: false, error: null }),
|
||||
resetError: useCallback(() => setState((s) => ({ ...s, error: null })), []),
|
||||
};
|
||||
}
|
||||
@@ -124,21 +124,43 @@ export function useImages(guildId: string) {
|
||||
);
|
||||
}
|
||||
|
||||
// ── Review ───────────────────────────────────────
|
||||
// ── Review ────────────────────────────────────────
|
||||
// WS-driven now (was: refreshInterval:15000 polling). The backend broadcasts
|
||||
// `moderation_action` events over WS whenever a moderation action is created;
|
||||
// the review list is revalidated on that event instead of polling.
|
||||
|
||||
export function useReview(channelId?: string) {
|
||||
const key = msgKeys.review(channelId);
|
||||
return useSWR<MessageRecord[]>(
|
||||
msgKeys.review(channelId),
|
||||
key,
|
||||
async () => {
|
||||
const result = await messagesApi.getReview(50, channelId || undefined);
|
||||
return result.results;
|
||||
},
|
||||
{
|
||||
refreshInterval: 15_000,
|
||||
revalidateOnFocus: true,
|
||||
// No more polling — WS sync handles real-time updates.
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to WS `moderation_action` events and invalidate the review
|
||||
* SWR cache. Replaces the old 15-second polling interval that caused
|
||||
* duplicate requests and race conditions with WS updates.
|
||||
*/
|
||||
export function useReviewWsSync(ws: WsHook, channelId?: string) {
|
||||
const { mutate } = useSWRConfig();
|
||||
const key = msgKeys.review(channelId);
|
||||
|
||||
useEffect(() => {
|
||||
const unsub = ws.on("moderation_action", () => {
|
||||
void mutate(key, undefined, { revalidate: true });
|
||||
});
|
||||
return unsub;
|
||||
}, [ws, key, mutate]);
|
||||
}
|
||||
|
||||
// ── Detail ───────────────────────────────────────
|
||||
|
||||
export function useMessageDetail(id: string | null) {
|
||||
@@ -168,7 +190,11 @@ export function useMessageDetail(id: string | null) {
|
||||
message: detail.data ?? null,
|
||||
attachments: attachments.data ?? [],
|
||||
loading: detail.isLoading || attachments.isLoading,
|
||||
error: detail.error,
|
||||
error: detail.error ?? attachments.error,
|
||||
refetch: () => {
|
||||
void detail.mutate();
|
||||
void attachments.mutate();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -216,7 +242,7 @@ export function useMessagesWsSync(ws: WsHook, guildId: string) {
|
||||
const { mutate } = useSWRConfig();
|
||||
useEffect(() => {
|
||||
if (!guildId) return;
|
||||
// Patch every message-list key for this guild (all channels + "**filtered**").
|
||||
// Patch every message-list key for this guild (all channels + "__all__").
|
||||
// The updater receives the SWR key so we can honor its channel filter:
|
||||
// a live `message_created`/updated for channel B must NOT be prepended to
|
||||
// a list that is filtered down to channel A.
|
||||
@@ -255,13 +281,13 @@ export function useMessagesWsSync(ws: WsHook, guildId: string) {
|
||||
});
|
||||
const unsub2 = ws.on("message_updated", (data) => {
|
||||
const msg = data as Partial<MessageRecord> & { id: string };
|
||||
// The gateway broadcasts a PARTIAL update ({ id, edited_content,
|
||||
// edited_at, ... }) — merge it over the existing record instead of
|
||||
// replacing it, or the card would lose username/content/channel/etc.
|
||||
// The gateway broadcasts a PARTIAL update ({ id } plus the changed fields
|
||||
// — merge it over the existing record instead of replacing it, or the
|
||||
// card would lose username/content/channel/etc.
|
||||
patchLists(
|
||||
(_k, m) =>
|
||||
(m as Partial<MessageRecord>).channel_id === undefined ||
|
||||
matchesFilter(_k as unknown[], m),
|
||||
matchesFilter(_k as unknown[], m as { channel_id?: string }),
|
||||
(old) =>
|
||||
old
|
||||
? {
|
||||
@@ -271,7 +297,7 @@ export function useMessagesWsSync(ws: WsHook, guildId: string) {
|
||||
),
|
||||
}
|
||||
: old,
|
||||
msg,
|
||||
msg as { channel_id?: string },
|
||||
);
|
||||
void mutate(
|
||||
msgKeys.detail(msg.id),
|
||||
@@ -321,7 +347,7 @@ export function useMessagesWsSync(ws: WsHook, guildId: string) {
|
||||
* `message_snapshot` into the SWR list as it arrives, so the UI renders
|
||||
* progressively. Falls back to the batched `messagesApi.list` if WS is down.
|
||||
*
|
||||
* Returns: { streaming, streamed, error }.
|
||||
* Returns: { streaming, error }.
|
||||
*/
|
||||
export function useMessagesStream(
|
||||
ws: WsHook,
|
||||
@@ -414,3 +440,5 @@ export function useRecentEdits(
|
||||
{ fallbackData: initialData },
|
||||
);
|
||||
}
|
||||
|
||||
export { msgKeys };
|
||||
@@ -4,19 +4,13 @@ import { useAction } from "@/hooks/use-action";
|
||||
import { voiceApi } from "@/lib/api";
|
||||
import { MicTransmitter } from "@/lib/audio/mic-transmit";
|
||||
import { PcmPlayer } from "@/lib/audio/pcm-player";
|
||||
import { hashUserId } from "@/lib/hash";
|
||||
import type { ActiveSpeaker, Channel, VoiceStatus } from "@/lib/types";
|
||||
import type { PcmChunk } from "@/lib/ws/types";
|
||||
import type { WsHook } from "@/lib/ws-hook";
|
||||
|
||||
/** FNV-1a 32-bit — same hash the gateway uses to tag PCM frames. */
|
||||
export function hashUserId(userId: string): number {
|
||||
let hash = 0x811c9dc5;
|
||||
for (let i = 0; i < userId.length; i++) {
|
||||
hash ^= userId.charCodeAt(i);
|
||||
hash = Math.imul(hash, 0x01000193);
|
||||
}
|
||||
return hash >>> 0;
|
||||
}
|
||||
// Re-export for components that still import hashUserId from this module.
|
||||
export { hashUserId };
|
||||
|
||||
const STATUS_KEY = ["voice-status"] as const;
|
||||
|
||||
@@ -34,6 +28,8 @@ export function useVoiceChannels(guildId: string) {
|
||||
);
|
||||
}
|
||||
|
||||
const SPEAKERS_KEY = ["voice-speakers"] as const;
|
||||
|
||||
/**
|
||||
* Live shared speaker state.
|
||||
*
|
||||
@@ -44,48 +40,66 @@ export function useVoiceChannels(guildId: string) {
|
||||
* every client with the same list);
|
||||
* - `voice_active_user` → incremental upsert of a single speaker delta.
|
||||
*
|
||||
* Now backed by SWR (consistent with all other hooks) so cache, revalidation,
|
||||
* and deduping apply. WS events mutate the SWR cache directly
|
||||
* ({ revalidate: false }) to avoid refetching the full status.
|
||||
*
|
||||
* This replaces the old per-browser model where each tab accumulated speakers
|
||||
* only from events it happened to receive while mounted.
|
||||
*/
|
||||
export function useSpeakers(initialStatusActive?: ActiveSpeaker[]) {
|
||||
const [speakers, setSpeakers] = useState<ActiveSpeaker[]>(
|
||||
initialStatusActive ?? [],
|
||||
const {
|
||||
data: speakers,
|
||||
error,
|
||||
mutate,
|
||||
isValidating,
|
||||
} = useSWR<ActiveSpeaker[]>(SPEAKERS_KEY, () => Promise.resolve([]), {
|
||||
fallbackData: initialStatusActive ?? [],
|
||||
revalidateOnMount: false,
|
||||
revalidateOnFocus: false,
|
||||
revalidateOnReconnect: false,
|
||||
});
|
||||
|
||||
const subscribe = useCallback(
|
||||
(ws: WsHook) => {
|
||||
const unsubSnapshot = ws.on("voice_state", (data) => {
|
||||
const state = data as { activeSpeakers?: ActiveSpeaker[] };
|
||||
if (Array.isArray(state?.activeSpeakers)) {
|
||||
void mutate(state.activeSpeakers, { revalidate: false });
|
||||
}
|
||||
});
|
||||
const unsub = ws.on("voice_active_user", (data) => {
|
||||
const speaker = data as ActiveSpeaker;
|
||||
void mutate(
|
||||
(prev: ActiveSpeaker[] | undefined) => {
|
||||
const arr = prev ?? [];
|
||||
const idx = arr.findIndex((s) => s.userId === speaker.userId);
|
||||
if (idx >= 0) {
|
||||
const next = [...arr];
|
||||
next[idx] = speaker;
|
||||
return next;
|
||||
}
|
||||
return [...arr, speaker];
|
||||
},
|
||||
{ revalidate: false },
|
||||
);
|
||||
});
|
||||
return () => {
|
||||
unsubSnapshot();
|
||||
unsub();
|
||||
};
|
||||
},
|
||||
[mutate],
|
||||
);
|
||||
|
||||
const subscribe = useCallback((ws: WsHook) => {
|
||||
const unsubSnapshot = ws.on("voice_state", (data) => {
|
||||
const state = data as { activeSpeakers?: ActiveSpeaker[] };
|
||||
if (Array.isArray(state?.activeSpeakers)) {
|
||||
setSpeakers(state.activeSpeakers);
|
||||
}
|
||||
});
|
||||
const unsub = ws.on("voice_active_user", (data) => {
|
||||
const speaker = data as ActiveSpeaker;
|
||||
setSpeakers((prev) => {
|
||||
const idx = prev.findIndex((s) => s.userId === speaker.userId);
|
||||
if (idx >= 0) {
|
||||
const next = [...prev];
|
||||
next[idx] = speaker;
|
||||
return next;
|
||||
}
|
||||
return [...prev, speaker];
|
||||
});
|
||||
});
|
||||
return () => {
|
||||
unsubSnapshot();
|
||||
unsub();
|
||||
setSpeakers([]);
|
||||
};
|
||||
}, []);
|
||||
|
||||
return { speakers, subscribe };
|
||||
return { speakers: speakers ?? [], subscribe, error, isValidating };
|
||||
}
|
||||
|
||||
function useStatusInvalidator() {
|
||||
const { mutate } = useSWRConfig();
|
||||
return () => {
|
||||
return useCallback(() => {
|
||||
void mutate(STATUS_KEY);
|
||||
};
|
||||
}, [mutate]);
|
||||
}
|
||||
|
||||
export function useVoiceConnect() {
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
/**
|
||||
* FNV-1a 32-bit hash — same hash the gateway uses to tag PCM frames.
|
||||
* Shared between use-voice hooks and ambient-canvas.
|
||||
*/
|
||||
export function hashUserId(userId: string): number {
|
||||
let hash = 0x811c9dc5;
|
||||
for (let i = 0; i < userId.length; i++) {
|
||||
hash ^= userId.charCodeAt(i);
|
||||
hash = Math.imul(hash, 0x01000193);
|
||||
}
|
||||
return hash >>> 0;
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
/**
|
||||
* Global SWR configuration for the GMW frontend.
|
||||
*
|
||||
* Centralizes revalidation policy, error retry with exponential backoff, and
|
||||
* deduplication so every `useSWR` call across the app gets consistent behaviour
|
||||
* without each hook re-declaring the same options.
|
||||
*
|
||||
* Import this in the root layout (or any component mounted once) via
|
||||
* `SWRConfig` from "swr".
|
||||
*/
|
||||
|
||||
import type { SWRConfiguration } from "swr";
|
||||
|
||||
/**
|
||||
* Exponential backoff retry — starts at ~1s, doubles up to 30s, then flatlines.
|
||||
* Matches the existing WsConnection reconnection philosophy.
|
||||
*/
|
||||
export const swrConfig: SWRConfiguration = {
|
||||
// Re-validate on focus (tab switch) but not on every interval by default.
|
||||
revalidateOnFocus: true,
|
||||
// Dedupe rapid identical requests within 2s.
|
||||
dedupingInterval: 2000,
|
||||
// Never throw unhandled rejections — every hook handles `error` gracefully.
|
||||
shouldRetryOnError: (error) => {
|
||||
// Don't retry on 404 (NotFound) — it's a client expectation, not transient.
|
||||
if (error?.statusCode === 404 || error?.code === "NOT_FOUND") return false;
|
||||
// Retry network errors, 5xx, and oRPC transport failures.
|
||||
return true;
|
||||
},
|
||||
onErrorRetry: (_error, _key, _config, revalidate, opts) => {
|
||||
const attemptCount = (opts as { attemptCount?: number }).attemptCount ?? 0;
|
||||
// Stop retrying after 3 attempts (≈ 1+2+4 = 7s max backoff).
|
||||
if (attemptCount >= 3) return;
|
||||
|
||||
// Exponential backoff: 1000 * 2^attempt, capped at 30000ms.
|
||||
const baseDelay = 1000 * 2 ** attemptCount;
|
||||
const delay = Math.min(baseDelay, 30000);
|
||||
|
||||
// Jitter ±25% to avoid thundering herd on shared endpoints.
|
||||
const jitter = delay * 0.25 * (Math.random() * 2 - 1);
|
||||
const timeout = Math.round(delay + jitter);
|
||||
|
||||
setTimeout(revalidate, timeout);
|
||||
},
|
||||
// Sensible defaults for the "loading" state — most data loads in <500ms.
|
||||
loadingTimeout: 15000,
|
||||
};
|
||||
@@ -37,6 +37,11 @@ export class WsConnection {
|
||||
return this._status;
|
||||
}
|
||||
|
||||
/** Current reconnect attempt count (0 = connected/first attempt). */
|
||||
get reconnectAttemptCount(): number {
|
||||
return this.reconnectAttempt;
|
||||
}
|
||||
|
||||
onStatusChange(listener: (status: WsStatus) => void): () => void {
|
||||
this.statusListeners.push(listener);
|
||||
return () => {
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
useRef,
|
||||
useState,
|
||||
} from "react";
|
||||
import { toast } from "@/components/primitives";
|
||||
import { WsConnection } from "./connection";
|
||||
import type { PcmChunk, WsEventHandler, WsEventType, WsStatus } from "./types";
|
||||
|
||||
@@ -29,16 +30,6 @@ interface WsContextValue {
|
||||
|
||||
const WsContext = createContext<WsContextValue | null>(null);
|
||||
|
||||
/** FNV-1a 32-bit hash matching the backend's hashUserId function */
|
||||
function _hashUserId(userId: string): number {
|
||||
let hash = 0x811c9dc5;
|
||||
for (let i = 0; i < userId.length; i++) {
|
||||
hash ^= userId.charCodeAt(i);
|
||||
hash = Math.imul(hash, 0x01000193);
|
||||
}
|
||||
return hash >>> 0;
|
||||
}
|
||||
|
||||
export function WsProvider({
|
||||
children,
|
||||
url,
|
||||
@@ -48,6 +39,9 @@ export function WsProvider({
|
||||
}) {
|
||||
const connRef = useRef<WsConnection | null>(null);
|
||||
const [status, setStatus] = useState<WsStatus>("disconnected");
|
||||
// Tracks whether we've ever been connected — used to suppress the
|
||||
// "reconnecting" toast on initial page load.
|
||||
const wasConnected = useRef(false);
|
||||
|
||||
// Event handler registry — Ref so listeners survive re-renders without reconnect
|
||||
// Using unknown as internal store; typed at the subscribe interface
|
||||
@@ -85,7 +79,31 @@ export function WsProvider({
|
||||
const conn = new WsConnection(url);
|
||||
connRef.current = conn;
|
||||
|
||||
const unsubStatus = conn.onStatusChange(setStatus);
|
||||
const unsubStatus = conn.onStatusChange((s) => {
|
||||
setStatus(s);
|
||||
// User feedback on WS lifecycle transitions.
|
||||
if (s === "connecting") {
|
||||
// Only toast if we were previously connected (i.e. a disconnect,
|
||||
// not the initial connect on page load).
|
||||
if (wasConnected.current) {
|
||||
toast({
|
||||
title: "Reconnecting…",
|
||||
description: "WebSocket connection lost. Attempting to reconnect.",
|
||||
tone: "neutral",
|
||||
});
|
||||
}
|
||||
wasConnected.current = false;
|
||||
} else if (s === "connected") {
|
||||
wasConnected.current = true;
|
||||
} else if (s === "error" && !wasConnected.current) {
|
||||
toast({
|
||||
title: "Connection error",
|
||||
description:
|
||||
"WebSocket failed to connect. Retrying in the background.",
|
||||
tone: "vermilion",
|
||||
});
|
||||
}
|
||||
});
|
||||
const unsubEvent = conn.onEvent((event) => {
|
||||
if (event.type === "text") {
|
||||
handleJsonEvent(event.data);
|
||||
|
||||
Reference in new issue
Block a user