Author SHA1 Message Date
asepharyana 1ae19074ee feat(gmw): moderation explainability + semantic message search
- Persist structured verdict (flags/severity/confidence/evidence) on
  moderation_actions so the public web can show WHY a message was moderated.
- Add a persistent Qdrant archive collection (gmw_message_archive); embed
  every captured message at capture time (fire-and-forget, best-effort).
- Public semantic search over the archive (backend oRPC + FE toggle on the
  messages view). Both features are read-only/public and fully automatic.

Migration: 0015_add_moderation_explainability.sql
2026-08-18 15:11:01 +07:00
asepharyana d68f6b653a perf(ai-moderation): compact system prompt + memoize build + hoist vision pass
- Memoize buildSystemPrompt by (mode|channelCulture); identical signatures
  now reuse the ~5k-token core instead of rebuilding per sub-batch call
  (textBatchProcessor rebuilt it inside the loop; a 200-msg batch re-sent
  the full system prompt ~4x). Correction tail stays per-attempt (uncached).
- Hoist URL-image -> vision evidence out of the per-sub-batch loop in
  textBatchProcessor: it depends only on fetched images + full target set,
  so compute once per whole batch, not per sub-batch.
- Compact system instructions: collapse 3x-duplicated 'evaluate by content
  alone' statements into one standalone rule; trim output.ts channel-culture
  + context framing already covered by rules.ts/system.ts; drop duplicate
  programming-error-log few-shot (id 17, covered by rules AMAN list).
- Fix misleading config default: AI_LLM_BASE_URL default -> omniroute
  (gateway already runs omniroute via BWS; 9router was dead/misleading).

typecheck + lint + build green.
2026-08-18 11:49:39 +07:00
asepharyana 29baba3a72 fix(backend): sort vitest import in stream-many test (CI biome gate)
Reorder `describe, it, expect` → `describe, expect, it` to satisfy the
Biome organizeImports check that gates the all-services CI pipeline.
2026-08-18 11:03:24 +07:00
asepharyana 217ecc1aa1 feat(frontend): content micro-animations — stagger, button press, toast polish
- animate-stagger utility + staggerDelay() helper; lists now rise in sequence
  (recordings grid, moderation rows, message list, media queue, dashboard tiles).
- Button gets a subtle active:scale-[0.97] press feedback.
- Toaster: toast-in slide-up, tone-accent border, rounded hover-close target.
- All motion is reduced-motion aware (killed under prefers-reduced-motion).
2026-08-18 11:01:00 +07:00
asepharyana 0cb0b82fb1 feat(messages): stream history one-message-per-WS-frame instead of 50-row batch
- backend: add streamMany generator (paginated, yields one record at a time)
  + messagesService.streamMessages + WS 'stream_messages' handler emitting
  'message_snapshot' per message, 'message_snapshot_end' with nextCursor
- frontend: useMessagesStream hook accumulates snapshots into SWR list,
  SSR getMessages seeds first paint, WsHook gains sendText
- add stream-many.test.ts locking the one-at-a-time + cursor contract
2026-08-18 10:21:08 +07:00
asepharyana 95f2903067 feat(frontend): micro-interactions — page-enter transition + MetricTile lift
- Add animate-fade-up utility (reduced-motion aware) and a PageTransition
  wrapper; every dashboard view now rises + settles on mount/route change.
- MetricTile gains a subtle hover lift (-translate-y) + ring-focus glow.
- Theme toggle and command palette (⌘K) already existed and persist; no-op.
2026-08-18 09:50:52 +07:00
asepharyana d78d7a0181 feat(frontend): skeleton loading states, empty-state glow, light-mode polish
- Add shared skeleton building blocks (SkeletonHero, SkeletonMetricRow,
  SkeletonPanel, SkeletonRows) and wire them into every view's initial
  loading branch, replacing bare spinners for a cohesive shimmering shell.
- Polish EmptyState with a glow-ring icon chip instead of a flat icon.
- Harden .light theme: color-scheme, tuned scrollbar + selection for pale
  canvas. Dark/light theme toggle (next-themes) already persisted in TopBar.
2026-08-18 09:21:12 +07:00
asepharyana ff8a9c50ae feat(frontend): polish remaining views + centralize time/status helpers
- Add formatDuration + formatRelativeTime to lib/format.ts and aiTone to
  lib/ai-status.ts; dedupe duplicated helpers in messages/analysis views.
- Recordings: upload-status badge (pending/processing/failed), channel chip,
  relative time, hover lift, surface upload errors.
- Media: mode pill + duration on now-playing, volume meter, per-track duration
  and queue total in header.
- Moderation: relative timestamps in action rows + executor hint.
- Analysis & Dashboard: rank bars for top reactors for scannable comparison.
2026-08-18 09:05:12 +07:00
asepharyana 6bd6ddcdca feat(frontend): mobile bottom nav, chatbot overhaul, responsive polish
- MobileNav: safe-area-aware bottom tab bar (< md), mirrors desktop nav
- Chatbot: timestamps, per-message copy, retry-on-fail, auto-grow composer,
  MarkdownLite (XSS-safe React nodes, no dangerouslySetInnerHTML)
- AppFrame: NavRail (md+) + MobileNav (< md) + bottom content padding
- Design system: SignalTone ambient (signal/amber/vermilion) driving WebGL
  haze + topbar status pill; SectionHeader/MetricTile; globals.css tokens
- Views: dashboard hero + AmbientField, analysis/media responsive grids
2026-08-17 21:29:37 +07:00
asepharyana b38616e051 fix(auto-delete): guard nickname reset on role hierarchy + surface LLM parse errors
- resetOffensiveNickname: skip when target role sits above bot
  (member.manageable) instead of hammering a doomed setNickname PATCH
  that Discord rejects with 50013 'Missing Permissions'. Log the
  Discord error code on failure for clear diagnosis.
- llmCaller: include contentPreview (first 200 chars) in the parse-
  failure warning so non-JSON LLM responses are debuggable.
2026-08-17 20:52:09 +07:00
asepharyana 479f4719ba refactor(ai): replace SearXNG with Wikipedia adapter for analysis enrichment
- Add wikipediaClient.ts: native fetch to Wikipedia REST/Action APIs
  (search + summary), no extra npm dependency.
- Extract shared Redis cache into cacheStore.ts (decoupled from search).
- Term glossary now uses wikipediaSummary for direct article lookup.
- Remove searxngSearch.ts entirely; drop SEARXNG_BASE_URL config,
  add WIKIPEDIA_LANG / WIKIPEDIA_TIMEOUT_MS.
- Rename backend searxngCalls metric to webSearchCalls.
2026-08-17 20:07:19 +07:00
asepharyana 2825250804 perf(ai-moderation): remove per-user reputation from analysis context
User: 'jangan ada reputasi juga' — no profile, no reputation in the prompt,
raw messages only.

- textBatchProcessor: drop initializeUserReputation fetch + <user_reputation>
  tag injection (kept the minimal <message> tag + reply/reference context).
- visionAnalyzer (prepareMediaMessage): same removal.
- prompts/system.ts + prompts/output.ts: replace <user_reputation>/<user_history>
  instructions with an explicit 'no per-user profile/reputation context'
  note so the LLM judges purely on message content + conversation/web/location.
- mediaBatchProcessor: fix stale comment.

Trust/infraction state is STILL written to the DB (userReputationsTable) for
enforcement — only the LLM context injection is removed, so moderation
actions (mute/ban via infraction thresholds) keep working.

Net: even smaller prompts (no per-user context at all) → more messages fit
per request, and one fewer DB round-trip per unique user per sub-batch.

tsc, biome, vitest (129) all clean.
2026-08-16 20:44:39 +07:00
asepharyana aa280c48b7 perf(ai-moderation): drop personal user-profile descriptions from context
User insight: personal profile summaries bloat the prompt (less room per
request) and add a per-user DB/Redis round-trip for little moderation signal.
Only the behavioural <user_reputation> history is kept.

- textBatchProcessor: stop fetching getUserProfile; remove <user_profiles>
  block + <user_profile_ref> from message tags. Keep <user_reputation>.
- mediaBatchProcessor + visionAnalyzer: same removal (profile fetch + ref).
- prompts/system.ts + prompts/output.ts: drop stale <user_profiles>/
  <user_profile_ref> instructions; point LLM at <user_reputation> instead.
- aiAnalyzer: gate userProfileLearner behind AI_USER_PROFILE_LEARNING_ENABLED
  (default false) — generates profiles nobody reads, pure LLM/DB waste.
- Add AI_USER_PROFILE_LEARNING_ENABLED config knob.

Net: smaller prompts (more messages fit per request), fewer DB round-trips
per sub-batch, and no background LLM calls learning unused profiles.

tsc, biome, vitest (129) all clean.
2026-08-16 19:56:27 +07:00
asepharyana 4cf5b87f2b perf(ai-moderation): pack more messages per LLM request (fewer API calls when busy)
User insight: rather than many small per-batch API requests, pack many
messages into ONE request so a burst is analyzed with far fewer calls.

- AI_LLM_TEXT_BATCH_SIZE 20 -> 60 (one request now carries ~3x more messages).
- AI_ANALYSIS_MAX_TARGET_TOKENS 4000 -> 14000 (the scheduler's token-budget
  gate was trimming pending messages to ~20 before they reached the sub-batch
  splitter; raising it lets ~60 messages through to a single LLM call).
- AI_LLM_TEXT_ANALYSIS_TIMEOUT_MS 30000 -> 45000 (one larger call needs more
  headroom; gemini-flash-lite has a 1M-token context so 14k+8k is trivial).

Net effect when ramai: a 60-message burst = 1-2 API calls instead of 3+,
less semaphore contention, faster throughput.
2026-08-16 19:00:29 +07:00
asepharyana 0dff7770a1 perf(ai-moderation): speed up analysis queue (ramai + sepi)
- Parallelize per-user reputation/profile fetches in textBatchProcessor
  (was a serial ~2N DB/Redis round-trip loop per sub-batch; now Promise.all
  over unique users). Cuts per-batch latency, biggest win on small/quiet
  batches.
- Make the LLM concurrency semaphore dynamic (cached per config value) instead
  of frozen at import time, so AI_LLM_MAX_CONCURRENT is tunable without code
  change and reflects current config.
- Bump AI_LLM_MAX_CONCURRENT default 5 -> 8 (gemini-flash-lite is cheap; helps
  throughput when busy).
- Lower AI_ANALYSIS_DEBOUNCE_MS 500 -> 250 (snappier first-message analysis
  when quiet).
- Lower AI_ANALYSIS_RECOVERY_INTERVAL_MS 15000 -> 10000 (stuck/errored
  messages re-analyze sooner).

tsc, biome, vitest (129) all clean.
2026-08-16 18:51:13 +07:00
asepharyana e3dd6a3427 fix(messages): Discord-style order (oldest top, newest bottom)
Backend returns messages DESC (newest first); the view previously rendered
that directly, so the feed was inverted vs Discord (old at bottom, new at top)
while the load-older control sat at the top — contradictory.

- Reverse the display list so it reads oldest→newest top→bottom, like DC.
- Load-older (cursor to lower created_at) prepends at the top; scroll position
  is preserved by offsetting scrollTop by the height added above.
- Open at the bottom (newest visible) on first load / scope change.
- New live messages append at the bottom and auto-scroll only when the user is
  already near the bottom (nearBottomRef), so reading history isn't disrupted.
- Scroll container now tracked via ref; onScroll updates nearBottom + triggers
  load-older when scrolled to the top.

tsc, biome, next build all clean.
2026-08-16 17:56:39 +07:00
asepharyana 55fdcfaae3 style: biome format chatbot.service (parseResponse call) 2026-08-16 17:17:03 +07:00
asepharyana cf1ec25c71 fix(chatbot): disable thinking + use non-streaming LLM call
- Set stream:false on the /chat/completions request so the bot gets one
  complete response instead of an SSE token stream.
- Add reasoning_effort:"none" to suppress extended-thinking/reasoning tokens
  (ignored by non-reasoning models like gemini-flash-lite).
- Add parseResponse(): handles both the JSON object 9router returns for
  stream:false and the SSE text it may still emit, delegating SSE to parseSse.
  Verified live: omniroute returns 200 application/json with message.content.
2026-08-16 17:08:29 +07:00
asepharyana 0ace758c79 feat(frontend): safe-area insets + hardened reduced-motion
- Add viewport export with viewportFit: "cover" so iOS exposes
  env(safe-area-inset-*) (required for the insets to take effect).
- NavRail / TopBar / main / Toaster now respect safe-area insets so content
  clears the iPhone notch and home indicator in both portrait and landscape.
- prefers-reduced-motion: the media query already disabled declared animation
  classes; harden it with a global transition/animation duration override and
  kill the scan-line shimmer so motion-sensitive users get a fully static UI.

Verified tsc --noEmit + next build clean.
2026-08-16 16:55:34 +07:00
asepharyana c89288191e fix(frontend): responsive layout across all dashboard pages
- SectionHeader: action (filters/legends) now wraps below the title on narrow
  screens instead of overflowing beside it (flex-wrap, gap-2 sm:gap-3).
- GuildChannelPicker: selects go full-width and stack on mobile (w-full
  sm:w-44 / sm:w-52) instead of fixed widths that exceeded a 375px viewport.
- Messages search: w-full sm:w-64 so it doesn't crowd the picker on mobile.
- TopBar: tighter padding (px-4 sm:px-5), smaller title on mobile, connection
  status uses compact (dot only) on mobile, ambient pill hidden < sm.
- Shell main + dashboard channel label: responsive padding / shrink-0 widths.

Verified tsc --noEmit + next build clean; targets breakpoints 375/768/1024/1440.
2026-08-16 16:36:34 +07:00
asepharyana 4655125541 feat(frontend): clearer load-older spinner + cap history pages (Messages)
- Show an explicit Loader2 spinner row ("Loading older…") while the next page
  fetches, instead of a disabled button.
- Cap appended older pages at MAX_OLDER_PAGES=10 (500 messages) so a long
  scroll-up never pulls the entire history; show a "capped" hint pointing to
  search. Reset the counter when guild/channel changes.
2026-08-16 16:21:10 +07:00
asepharyana ba60448d05 feat(frontend): load older messages in Messages view (cursor pagination)
Wire the existing useLoadMore + useMessagesHasMore pagination hooks into the
Messages view: add a "↑ Load older messages" button at the top of the list and
auto-load the next (older) page when the user scrolls to the top. Backend
messages.list already returns a created_at-based nextCursor (DESC order), so
older pages are just subsequent cursors. Newest-first live feed is preserved;
the load-older control is hidden during search.
2026-08-16 16:06:25 +07:00
asepharyana 1d809b2c95 fix(proxy): route /trpc to backend so browser oRPC WebSocket opens
Browser connects oRPC over wss://…/trpc (partysocket). The gmw-proxy nginx
only forwarded /api and /ws to the backend, so /trpc upgrades fell through to
Next.js SSR and the socket never opened ("WebSocket is not open"). Add a
/trpc location (WS upgrade headers) mirroring /ws. Backend already serves
oRPC on /trpc (HTTP RPCHandler + WS ORPCWebSocketServer on :4001).

Verified: ws://127.0.0.1:4001/trpc upgrade OPEN; SSR + server-side fetch RPCLink
also use /trpc directly so only the browser path was broken.
2026-08-16 15:28:58 +07:00
asepharyana 38bda66933 style: fix Biome dead-code warnings from oRPC migration (CI lint gate) 2026-08-16 14:51:47 +07:00
asepharyana 726a8e116b fix(build): make oRPC/tRPC dist runnable under node ESM (deploy crashloop)
flake.nix only rewrote @/ aliases but left extensionless relative imports
(./router) in compiled dist/. node dist/index.js (how prod runs) cannot
resolve extensionless ESM specifiers -> ERR_MODULE_NOT_FOUND -> backend
crashlooped (444 restarts, port 4001 dead). Extract the fixer into a shared
scripts/fix-imports.mjs that appends .js to extensionless relative imports and
rewrites @/ aliases, and wire it into backend + discord-gateway build phases.

Verified: fresh tsc + fixer -> node dist/index.js boots; oRPC over /trpc
serves both HTTP POST and WebSocket (config/dashboard/voice/moderation/
media/chatbot/analysis) end-to-end against Postgres + Redis. next build
passes with the oRPC client + partysocket.
2026-08-16 14:45:02 +07:00
asepharyanaandClaude Opus 4.5 2fa1827f17 feat(backend,frontend): migrate data APIs from REST to native tRPC over WebSocket
Replace REST module routers with a single typed tRPC appRouter served over
/trpc (HTTP + WebSocket), and rewire the frontend to call it via
@trpc/client wsLink (browser) and httpLink (RSC data layer). Existing
/api/health + /api/metrics stay as plain Express for infra scraping.

Notable fixes surfaced by the live smoke test:
- Express 5 / path-to-regexp v8 rejects the /trpc/* wildcard route; use a
  prefix middleware that computes opts.path from the URL instead.
- nodeHTTPRequestHandler treats opts.path as the literal procedure path, so
  it is derived per-request from req.url.
- Two ws servers on one http.Server (the /ws voice socket + /trpc) collided
  and returned 400 on upgrade; both now use noServer + a manually routed
  server.on('upgrade') keyed by path.

Verified: BE tsc+biome+40 vitest green; FE tsc+biome green; live
HTTP and WebSocket calls returned real prod data.

Co-Authored-By: Claude Opus 4.5 (1M context) <noreply@anthropic.com>
2026-08-16 13:25:10 +07:00
asepharyanaandClaude Opus 5 (Nous Research) d8552a9fb8 feat(ai): make standalone image/vision analysis timeout explicit (1 min)
The standalone image analysis path (analyzeSingleMediaImage → llmVision →
llmChat) previously had no request-level timeout of its own — it silently
inherited the shared OpenAI client default (60s), and AI_LLM_MEDIA_ANALYSIS_
TIMEOUT_MS only governed the text+media *batch*, not a single vision call.

- Add AI_LLM_VISION_ANALYSIS_TIMEOUT_MS (default 60000) to config.
- llmChat now accepts an optional per-request `timeout` in LlmCallOpts,
  forwarded to the OpenAI request options (falls back to the 60s client
  default when omitted).
- llmVision passes config.AI_LLM_VISION_ANALYSIS_TIMEOUT_MS, so a single
  image/sticker/emoji analysis gets a guaranteed 1-minute budget and is
  independently tunable from the text path.

Verified: tsc + biome green, 129 gateway tests pass.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 10:47:24 +07:00
asepharyanaandClaude Opus 5 (Nous Research) a4abe3abea fix(frontend): remove duplicate Dashboard entry in nav rail
The sidebar rendered /dashboard twice: once as a hardcoded NavItem
(lines 45-50) and again via navItems.map() (navItems[0] is also
/dashboard). Dropped the hardcoded item so the single source of truth
(navItems in lib/navigation.ts) drives the rail. Removed the now-unused
LayoutDashboard import.

tsc + biome green.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 09:25:07 +07:00
asepharyanaandClaude Opus 5 (Nous Research) b67856462f feat(chatbot): expand tool set to cover all server-watcher situations
The chatbot agent now has 14 tools (was 4) so it can answer about ANY
server situation from live data instead of a static snapshot:

- get_server_stats (now also returns clean count)
- get_top_channels, get_recent_activity, get_top_flagged
- search_messages (LIKE keyword search)
- get_user_messages, get_user_profile, get_user_reputation
- get_channel_culture
- get_message_detail (full AI analysis of one message)
- get_message_reviews (human moderation queue by status)
- get_voice_recordings (with transcriptions)
- get_moderation_timeline (daily flagged/warn/clean trend)
- get_corrections (AI false-positive correction history)

Security/quality:
- Every executor now uses parameterized drizzle queries (eq/like/and).
  The old code interpolated model-supplied IDs into sql.raw() — a SQL
  injection vector. Removed.
- Split static tool *definitions* into chatbot.toolDefs.ts (no DB import)
  so the LLM-facing schema can be unit-tested without loading the
  database/config layer. chatbot.tools.ts keeps only the executor.

Verified: tsc + biome clean, 40 backend tests pass (4 new covering the
tool-contract: names unique, required args declared, full situation
coverage).

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 09:22:20 +07:00
asepharyanaandClaude Opus 5 (Nous Research) 30828a5534 refactor(chatbot): drop static server-stats context, go fully tool-based
The chatbot already had an agentic tool loop (get_server_stats,
get_top_channels, get_recent_activity, get_top_flagged), but processMessage
still baked a serverInsights snapshot into the system prompt and told the
model to "answer from that data". That defeats the tools: the model answered
from a stale snapshot instead of living numbers, and the guild/channel scope
the frontend sends was never forwarded to the tools.

Changes (services/backend/src/modules/chatbot):
- Remove getServerInsights() + ServerInsights (dead after this change).
- buildSystemPrompt(): drop the hardcoded stats block; instruct the model it
  has NO memorized server numbers and MUST call a tool for any server-data
  question, answering only from tool results.
- processMessage(): stop fetching insights; pass the request guildId/channelId
  scope through to callLLM.
- callLLM(): accept scope; auto-fill empty guildId/channelId on tool calls from
  the request scope so the model never has to guess IDs and tools always query
  the right server.

Behavior: answers now come from live DB data via tools, scoped to the server
the user is chatting in. tsc + biome + 36 backend tests green.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 09:15:39 +07:00
asepharyanaandClaude Opus 5 (Nous Research) a3e5a8c1b9 perf(gateway): hoist correctedExamples query out of retry closure
buildCorrectedFewShotExamples() (a getRecentCorrectedModerations(5)
DB hit) was called inside the per-sub-batch buildContent closure in
textBatchProcessor.ts — re-queried for every sub-batch (≈10× for a
200-msg burst) AND re-fired on each parse-error retry. mediaBatchProcessor
already hoisted it once. Mirror that: fetch once per runTextOnlyBatch,
reuse the cached string inside the closure.

No behavior change — identical content, fewer identical DB reads.
tsc + 129 tests + biome green.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 09:06:17 +07:00
asepharyanaandClaude Opus 5 (Nous Research) f82b5caae4 refactor(gateway): strip boilerplate fields from few-shot examples
The 32 few-shot examples each re-echoed score/confidence/
recommended_action/categories/policy_version inline (~150 chars ×
32). Those fields carry zero moderation-decision signal — the schema
and their ??-default coercion already live in OUTPUT_INSTRUCTIONS +
moderationResponseParser.ts. Removed 96 redundant key/value pairs.

Kept per-example: message_id, status, flags, severity, evidence,
analysis — the fields that actually teach decisions. Parser derives
the rest via ?? fallback, so real output shape is unchanged.

examples.ts: 21.7K→18.5K chars; FEW_SHOT(mixed) 15.3K→13.4K.
Total mixed system prompt now 33.9K (was 39.3K at audit start,
~14% leaner). tsc + 129 tests + biome green.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 09:00:55 +07:00
asepharyanaandClaude Opus 5 (Nous Research) 9e2b107fcd refactor(gateway): compact AI analysis system prompt, preserve all rules
- prompts/system.ts: merge 3 overlapping framing blocks (Blok Data /
  Konteks Pengguna / Framing Konteks vs Target) into 1 tight block —
  same coverage, no duplicated "standalone judgment / profile-is-
  reference-not-evidence" prose.
- prompts/output.ts: trim duplicated user_history/standalone paragraph
  in PERSONALITY & MEMORI (keep concrete per-case lessons).
- prompts/examples.ts: drop 2 exact-duplicate-lesson few-shots (LGBT id=19
  dup of id=30; weapons-tech id=33 dup of id=32). All teaching signals
  retained via the surviving example of each lesson.

Static system prompt: text 32.7K→29.2K, mixed 39.3K→35.8K chars
(~10% smaller). No moderation rule, zero-tolerance category, or decision
tree altered — accuracy-controlling content untouched. tsc + 129 tests +
biome green.

Co-Authored-By: Claude Opus 5 (Nous Research)
2026-08-16 08:52:11 +07:00
130 changed files with 7866 additions and 2521 deletions

No files matched your search

@@ -0,0 +1,500 @@
# GMW — Moderation Explainability (#1) + Semantic Search (#3) Implementation Plan
> **For Hermes:** Use subagent-driven-development to implement task-by-task.
> Hard constraint from user (2026-08-18): web is PUBLIC, read-only, for USERS not admins. Moderation MUST stay FULLY AUTOMATIC. Rules stay in CODE (no per-channel config UI).
**Goal:** Make GMW transparent (users see why a message was moderated) and searchable (users can semantic-search the message corpus), via two fully-automatic, code-driven, read-only-public features.
**Architecture:**
- **#1 Explainability:** Persist the structured moderation verdict that already exists in `AnalysisResult` (`flags[]`, `categories[]`, `severity`, `confidence`, `evidence[]`) into new columns on `moderation_actions`, surface them through the existing public moderation oRPC + the existing public `moderation` dashboard view. No new behavior — only new *data* + new *read* paths.
- **#3 Semantic Search:** Add a SECOND persistent Qdrant collection (`gmw_message_archive`) keyed by message id (NOT the TTL cache). Embed each captured text message at capture time (reuse `embedText`) and upsert. Add a public `messages.semanticSearch` oRPC + a read-only search UI on the public `messages` view. Best-effort / non-blocking — embed failures never affect moderation or capture.
**Tech Stack:** TypeScript (discord-gateway + backend + frontend monorepo), Drizzle ORM + Postgres (PgBouncer on imrnes), Qdrant (100.121.180.82:6333), Next.js 16 App Router + shadcn/ui, oRPC over `/trpc`. pnpm. Deploy via GitHub Actions Nix build + `systemctl restart`.
**Critical existing facts (verified in repo):**
- `AnalysisResult` shape (`src/modules/ai-moderation/ai-analysis-worker.ts:55`): `messageId, status, flags[], categories[], severity, confidence, recommendedAction, score, analysis, correctedFlags?`. The shared `AnalysisResult` (`src/shared/moderation-types.ts:142`) ALSO has `evidence?: string[]` and `policyVersion?: string`. **THESE ARE ALREADY COMPUTED but only logged, never persisted to `moderation_actions`.**
- `moderation_actions` schema is DEFINED TWICE with a divergence:
- `src/shared/database/schema.ts:638` → `pgModerationActionsTable` (authoritative, has `reset_nickname` in `action_type` enum).
- `src/shared/database/schema/messages.ts:25` → another `pgModerationActionsTable` (NO `reset_nickname`; gateway-local copy).
- The gateway's `ModerationActionsDb` (`src/modules/message-capture/moderationActionsDb.ts`) imports from `schema.ts` (the authoritative one). The `messages.ts` copy appears UNUSED for DB ops — but WE MUST ADD NEW COLUMNS TO BOTH to avoid type drift, OR confirm the `messages.ts` copy is dead and delete it. **Decision: add columns to `schema.ts` (authoritative) AND the `messages.ts` copy to keep `$inferInsert`/`$inferSelect` in sync (the gateway `ModerationAction` type flows from shared).** Line truncated
- Migration mechanism: Drizzle-managed via `drizzle/migrations/*.sql` (journal `_journal.json`) applied by `runMigrations()` → `migratePostgres`. **New tables/columns must be added with `drizzle-kit generate` to produce a numbered `.sql` + journal entry**, OR (simpler, matches `0013_rename_*.sql` manual style) write a raw idempotent `.sql` under `drizzle/migrations/` AND register it in `_journal.json`. **Preferred here: use `pnpm drizzle-kit generate` so the journal stays consistent.** The leLine truncated
- **Historical lesson (MUST respect):** a prior migration (`0004_drop_unused_ai_columns.sql` = old `001_drop`) DELETED `ai_evidence`, `ai_policy_version`, `ai_moderation_raw` from `messages` with the note "written but never read". → Our new `moderation_actions` columns MUST be read (serializer + FE render). No write-only columns.
- Embedding client: `embedText(text)` / `embedTexts(texts[])` in `src/modules/ai-moderation/embeddingClient.ts`. Returns `null` if `AI_LLM_EMBEDDING_MODEL` not configured. Reuses `config.AI_LLM_BASE_URL` + `config.AI_LLM_API_KEY`. OpenAI SDK v6 → `encoding_format: "float"` REQUIRED (Nvidia rejects base64).
- Qdrant client: `src/modules/ai-moderation/qdrantClient.ts`. Has `ensureQdrantCollection(vectorSize)`, `upsertQdrantPoint(cacheKey, vector, payload)`, `searchQdrant(vector, limit, scoreThreshold)`. These are hardcoded to the cache collection name `config.QDRANT_COLLECTION ?? "gmw_text_moderation"`. **#3 needs a second collection** → generalize the client to accept a collection name param (add `ensureQdrantCollectionV2(name, size)` / `upsertQdrantPointV2(name, id, vector, payload)` / `searchQLine truncated
- Capture hook: `captureMessage()` (`src/modules/message-capture/messageCapture.ts:201`) calls `messageStore.upsertMessageForCapture(messageRecord)` then (if not backlog) `queueMessageAnalysis`. **#3 embed must happen here**, async + fire-and-forget, after successful insert.
- Public moderation view: `services/frontend/src/app/(dashboard)/moderation/view.tsx` renders `ActionRow` per action. The `ModerationAction` FE type is in `services/frontend/src/lib/types/moderation.ts` (NO new fields yet). The backend `moderationService.listActions` SQL is in `services/backend/src/modules/moderation/moderation.repository.ts:68` (raw SQL, selects fixed columns, joins `messages`).
- oRPC wiring: `services/backend/src/orpc/router.ts` → `moderationRouter` (stats, actions) and `messagesRouter` (list, byChannel, getById, review, attachments). New procedures added here.
---
## TASK 1 — Schema: add explainability columns to `moderation_actions`
**Objective:** Persist structured verdict on moderation actions so it can be surfaced (read) later.
**Files:**
- Modify: `services/discord-gateway/src/shared/database/schema.ts` (authoritative `pgModerationActionsTable`, ~line 638)
- Modify: `services/discord-gateway/src/shared/database/schema/messages.ts` (`pgModerationActionsTable` copy, ~line 25) to keep type in sync
- Create: `services/discord-gateway/drizzle/migrations/0015_add_moderation_explainability.sql`
- Update: `services/discord-gateway/drizzle/migrations/meta/_journal.json` (add new entry)
**Step 1: Add columns to both schema definitions**
Add after `executed_at` in BOTH `pgModerationActionsTable` definitions:
```ts
// ── Explainability (structured verdict, surfaced read-only to public web) ──
flags: pgText("flags"), // JSON array of string flags, e.g. ["sara_agama","vulgar"]
categories: pgText("categories"), // JSON array of category strings
severity: pgText("severity", {
enum: ["none", "low", "medium", "high", "critical"],
}),
confidence: pgReal("confidence"), // 0..1
score: pgReal("score"), // 0..1 raw model score
evidence: pgText("evidence"), // JSON array of short quoted snippets
policy_version: pgText("policy_version"), // rules.ts policy version string
```
Note: `flags`/`categories`/`evidence` stored as JSON-stringified TEXT (consistent with how `messages.ai_moderation_flags`/`ai_categories` are stored as TEXT elsewhere — confirm storage format in `updateMessageAIAnalysis`). Keep nullable.
**Step 2: Generate/author the migration SQL**
`0015_add_moderation_explainability.sql` (idempotent):
```sql
-- Add structured explainability columns to moderation_actions (read-only surfaced to public web).
ALTER TABLE IF EXISTS "moderation_actions"
ADD COLUMN IF NOT EXISTS "flags" text,
ADD COLUMN IF NOT EXISTS "categories" text,
ADD COLUMN IF NOT EXISTS "severity" text
CHECK ("severity" IS NULL OR "severity" IN ('none','low','medium','high','critical')),
ADD COLUMN IF NOT EXISTS "confidence" real,
ADD COLUMN IF NOT EXISTS "score" real,
ADD COLUMN IF NOT EXISTS "evidence" text,
ADD COLUMN IF NOT EXISTS "policy_version" text;
```
Register in `_journal.json`: append an entry with `idx: 15`, a new unique `tag` (hash), `version`, `when` = Date.now(), `tag` short, `breakpoints: false`. Use `pnpm drizzle-kit generate` if possible to get a correct tag; otherwise hand-edit the journal carefully (copy an existing entry's shape).
**Step 3: Type-check gateway**
Run: `cd services/discord-gateway && pnpm typecheck`
Expected: PASS (no new compile errors).
**Step 4: Commit**
```bash
git add services/discord-gateway/src/shared/database/schema.ts \
services/discord-gateway/src/shared/database/schema/messages.ts \
services/discord-gateway/drizzle/migrations/0015_add_moderation_explainability.sql \
services/discord-gateway/drizzle/migrations/meta/_journal.json
git commit -m "feat(db): add explainability columns to moderation_actions"
```
---
## TASK 2 — Persist verdict at the auto-delete + command-handler call sites
**Objective:** Populate the new columns from the already-computed `AnalysisResult` when a moderation action is logged. Fully automatic, no new behavior.
**Files:**
- Modify: `services/discord-gateway/src/modules/ai-moderation/autoDeleteManager.ts` (`logAutoDeleteAttempt` ~line 166, and the second `createModerationAction` call ~line 234 for nickname/mute paths)
- Modify: `services/discord-gateway/src/modules/command-handler/moderation.handler.ts` (`createModerationAction` ~line 95)
- Helper (create): `services/discord-gateway/src/modules/ai-moderation/verdictToActionFields.ts` — shared mapper so all 3 call sites stay DRY.
**Step 1: Create the mapper helper**
`verdictToActionFields.ts`:
```ts
import type { AnalysisResult } from "@/modules/ai-moderation/ai-analysis-worker";
import type { ModerationActionInsert } from "@/shared/index"; // or inline shape
/**
* Map a computed AI verdict into the explainability columns of a moderation
* action. Null-safe: missing fields stay null (e.g. manual admin actions have
* no AnalysisResult). This is read-only structured data — it does NOT change
* any enforcement decision.
*/
export function verdictToActionFields(result?: {
flags?: string[];
categories?: string[];
severity?: string;
confidence?: number;
score?: number;
evidence?: string[];
policyVersion?: string;
}): {
flags: string | null;
categories: string | null;
severity: string | null;
confidence: number | null;
score: number | null;
evidence: string | null;
policy_version: string | null;
} {
if (!result) {
return { flags: null, categories: null, severity: null, confidence: null,
score: null, evidence: null, policy_version: null };
}
const j = (v: unknown) => (v == null ? null : JSON.stringify(v));
return {
flags: j(result.flags),
categories: j(result.categories),
severity: result.severity ?? null,
confidence: result.confidence ?? null,
score: result.score ?? null,
evidence: j(result.evidence),
policy_version: result.policyVersion ?? null,
};
}
```
**Step 2: Wire `logAutoDeleteAttempt`**
Find the `createModerationAction({...})` in `logAutoDeleteAttempt` and spread the verdict fields:
```ts
await messageStore.createModerationAction({
message_id: message.id,
user_id: message.user_id,
guild_id: message.guild_id,
action_type: "delete_message",
reason: result.reason,
...verdictToActionFields(result.analysisResult), // <-- pass the AnalysisResult through
executed_by: "auto-delete-manager",
status: ...,
});
```
**IMPORTANT:** `result` here is `AutoDeleteResult` — verify it carries the `AnalysisResult` (or the verdict). If `AutoDeleteResult` does NOT carry the full `AnalysisResult`, trace where `attemptAutoDeleteFlaggedMessage` is called from and pass the `AnalysisResult` down (it is available in the analysis worker that triggered the delete). Confirm by reading `AutoDeleteResult` type + its producer. If the verdict is only available at the orchestrator level, add an optional `verdict?: AnalysisResulLine truncated
**Step 3: Wire the second call site in `autoDeleteManager.ts`** (the mute/nickname path ~line 234) similarly, if it has an `AnalysisResult` available; otherwise leave fields null (manual-style action).
**Step 4: Wire `moderation.handler.ts`** command path (~line 95) — pass `verdictToActionFields(verdict)` if the command handler has the `AnalysisResult` for the target message; otherwise nulls. Confirm what the handler receives.
**Step 5: Type-check + lint**
Run: `cd services/discord-gateway && pnpm typecheck && pnpm lint`
Expected: PASS.
**Step 6: Commit**
```bash
git add services/discord-gateway/src/modules/ai-moderation/verdictToActionFields.ts \
services/discord-gateway/src/modules/ai-moderation/autoDeleteManager.ts \
services/discord-gateway/src/modules/command-handler/moderation.handler.ts
git commit -m "feat(mods): persist structured verdict into moderation_actions"
```
---
## TASK 3 — Backend: surface explainability in `moderation.actions`
**Objective:** Read the new columns in the public oRPC so the frontend can render them. (Read path — satisfies the "never write-only" rule.)
**Files:**
- Modify: `services/backend/src/modules/moderation/moderation.repository.ts` (`listActions` raw SQL ~line 68) — add new columns to SELECT + map.
- Modify: `services/frontend/src/lib/types/moderation.ts` (`ModerationAction` interface) — add new fields.
- Modify: `services/frontend/src/app/(dashboard)/moderation/view.tsx` (`ActionRow`) — render flags/categories badges + severity + confidence + evidence snippet.
**Step 1: Extend backend SELECT**
In `listActions`, add to the SELECT list: `a.flags, a.categories, a.severity, a.confidence, a.score, a.evidence, a.policy_version`. In the `.map(...)` add:
```ts
flags: r.flags ? safeJsonArray(String(r.flags)) : null,
categories: r.categories ? safeJsonArray(String(r.categories)) : null,
severity: r.severity ? String(r.severity) : null,
confidence: r.confidence != null ? Number(r.confidence) : null,
score: r.score != null ? Number(r.score) : null,
evidence: r.evidence ? safeJsonArray(String(r.evidence)) : null,
policy_version: r.policy_version ? String(r.policy_version) : null,
```
where `safeJsonArray(s)` = `JSON.parse(s)` wrapped in try/catch returning `[]` on failure (define a tiny local helper in the repository file).
**Step 2: Extend FE type**
In `services/frontend/src/lib/types/moderation.ts` `ModerationAction`:
```ts
flags: string[] | null;
categories: string[] | null;
severity: "none" | "low" | "medium" | "high" | "critical" | null;
confidence: number | null;
score: number | null;
evidence: string[] | null;
policy_version: string | null;
```
**Step 3: Render in `ActionRow`**
After the existing reason block, add (using existing `Badge` + `aiTone` from `@/lib/ai-status`):
```tsx
{a.severity && (
<Badge tone={aiTone(a.severity === "none" ? "clean" : a.severity)}>
{a.severity}
</Badge>
)}
{a.flags?.length ? (
<div className="mt-1 flex flex-wrap gap-1">
{a.flags.map((f) => <Badge key={f} tone="amber">{f}</Badge>)}
</div>
) : null}
{a.evidence?.length ? (
<div className="mt-1 text-xs text-ink-faint border-l-2 border-hairline pl-2">
“{a.evidence[0]}”
</div>
) : null}
{a.confidence != null && (
<div className="mono mt-0.5 text-[0.6rem] text-ink-faint">
conf {(a.confidence * 100).toFixed(0)}%
</div>
)}
```
Keep `ActionRow` read-only. No admin controls.
**Step 4: Type-check both services**
Run: `cd services/backend && pnpm typecheck && pnpm lint` and `cd services/frontend && pnpm typecheck && pnpm lint`
Expected: PASS.
**Step 5: Commit**
```bash
git add services/backend/src/modules/moderation/moderation.repository.ts \
services/frontend/src/lib/types/moderation.ts \
services/frontend/src/app/\(dashboard\)/moderation/view.tsx
git commit -m "feat(web): surface moderation explainability (flags/severity/evidence)"
```
---
## TASK 4 — Qdrant client: support a second persistent collection
**Objective:** Generalize the Qdrant client so #3 can use a dedicated archive collection without disturbing the automod cache.
**Files:**
- Modify: `services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts`
**Step 1: Add collection-aware variants**
Refactor `collectionName()` to accept an optional name, and add V2 functions that take an explicit collection:
```ts
function collectionName(fallback = config.QDRANT_COLLECTION ?? "gmw_text_moderation"): string {
return fallback;
}
export const ARCHIVE_COLLECTION = config.QDRANT_ARCHIVE_COLLECTION ?? "gmw_message_archive";
export async function ensureQdrantCollectionV2(name: string, vectorSize: number): Promise<boolean> {
// same body as ensureQdrantCollection but uses `name` instead of collectionName()
}
export async function upsertQdrantPointV2(
name: string, pointId: number, vector: number[], payload: QdrantVerdictPayload,
): Promise<boolean> { /* PUT /collections/{name}/points with wait:true */ }
export async function searchQdrantV2(
name: string, vector: number[], limit: number, scoreThreshold: number,
): Promise<QdrantSearchHit[]> { /* POST /collections/{name}/points/search */ }
```
Keep all existing `ensureQdrantCollection` / `upsertQdrantPoint` / `searchQdrant` UNCHANGED (cache path). V2 functions mirror them with the `name` param. Reuse `request()` and the existing payload/score types.
**Step 2: Type-check**
Run: `cd services/discord-gateway && pnpm typecheck`
Expected: PASS.
**Step 3: Commit**
```bash
git add services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts
git commit -m "feat(qdrant): add collection-aware V2 upsert/search for archive"
```
---
## TASK 5 — Capture-time embed + archive upsert
**Objective:** Make every (non-backlog, text) captured message searchable in the persistent archive. Non-blocking / best-effort.
**Files:**
- Modify: `services/discord-gateway/src/modules/message-capture/messageCapture.ts` (`captureMessage` ~line 201)
- Create: `services/discord-gateway/src/modules/message-capture/archiveEmbedder.ts` — wraps embed + upsert with fire-and-forget + rate-limit guard.
**Step 1: Create `archiveEmbedder.ts`**
```ts
import { createChildLogger } from "@/shared/logger/index";
import { embedText } from "@/modules/ai-moderation/embeddingClient";
import { ARCHIVE_COLLECTION, ensureQdrantCollectionV2, upsertQdrantPointV2, qdrantPointId } from "@/modules/ai-moderation/qdrantClient";
import { config } from "@/shared/config/config";
const log = createChildLogger("archive-embedder");
/**
* Fire-and-forget: embed a captured message and upsert into the persistent
* archive collection. Failures are swallowed — searching is a nice-to-have,
* never a precondition for capture or moderation.
*/
export function archiveMessageEmbedded(message: {
id: string; content: string; username: string; channel_id: string; guild_id: string; created_at: number;
}): void {
if (!config.AI_LLM_EMBEDDING_MODEL) return; // embeddings disabled → skip
if (!message.content || message.content.trim().length < 3) return;
void (async () => {
try {
const vector = await embedText(message.content);
if (!vector) return;
const ok = await ensureQdrantCollectionV2(ARCHIVE_COLLECTION, vector.length);
if (!ok) return;
await upsertQdrantPointV2(ARCHIVE_COLLECTION, qdrantPointId(`archive:${message.id}`), vector, {
text: message.content.slice(0, 4000),
flags: "", // not a verdict payload; keep shape compatible
analyzed_at: Date.now(),
expires_at: Date.now() + 1000 * 60 * 60 * 24 * 365 * 5, // 5y persistent
content_hash: undefined,
});
} catch (err) {
log.debug({ messageId: message.id, error: err instanceof Error ? err.message : String(err) }, "archive embed skipped");
}
})();
}
```
Payload type reuse: `QdrantVerdictPayload` has `text`, `flags`, `analyzed_at`, `expires_at`, `content_hash?`. For the archive we only need `text` + timestamps; set `flags: ""` (empty, ignored by search filter which keys on `expires_at`). **Acceptable:** the search path filters `expires_at >= now` — 5y window satisfies that.
**Step 2: Call from `captureMessage`**
In `captureMessage`, after `const inserted = await messageStore.upsertMessageForCapture(messageRecord); if (!inserted) return;` and BEFORE the backlog branch, add:
```ts
if (!isBacklog && messageRecord.content) {
archiveMessageEmbedded(messageRecord);
}
```
(`messageRecord` is the `MessageRecord` from `buildMessageRecord`; confirm it carries `content`, `channel_id`, `guild_id`, `created_at`. It does — see `messagesCrud`/`types`.)
**Step 3: Type-check + lint**
Run: `cd services/discord-gateway && pnpm typecheck && pnpm lint`
Expected: PASS.
**Step 4: Commit**
```bash
git add services/discord-gateway/src/modules/message-capture/archiveEmbedder.ts \
services/discord-gateway/src/modules/message-capture/messageCapture.ts
git commit -m "feat(archive): embed captured messages into persistent Qdrant archive"
```
---
## TASK 6 — Backend: `messages.semanticSearch` oRPC
**Objective:** Public, read-only semantic search over the message archive.
**Files:**
- Modify: `services/backend/src/modules/messages/messages.repository.ts` — add `semanticSearch(query, limit, guildId?)`.
- Modify: `services/backend/src/modules/messages/messages.service.ts` — expose `semanticSearch`.
- Modify: `services/backend/src/orpc/router.ts` — add `messages.semanticSearch` procedure.
- Create (or reuse): an embedding call from the backend. The backend does NOT import the gateway's `embeddingClient`. **Decision:** add a minimal backend embed helper `services/backend/src/modules/messages/embed.ts` that calls the same OpenAI-compatible endpoint via `config` (reuse `config.AI_LLM_BASE_URL`/`AI_LLM_API_KEY`/`AI_LLM_EMBEDDING_MODEL` if present on the backend; if not configured, return a clear "search unavailable" error). Mirror `encoding_format: "float"`.
- Modify: `services/backend/src/modules/messages/messages.schema.ts` — add `semanticSearchQuery` zod schema (limit, guildId?, query).
**Step 1: Backend embed helper** (`embed.ts`)
```ts
import OpenAI from "openai";
import { config } from "@/shared/config/index";
import { createChildLogger } from "@/shared/logger/index";
const log = createChildLogger("messages-embed");
let client: OpenAI | null = null;
function getClient() {
if (!config.AI_LLM_API_KEY || !config.AI_LLM_EMBEDDING_MODEL) return null;
if (!client) client = new OpenAI({ apiKey: config.AI_LLM_API_KEY, baseURL: config.AI_LLM_BASE_URL, maxRetries: 0, timeout: 30_000 });
return client;
}
export async function embedQuery(text: string): Promise<number[] | null> {
const c = getClient(); if (!c) return null;
try {
const r = await c.embeddings.create({ model: config.AI_LLM_EMBEDDING_MODEL as string, input: text, encoding_format: "float" });
return r.data[0].embedding;
} catch (e) { log.warn({ error: e instanceof Error ? e.message : String(e) }, "query embed failed"); return null; }
}
```
**Step 2: Repository `semanticSearch`**
```ts
async semanticSearch(queryVector: number[], limit: number, guildId?: string) {
// Search archive collection, then join messages for text + channel.
const hits = await searchQdrantV2(ARCHIVE_COLLECTION, queryVector, limit, 0.6);
const ids = hits.map(h => h.cacheKey.replace("qdrant:", "")); // point id → we stored archive:<messageId>
// decode: qdrantPointId is a uint64; we need the original message id.
// SIMPLER: store message_id inside the payload too. → update archiveEmbedder payload to include `message_id`.
...
}
```
**REFINEMENT (important):** `qdrantPointId` is a hash, not reversible. So the archive payload MUST carry `message_id` (and `channel_id`, `guild_id`, `username`, `created_at`) so the backend can return full results without a reverse lookup. **Update `archiveEmbedder.ts` payload** to include those fields, and relax `QdrantVerdictPayload` (or create `QdrantArchivePayload`) to allow them. Then `semanticSearch` returns the payloads directly (already contain text + metadata) — no DB join needed, anLine truncated
**Step 3: Service + router**
`messages.service.ts`: `async semanticSearch(query: string, limit: number, guildId?: string)` → embed → `repository.semanticSearch`.
`orpc/router.ts` under `messagesRouter`:
```ts
semanticSearch: os
.input(z.object({ query: z.string().min(1), limit: z.coerce.number().int().positive().max(50).default(10), guildId: z.string().optional() }))
.handler(async ({ input }) => {
const results = await messagesService.semanticSearch(input.query, input.limit, input.guildId);
return { results, nextCursor: null };
}),
```
**Step 4: Type-check + lint (backend)**
Run: `cd services/backend && pnpm typecheck && pnpm lint`
Expected: PASS.
**Step 5: Commit**
```bash
git add services/backend/src/modules/messages/embed.ts \
services/backend/src/modules/messages/messages.repository.ts \
services/backend/src/modules/messages/messages.service.ts \
services/backend/src/modules/messages/messages.schema.ts \
services/backend/src/orpc/router.ts
git commit -m "feat(api): public semantic message search over archive"
```
---
## TASK 7 — Frontend: semantic search UI on `messages` view
**Objective:** Public, read-only search box + results on the existing messages dashboard.
**Files:**
- Modify: `services/frontend/src/app/(dashboard)/messages/page.tsx` + `view.tsx` — add a search input (debounced) that calls a new `useMessagesSemanticSearch` hook → `messages.semanticSearch` oRPC, renders results as message cards (reuse `GlassPanel`/`Badge`/existing message row components).
- Modify: `services/frontend/src/lib/types/message.ts` — add `SemanticSearchResult` + `SemanticSearchResponse` types.
- Modify: `services/frontend/src/lib/api/client.ts` (or `server.ts`) — add `semanticSearch` fetcher/routers export if using oRPC client; if the FE uses raw fetch through the proxy, add a `POST /api/messages/semantic-search` or an oRPC client call consistent with existing `messages.*` calls (follow the EXISTING pattern in `src/lib/api/` — inspect how `messages.list` is called and replicate).
**Step 1: Add FE types**
```ts
export interface SemanticSearchResult {
message_id: string;
content: string;
username: string;
channel_id: string;
guild_id: string;
created_at: number;
score: number;
}
export interface SemanticSearchResponse { results: SemanticSearchResult[]; nextCursor: string | null; }
```
**Step 2: Add hook + wire view**
Follow the existing `use-moderation.ts` SWR pattern. Add `useMessagesSemanticSearch(query, guildId?)` returning `{ data, isLoading, error }`. In `messages/view.tsx`, add a search `Input` (from `@/components/primitives`) at the top, debounce ~300ms, and render results below the live list when a query is present. Reuse the message-row rendering already in that view (do not invent a new component).
**Step 3: Type-check + lint (frontend)**
Run: `cd services/frontend && pnpm typecheck && pnpm lint`
Expected: PASS.
**Step 4: Commit**
```bash
git add services/frontend/src/app/\(dashboard\)/messages/ \
services/frontend/src/lib/types/message.ts \
services/frontend/src/lib/api/ \
services/frontend/src/hooks/
git commit -m "feat(web): public semantic message search UI"
```
---
## TASK 8 — Build all services + deploy + verify
1. `cd services/discord-gateway && pnpm typecheck && pnpm build && pnpm lint`
2. `cd services/backend && pnpm typecheck && pnpm build && pnpm lint`
3. `cd services/frontend && pnpm typecheck && pnpm build && pnpm lint`
4. Commit any formatting fixes (biome `--unsafe` if import order), author `asepharyana`, NO Co-Authored-By.
5. `git push origin main` → watch `gh run watch` on "Build & Deploy (Nix)".
6. After deploy: verify `systemctl show gmw-discord-gateway.service --property=ActiveEnterTimestamp,SubState` reflects new timestamp; same for backend + frontend.
7. Smoke: `curl -s http://127.0.0.1:4001/trpc/messages.semanticSearch?input=<urlencoded json>` OR via the public web `imphnen.asepharyana.my.id` messages page → type a query → expect results (after some messages have been embedded; embeddings only run on NEW captures post-deploy, so seed a few test messages or backfill).
8. Moderation explainability: trigger/observe a flagged message → confirm `moderation_actions.flags` is populated (SQL `SELECT flags, severity FROM moderation_actions ORDER BY created_at DESC LIMIT 5;`) and the public moderation view shows badges.
---
## RISKS / TRADEOFFS / OPEN QUESTIONS
- **Embedding cost:** #3 embeds EVERY captured message → more embedding API calls. Mitigated: only text ≥3 chars, fire-and-forget, skip if model unconfigured. If cost is a concern, batch embed (reuse `embedTexts`) per capture burst — but start simple (per-message) and observe.
- **Backfill:** post-deploy, the archive is empty until new messages arrive. Optional follow-up: a one-off backfill script over existing `messages` (out of scope for this plan unless user asks).
- **Schema duplication:** the `pgModerationActionsTable` double-definition must stay in sync (Task 1 patches both). If `messages.ts` copy is provably dead, a follow-up can delete it — but NOT in this plan (avoid scope creep / risk).
- **`AutoDeleteResult` verdict availability (Task 2):** requires confirming the `AnalysisResult` is reachable at the `createModerationAction` call sites. If not, we add an optional field to `AutoDeleteResult` at the orchestrator call site. This is the highest-risk integration point — verify before assuming.
- **Public exposure:** semantic search returns message text + usernames. This is INTENDED (web is public for users). No auth added. If a guild wants private, that is a future config (out of scope).
- **Qdrant payload type:** reusing `QdrantVerdictPayload` for archive is slightly awkward (carries `flags`/`expires_at` semantics). Cleaner: introduce `QdrantArchivePayload` with `message_id`, `channel_id`, `guild_id`, `username`, `content`, `created_at`, `expires_at`. **Prefer the dedicated payload type in Task 4/5** to avoid confusion.
## VERIFICATION CHECKLIST
- [ ] `moderation_actions` has 7 new columns (DB + both schema defs).
- [ ] A real auto-delete populates `flags`/`severity`/`evidence` (verified via SQL).
- [ ] Public moderation view renders badges + evidence (manual browser check on imphnen.asepharyana.my.id/moderation).
- [ ] New message capture upserts a point into `gmw_message_archive` (verify via Qdrant `/collections/gmw_message_archive/points/count`).
- [ ] `messages.semanticSearch` returns relevant results for a known phrase.
- [ ] All three services: typecheck + build + lint green; CI "Build & Deploy (Nix)" green; systemd timestamps updated.
+4 -50
View File
@@ -101,31 +101,8 @@
buildPhase = pnpmInstall + ''
echo "=== Compiling TypeScript ==="
npx tsc 2>&1
echo "=== Fixing @/ path aliases to relative paths ==="
node -e "
const fs = require('fs');
const path = require('path');
let count = 0;
function walk(dir) {
if (!fs.existsSync(dir)) return;
for (const e of fs.readdirSync(dir, {withFileTypes: true})) {
const p = path.join(dir, e.name);
if (e.isDirectory()) walk(p);
else if (e.name.endsWith('.js')) {
const c = fs.readFileSync(p, 'utf8');
const pat = /from\s+['\"]@\/([^'\"]+)['\"]/g;
const n = c.replace(pat, (m, p1) => {
const target = path.join('dist', p1) + '.js';
const rel = path.relative(path.dirname(p), target);
return 'from \"' + (rel.startsWith('.') ? rel : './' + rel) + '\"';
});
if (n !== c) { fs.writeFileSync(p, n); count++; }
}
}
}
walk('dist');
console.log('Fixed ' + count + ' files');
"
echo "=== Fixing @/ path aliases + extensionless relative imports for node ESM ==="
node scripts/fix-imports.mjs
echo "=== Build complete ==="
'' + pruneProd;
@@ -202,31 +179,8 @@ WRAPPER
pnpm rebuild @discordjs/opus 2>&1 || true
echo "=== Compiling TypeScript ===="
npx tsc 2>&1
echo "=== Fixing @/ path aliases to relative paths ==="
node -e "
const fs = require('fs');
const path = require('path');
let count = 0;
function walk(dir) {
if (!fs.existsSync(dir)) return;
for (const e of fs.readdirSync(dir, {withFileTypes: true})) {
const p = path.join(dir, e.name);
if (e.isDirectory()) walk(p);
else if (e.name.endsWith('.js')) {
const c = fs.readFileSync(p, 'utf8');
const pat = /from\s+['\"]@\/([^'\"]+)['\"]/g;
const n = c.replace(pat, (m, p1) => {
const target = path.join('dist', p1) + '.js';
const rel = path.relative(path.dirname(p), target);
return 'from \"' + (rel.startsWith('.') ? rel : './' + rel) + '\"';
});
if (n !== c) { fs.writeFileSync(p, n); count++; }
}
}
}
walk('dist');
console.log('Fixed ' + count + ' files');
"
echo "=== Fixing @/ path aliases + extensionless relative imports for node ESM ==="
node scripts/fix-imports.mjs
echo "=== Build complete ==="
'' + pruneProd;
+19
View File
@@ -61,6 +61,25 @@ http {
proxy_send_timeout 86400s;
}
# ── Backend oRPC (structured data RPCs over WebSocket + HTTP POST)
# Browser reaches this via partysocket (wss://…/trpc); SSR/RSC uses
# the fetch RPCLink (POST /trpc). Same path, same backend handler:
# oRPC's RPCHandler (HTTP) + ORPCWebSocketServer (WS) on :4001. ──
location ^~ /trpc {
proxy_pass http://gmw_backend$uri$is_args$args;
proxy_http_version 1.1;
# Upgrade headers required for the WebSocket transport; harmless for POST.
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_buffering off;
proxy_read_timeout 86400s;
proxy_send_timeout 86400s;
}
# ── Next.js build assets — immutable, edge/shareable ───────────
location ^~ /_next/static/ {
proxy_pass http://gmw_next$uri$is_args$args;
+2 -1
View File
@@ -15,6 +15,7 @@
},
"dependencies": {
"@discordjs/voice": "^0.19.2",
"@orpc/server": "1.15.0",
"axios": "^1.16.1",
"dotenv": "^17.4.2",
"drizzle-orm": "^0.45.2",
@@ -31,10 +32,10 @@
"@biomejs/biome": "latest",
"@types/express": "^5.0.6",
"@types/node": "^25.9.0",
"@types/pg": "^8.20.0",
"@types/ws": "^8.18.1",
"tsx": "^4.22.2",
"typescript": "^5.9.3",
"@types/pg": "^8.20.0",
"vitest": "latest"
}
}
+2735
View File
File diff suppressed because it is too large. Load diff
+49
View File
@@ -0,0 +1,49 @@
// Rewrite import specifiers in the compiled dist/ so the output runs under
// plain `node dist/index.js` (native ESM, no bundler / no tsx).
//
// Background: tsconfig uses moduleResolution:"bundler", so `tsc` emits BARE
// relative specifiers WITHOUT extensions (e.g. `import "./router"`) and leaves
// the `@/*` path-alias imports untouched. Node's native ESM resolver rejects
// extensionless relative specifiers and knows nothing about the `@/` alias, so
// the emitted dist/ crashes at startup (`ERR_MODULE_NOT_FOUND`). This script
// fixes both:
// 1. `@/foo` -> relative path to dist/foo.js
// 2. `./foo` / `../foo` -> `./foo.js` / `../foo.js` (append .js)
// Already-extensioned relative imports (.js/.json/.node/.mjs/.cjs) and bare
// package specifiers are left untouched (idempotent).
import { readFileSync, writeFileSync, existsSync, readdirSync } from "node:fs";
import { join, relative, dirname } from "node:path";
let count = 0;
function walk(dir) {
if (!existsSync(dir)) return;
for (const e of readdirSync(dir, { withFileTypes: true })) {
const p = join(dir, e.name);
if (e.isDirectory()) walk(p);
else if (e.name.endsWith(".js")) {
const c = readFileSync(p, "utf8");
const pat = /from\s+['"]([^'"]+)['"]/g;
const n = c.replace(pat, (m, spec) => {
if (spec.startsWith("@/")) {
const target = join("dist", spec.slice(2)) + ".js";
let rel = relative(dirname(p), target);
if (!rel.startsWith(".")) rel = "./" + rel;
return `from "${rel}"`;
}
if (
(spec.startsWith("./") || spec.startsWith("../")) &&
!/\.(js|json|node|mjs|cjs)$/.test(spec)
) {
return `from "${spec}.js"`;
}
return m;
});
if (n !== c) {
writeFileSync(p, n);
count++;
}
}
}
}
walk("dist");
console.log(`Fixed ${count} import specifiers in dist/`);
+37 -24
View File
@@ -1,3 +1,5 @@
import { onError } from "@orpc/server";
import { RPCHandler } from "@orpc/server/node";
import express, {
type Express,
type NextFunction,
@@ -6,20 +8,17 @@ import express, {
} from "express";
import helmet from "helmet";
import { createChildLogger } from "@/shared/logger/index";
import { createAnalysisRouter } from "../modules/analysis/index.js";
import { createChatbotRouter } from "../modules/chatbot/index.js";
import { createConfigRouter } from "../modules/config/index.js";
import { createDashboardRouter } from "../modules/dashboard/index.js";
import { createHealthRouter } from "../modules/health/index.js";
import { createMediaRouter } from "../modules/media/index.js";
import { createMessagesRouter } from "../modules/messages/index.js";
import { createModerationRouter } from "../modules/moderation/index.js";
import { createRecordingsRouter } from "../modules/recordings/index.js";
import { createUiStateRouter } from "../modules/ui-state/index.js";
import { createVoiceRouter } from "../modules/voice/index.js";
import { appRouter } from "../orpc/router";
import { errorHandler } from "../shared/middlewares/index.js";
// Auth removed — dashboard is public
// Auth removed — dashboard is public.
// All data APIs (dashboard, messages, moderation, media, voice, recordings,
// analysis, chatbot, config, ui-state) now flow over oRPC, served on TWO
// transports sharing the /trpc path:
// - WebSocket (browser live RPCs) — see orpc/ws.ts
// - HTTP POST (server-side / RSC fetch) — handled below
// Only infra endpoints (health, prometheus metrics) remain plain HTTP.
const logger = createChildLogger("http.app");
@@ -33,7 +32,7 @@ export function createHttpApp(): Express {
}),
);
// Body parsing
// Body parsing (still needed for any JSON POST; oRPC is WS/HTTP-based)
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
@@ -59,24 +58,38 @@ export function createHttpApp(): Express {
next();
});
// All routes are public
// Infra-only HTTP endpoints
app.use("/api", createHealthRouter());
app.use("/api", createConfigRouter());
app.use("/api", createDashboardRouter());
app.use("/api", createMessagesRouter());
app.use("/api", createAnalysisRouter());
app.use("/api", createChatbotRouter());
app.use("/api", createRecordingsRouter());
app.use("/api", createUiStateRouter());
app.use("/api", createMediaRouter());
app.use("/api", createVoiceRouter());
app.use("/api", createModerationRouter());
// oRPC over HTTP (server-side / RSC fetch). The same appRouter the browser
// reaches over the /trpc WebSocket. oRPC's node RPCHandler writes the full
// response itself; if no procedure matched we fall through to the 404 below.
const orpcHandler = new RPCHandler(appRouter, {
interceptors: [onError((error) => logger.error({ error }, "oRPC error"))],
});
app.use((req: Request, res: Response, next: NextFunction) => {
if (!req.path.startsWith("/trpc")) {
next();
return;
}
orpcHandler
.handle(req, res, { prefix: "/trpc", context: {} })
.then(({ matched }) => {
if (!matched) next();
})
.catch((err: unknown) => {
logger.error({ err }, "oRPC HTTP handler failed");
if (!res.headersSent) res.status(500).json({ error: "INTERNAL" });
});
});
// 404 handler
app.use((_req: Request, res: Response) => {
res.status(404).json({
error: "NOT_FOUND",
message: "Endpoint not found",
message:
"Endpoint not found — data APIs are served over /trpc (WebSocket/HTTP)",
});
});
+4 -2
View File
@@ -1,5 +1,6 @@
import { createServer, type Server } from "node:http";
import { createChildLogger } from "@/shared/logger/index";
import { createORPCWebSocketServer } from "../orpc/ws.js";
import { config } from "../shared/config/index.js";
import { initializeDatabase } from "../shared/database/index.js";
import { startRedisBridge } from "../ws/redis-bridge.js";
@@ -16,8 +17,9 @@ export async function startHttpServer(): Promise<Server> {
const server = createServer(app);
// Attach WebSocket server to the same HTTP server
createWebSocketServer(server);
// Attach WebSocket servers to the same HTTP server
createWebSocketServer(server); // /ws — voice PCM + gateway events
createORPCWebSocketServer(server); // /trpc — structured data RPCs
// Start Redis pub/sub bridge to forward discord-gateway events to WS clients
await startRedisBridge();
@@ -1,27 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { analysisService } from "./analysis.service.js";
const logger = createChildLogger("analysis.routes");
export function createAnalysisRouter(): Router {
const router = express.Router();
// GET /api/analysis/search
router.get(
"/analysis/search",
asyncHandler(async (req: Request, res: Response) => {
const q = (req.query.q as string) || "";
const channelId = (req.query.channelId as string) || undefined;
const limit = Number(req.query.limit) || 20;
logger.debug({ q, channelId, limit }, "Analysis search requested");
const result = await analysisService.search({ q, channelId, limit });
res.json(result);
}),
);
return router;
}
@@ -1 +0,0 @@
export { createAnalysisRouter } from "./analysis.routes.js";
@@ -1,96 +0,0 @@
import type { Request, Response } from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { chatbotService } from "./chatbot.service.js";
const logger = createChildLogger("chatbot.controller");
interface AuthenticatedRequest extends Request {
userId?: string;
}
/**
* Resolve the actor id for a request. Frontend (no-login) sends a per-device
* UUID via X-User-Id so chat history stays isolated per visitor; a registered
* auth middleware userId takes precedence when present.
*/
function resolveUserId(req: Request): string {
const authId = (req as AuthenticatedRequest).userId;
if (authId) return authId;
const header = (req.headers["x-user-id"] as string | undefined)?.trim();
return header || "anonymous";
}
export const handleChatbotChat = asyncHandler(
async (req: Request, res: Response) => {
const { message, context } = req.body as {
message: string;
context?: Record<string, unknown>;
};
// Validate required fields
if (!message || typeof message !== "string") {
return res.status(400).json({
error: "INVALID_INPUT",
message: "Message is required and must be a string",
});
}
// Get user ID from X-User-Id header (no-login device uuid) or auth
const userId = resolveUserId(req);
logger.debug(
{ userId, messageLength: message.length, context },
"Received chatbot chat message",
);
// Process message & generate response
const response = await chatbotService.processMessage(
message,
context,
userId,
);
// Save conversation to database
await chatbotService.saveConversation({
userId,
userMessage: message,
botResponse: response,
context,
timestamp: new Date(),
});
logger.info({ userId }, "Chatbot chat processed successfully");
res.status(200).json({
response,
timestamp: new Date().toISOString(),
});
},
);
export const getChatbotHistory = asyncHandler(
async (req: Request, res: Response) => {
const userId = resolveUserId(req);
const limit = Math.min(parseInt(req.query.limit as string, 10) || 50, 100);
const history = await chatbotService.getChatHistory(userId, limit);
res.status(200).json({
history,
total: history.length,
});
},
);
export const clearChatbotHistory = asyncHandler(
async (req: Request, res: Response) => {
const userId = resolveUserId(req);
await chatbotService.clearChatHistory(userId);
res.status(200).json({
message: "Chat history cleared successfully",
});
},
);
@@ -1,6 +1,6 @@
import { and, desc, eq, type SQL, sql } from "drizzle-orm";
import { desc, eq } from "drizzle-orm";
import { getDatabase } from "../../shared/database/index.js";
import { pgChatbotMessagesTable, pgMessagesTable } from "../../shared/index.js";
import { pgChatbotMessagesTable } from "../../shared/index.js";
import { createChildLogger } from "../../shared/logger/index.js";
const logger = createChildLogger("chatbot.repository");
@@ -31,13 +31,6 @@ export interface ChatbotHistoryRow {
created_at: string;
}
export interface ServerInsights {
total_messages: number;
active_users: number;
flagged: number;
warned: number;
}
export class ChatbotRepository {
async saveConversation(input: SaveConversationInput): Promise<void> {
const db = getDatabase();
@@ -83,56 +76,6 @@ export class ChatbotRepository {
"Chat history cleared",
);
}
async getServerInsights(
guildId?: string,
channelId?: string,
): Promise<ServerInsights> {
try {
const db = getDatabase();
const conditions: SQL[] = [];
if (guildId) {
conditions.push(eq(pgMessagesTable.guild_id, guildId));
}
if (channelId) {
conditions.push(eq(pgMessagesTable.channel_id, channelId));
}
const where = conditions.length > 0 ? and(...conditions) : undefined;
const [result] = await db
.select({
total_messages: sql<number>`COUNT(*)::int`,
active_users: sql<number>`COUNT(DISTINCT ${pgMessagesTable.user_id})::int`,
flagged: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'flagged')::int`,
warned: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'warn')::int`,
})
.from(pgMessagesTable)
.where(where);
const insights = result ?? {
total_messages: 0,
active_users: 0,
flagged: 0,
warned: 0,
};
logger.debug({ guildId, channelId, insights }, "Server insights fetched");
return insights;
} catch (error) {
logger.warn(
{ error, guildId, channelId },
"Failed to load server insights",
);
return {
total_messages: 0,
active_users: 0,
flagged: 0,
warned: 0,
};
}
}
}
export const chatbotRepository = new ChatbotRepository();
@@ -1,18 +0,0 @@
import express, { type Router } from "express";
import { validateBody } from "../../shared/middlewares/index.js";
import {
clearChatbotHistory,
getChatbotHistory,
handleChatbotChat,
} from "./chatbot.controller.js";
import { chatRequestSchema } from "./chatbot.schema.js";
export function createChatbotRouter(): Router {
const router = express.Router();
router.post("/chat", validateBody(chatRequestSchema), handleChatbotChat);
router.get("/chat/history", getChatbotHistory);
router.delete("/chat/history", clearChatbotHistory);
return router;
}
@@ -6,7 +6,8 @@ import type {
SaveConversationInput,
} from "./chatbot.repository.js";
import { chatbotRepository } from "./chatbot.repository.js";
import { executeTool, tools } from "./chatbot.tools.js";
import { tools } from "./chatbot.toolDefs.js";
import { executeTool } from "./chatbot.tools.js";
const logger = createChildLogger("chatbot.service");
@@ -17,22 +18,26 @@ class ChatbotService {
userId: string,
): Promise<string> {
logger.info(
{ userId, messageLength: message.length },
{ userId, messageLength: message.length, context },
"processMessage called",
);
const recentContext = await this.getRecentConversationContext(userId);
const serverInsights = await chatbotRepository.getServerInsights(
context?.guildId,
context?.channelId,
);
// Scope the agent to the server/channel the user is chatting in. We no
// longer bake server stats into the prompt — the model must pull current
// data via tools (see buildSystemPrompt), so it always answers from live
// numbers instead of a stale snapshot.
const scope = {
guildId: context?.guildId,
channelId: context?.channelId,
};
// Build LLM messages
const systemPrompt = this.buildSystemPrompt(serverInsights);
const systemPrompt = this.buildSystemPrompt(scope);
const conversationHistory = this.buildHistoryMessages(recentContext);
const llmResponse = await this.callLLM(
systemPrompt,
conversationHistory,
message,
scope,
);
return llmResponse;
@@ -66,27 +71,29 @@ class ChatbotService {
]);
}
private buildSystemPrompt(insights: {
total_messages: number;
active_users: number;
flagged: number;
warned: number;
private buildSystemPrompt(scope: {
guildId?: string;
channelId?: string;
}): string {
return `Kamu lagi ngobrol sama chatbot Discord Watcher — temen ngobrol yang tau keadaan server.
const scopeLine = scope.guildId
? `- Scope: kamu menjawab soal server/guild id="${scope.guildId}"${scope.channelId ? `, channel id="${scope.channelId}"` : ""}.`
: "- Scope: tidak ada guild spesifik — jawab umum soal server ini.";
return `Kamu adalah chatbot Discord Watcher — temen ngobrol yang tau keadaan server, dan kamu PUNYA AKSES ke data server lewat tools.
Data server saat ini:
- Pesan: ${insights.total_messages}
- User aktif: ${insights.active_users}
- Flagged: ${insights.flagged}
- Warning: ${insights.warned}
${scopeLine}
ATURAN PENTING — JANGAN PAKAI KONTEKS STATIS:
- Kamu TIDAK punya hafalan soal angka server (jumlah pesan, user aktif, flagged, dll). JANGAN tebak atau karang angka.
- Untuk SEMUA pertanyaan soal data server (jumlah pesan, user aktif, channel ramai, aktivitas terbaru, pesan di-flag), WAJIB panggil tool yang sesuai (get_server_stats, get_top_channels, get_recent_activity, get_top_flagged). Jawab HANYA dari hasil tool.
- Tool otomatis di-scope ke guild/channel di atas — kalau argumen guildId/channelId kosong, biarkan kosong (sudah otomatis ter-isi). Jangan isi ID yang kamu tebak.
- Kalau tool balas error atau kosong, bilang aja data lagi ga ketemu, jangan karang.
Gaya ngobrol:
- Santai, hangat, kayak ngobrol sama temen
- Pake Bahasa Indonesia sehari-hari, ga perlu kaku
- Sesekali pake emoji wajar aja, ga berlebihan
- Kalo ditanya sesuatu yang kamu tau dari data server, jawab pake data itu
- Kalo ga tau atau ga nyambung, bilang aja terus tanya balik biar ngobrolnya jalan
- Jangan sebut "rule", "instruksi", "prompt" atau apapun soal cara kamu berpikir
- Kalo ditanya di luar data server dan kamu ga tau, bilang aja terus tanya balik biar ngobrolnya jalan
- Jangan sebut "rule", "instruksi", "prompt", "tool", atau apapun soal cara kamu berpikir
- Biasa aja, ga usaha lucu-lucu amat — natural`;
}
@@ -106,6 +113,7 @@ Gaya ngobrol:
systemPrompt: string,
history: Array<{ role: "user" | "assistant"; content: string }>,
userMessage: string,
scope: { guildId?: string; channelId?: string },
): Promise<string> {
const apiKey = config.AI_LLM_API_KEY;
const baseUrl = config.AI_LLM_BASE_URL;
@@ -150,7 +158,13 @@ Gaya ngobrol:
tool_choice: "auto",
max_tokens: 600,
temperature: 0.4,
stream: true,
// Non-streaming: request a single complete response. 9router may
// still emit SSE even with stream:false, so the parser below
// handle both raw-JSON and SSE bodies.
stream: false,
// Disable extended thinking / reasoning tokens so the bot answers
// directly (ignored by non-reasoning models).
reasoning_effort: "none",
},
{
headers: {
@@ -158,14 +172,16 @@ Gaya ngobrol:
"Content-Type": "application/json",
},
timeout: 45_000,
// 9router returns SSE even without stream:true; force stream:true
// in the body and read the raw SSE text.
responseType: "text",
},
);
// Parse SSE `data:` lines → content + tool_calls.
const { content, toolCalls } = this.parseSse(response.data as string);
// Parse the body into content + tool_calls. 9router may return either
// a single JSON object (stream:false honored) or SSE text (stream
// implied) — parseResponse handles both.
const { content, toolCalls } = this.parseResponse(
response.data as string,
);
logger.debug(
{
@@ -190,9 +206,19 @@ Gaya ngobrol:
},
],
});
// Auto-scope: if the model omitted guildId/channelId, fill them
// from the request scope so tools query the right server without
// the model having to guess IDs.
const scopedArgs = { ...tc.args };
if (scope.guildId && scopedArgs.guildId == null) {
scopedArgs.guildId = scope.guildId;
}
if (scope.channelId && scopedArgs.channelId == null) {
scopedArgs.channelId = scope.channelId;
}
let result = "";
try {
result = await executeTool(tc.name, tc.args);
result = await executeTool(tc.name, scopedArgs);
} catch (e) {
result = `Tool error: ${(e as Error).message}`;
}
@@ -224,6 +250,60 @@ Gaya ngobrol:
}
}
/**
* Parse an LLM HTTP body into content + tool_calls. Handles both shapes
* 9router can return: a single JSON object (stream:false honored) or SSE
* text (stream implied). For SSE we delegate to parseSse.
*/
private parseResponse(body: string): {
content: string;
toolCalls: Array<{
id: string;
name: string;
arguments: string;
args: Record<string, unknown>;
}>;
} {
const trimmed = body.trim();
// Non-streaming response: a single JSON object.
if (trimmed.startsWith("{")) {
try {
const json = JSON.parse(trimmed) as {
choices?: Array<{
message?: {
content?: string | null;
tool_calls?: Array<{
id?: string;
type?: string;
function?: { name?: string; arguments?: string };
}>;
};
delta?: unknown;
}>;
};
const msg = json.choices?.[0]?.message;
// If the router returned SSE-style shape under `choices[].delta`
// (rare), fall through to the SSE parser.
if (msg) {
const content = msg.content ?? "";
const toolCalls = (msg.tool_calls ?? []).map((tc, i) => {
const id = tc.id || `tool_${i}_${Date.now()}`;
return {
id,
name: tc.function?.name ?? "",
arguments: tc.function?.arguments ?? "",
args: this.safeJsonParse(tc.function?.arguments ?? ""),
};
});
return { content: content.trim(), toolCalls };
}
} catch {
// Not valid JSON after all — treat as SSE below.
}
}
return this.parseSse(body);
}
/**
* Parse an SSE stream body into accumulated content + any tool_calls.
* 9router (and most OpenAI-compatible routers) emit `data: {json}` lines
@@ -0,0 +1,270 @@
/**
* Static tool *definitions* for the chatbot LLM (OpenAI function-calling
* format). Kept separate from the executor (chatbot.tools.ts) so the schema
* the model depends on can be imported without pulling in the database /
* config layer.
*
* The chatbot is a server-watcher agent: it can answer about ANY server
* situation — activity, moderation queue, specific users, channels, voice
* recordings, AI correction history, and trends over time — by calling these
* tools, which the executor implements against real tables.
*/
export interface ToolDef {
type: "function";
function: {
name: string;
description: string;
parameters: {
type: "object";
properties: Record<string, unknown>;
required?: string[];
};
};
}
export const tools: ToolDef[] = [
{
type: "function",
function: {
name: "get_server_stats",
description:
"Ambil statistik ringkas server/guild: total pesan, user aktif, jumlah pesan flagged, warn, dan clean. Panggil untuk jawab pertanyaan umum soal kondisi server. guildId/channelId otomatis ter-isi dari scope; kosongkan untuk semua data.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
},
},
},
},
{
type: "function",
function: {
name: "get_top_channels",
description:
"Ambil daftar channel paling aktif (jumlah pesan terbanyak). Panggil untuk 'channel mana paling ramai' atau aktivitas per-channel.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
limit: {
type: "number",
description: "Jumlah channel teratas (default 5, max 10).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_recent_activity",
description:
"Ambil pesan terbaru di server: siapa, di channel mana, jam berapa, isinya. Panggil untuk 'lagi ngapain' / aktivitas terbaru.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
limit: {
type: "number",
description: "Jumlah pesan terakhir (default 5, max 20).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_top_flagged",
description:
"Ambil pesan dengan ai_status flagged (beserta alasan, severity, analysis). Panggil untuk bahas pesan bermasalah / kerjaan moderator.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
limit: { type: "number", description: "Jumlah pesan (default 5)." },
},
},
},
},
{
type: "function",
function: {
name: "search_messages",
description:
"Cari pesan berdasarkan kata kunci di isi pesan (case-insensitive, LIKE). Untuk 'ada yang bahas X gak?' / temukan topik tertentu. Hindari kata terlalu umum.",
parameters: {
type: "object",
properties: {
query: {
type: "string",
description: "Kata kunci pencarian (wajib).",
},
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
limit: { type: "number", description: "Jumlah hasil (default 5)." },
},
required: ["query"],
},
},
},
{
type: "function",
function: {
name: "get_user_messages",
description:
"Ambil pesan terbaru dari satu user tertentu (user_id), opsional di-scope ke guild/channel. Untuk 'chat si A gimana akhir-akhir ini?' — butuh user_id.",
parameters: {
type: "object",
properties: {
userId: { type: "string", description: "ID user (wajib)." },
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
limit: { type: "number", description: "Jumlah pesan (default 10)." },
},
required: ["userId"],
},
},
},
{
type: "function",
function: {
name: "get_user_profile",
description:
"Ambil ringkasan profil AI dari seorang user (pola perilaku, gaya bicara) dari tabel user_profiles. Untuk 'siapa si A?' / konteks perilaku. Butuh user_id.",
parameters: {
type: "object",
properties: {
userId: { type: "string", description: "ID user (wajib)." },
guildId: { type: "string", description: "ID server (opsional)." },
},
required: ["userId"],
},
},
},
{
type: "function",
function: {
name: "get_user_reputation",
description:
"Ambil skor trust, jumlah infraction, dan streak pesan bersih seorang user dari user_reputations. Untuk 'berapa trust score si A?' / riwayat pelanggaran. Butuh user_id.",
parameters: {
type: "object",
properties: {
userId: { type: "string", description: "ID user (wajib)." },
guildId: { type: "string", description: "ID server (opsional)." },
},
required: ["userId"],
},
},
},
{
type: "function",
function: {
name: "get_channel_culture",
description:
"Ambil ringkasan norma/slang channel dari tabel channel_cultures (AI-generated). Untuk 'norma channel ini gimana?' / konteks sebelum nge-flag. Butuh channel_id.",
parameters: {
type: "object",
properties: {
channelId: { type: "string", description: "ID channel (wajib)." },
},
required: ["channelId"],
},
},
},
{
type: "function",
function: {
name: "get_message_detail",
description:
"Ambil 1 pesan lengkap beserta hasil analisis AI-nya (status, flags, score, severity, kategori, analysis, recommended action). Untuk jelasin keputusan moderasi pada pesan tertentu. Butuh message_id.",
parameters: {
type: "object",
properties: {
messageId: { type: "string", description: "ID pesan (wajib)." },
},
required: ["messageId"],
},
},
},
{
type: "function",
function: {
name: "get_message_reviews",
description:
"Ambil antrean review moderasi manual (message_reviews) berdasarkan status: pending/approved/rejected/escalated. Untuk 'ada review moderasi pending?' / cek kerjaan human moderator. guildId otomatis ter-isi.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
status: {
type: "string",
description:
"Status review: pending / approved / rejected / escalated (opsional, default semua).",
},
limit: { type: "number", description: "Jumlah (default 10)." },
},
},
},
},
{
type: "function",
function: {
name: "get_voice_recordings",
description:
"Ambil rekaman suara terbaru (voice_recordings): user, channel, transkripsi, status upload. Untuk 'ada rekaman suara terbaru?' / cek transkripsi. Bisa di-scope ke user_id atau channel_id.",
parameters: {
type: "object",
properties: {
userId: { type: "string", description: "Filter user (opsional)." },
channelId: {
type: "string",
description: "Filter channel (opsional).",
},
guildId: { type: "string", description: "ID server (opsional)." },
limit: { type: "number", description: "Jumlah (default 10)." },
},
},
},
},
{
type: "function",
function: {
name: "get_moderation_timeline",
description:
"Ambil tren harian: per hari, jumlah total pesan vs flagged vs warn vs clean. Untuk 'minggu ini pelanggaran naik?' / lihat tren moderasi. guildId otomatis ter-isi.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
channelId: { type: "string", description: "ID channel (opsional)." },
days: {
type: "number",
description: "Jumlah hari ke belakang (default 14, max 60).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_corrections",
description:
"Ambil riwayat koreksi false-positive AI (corrected_moderations): pesan yang awalnya di-flag tapi dikoreksi manusia, beserta alasannya. Untuk 'AI pernah salah nge-flag apa aja?' / audit akurasi moderasi.",
parameters: {
type: "object",
properties: {
guildId: { type: "string", description: "ID server (opsional)." },
limit: { type: "number", description: "Jumlah (default 10)." },
},
},
},
},
];
@@ -1,116 +1,25 @@
import { sql } from "drizzle-orm";
import { and, desc, eq, like, sql } from "drizzle-orm";
import { getDatabase } from "../../shared/database/index.js";
import {
pgChannelCulturesTable,
pgCorrectedModerationsTable,
pgMessageReviewsTable,
pgMessagesTable,
pgUserProfilesTable,
pgUserReputationsTable,
pgVoiceRecordingsTable,
} from "../../shared/index.js";
/**
* Tools the chatbot LLM can call. Definitions describe the schema to the
* model; the executor implements each one against the real database.
* This turns the chatbot from "blind stats guesser" into an agent that
* pulls real, current server data on demand.
* Executor for the chatbot's server-watcher tools. The tool *definitions*
* live in chatbot.toolDefs.ts (no DB import); this file implements each one
* against the real database.
*
* All queries use parameterized drizzle operators (eq/like/and) — never string
* interpolation into raw SQL — so model-supplied arguments cannot inject SQL.
*/
export type ToolResult = string;
/** JSON schema for a tool definition (OpenAI function-calling format). */
export interface ToolDef {
type: "function";
function: {
name: string;
description: string;
parameters: {
type: "object";
properties: Record<string, unknown>;
required?: string[];
};
};
}
export const tools: ToolDef[] = [
{
type: "function",
function: {
name: "get_server_stats",
description:
"Ambil statistik ringkas server/guild saat ini: total pesan, user aktif, jumlah pesan flagged, dan jumlah warning. Panggil ini untuk menjawab pertanyaan umum tentang kondisi server. Opsional fill guild_id untuk scope ke guild tertentu, channel_id untuk scope ke channel.",
parameters: {
type: "object",
properties: {
guildId: {
type: "string",
description: "ID guild/server (opsional). Kosongkan = semua data.",
},
channelId: {
type: "string",
description: "ID channel (opsional).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_top_channels",
description:
"Ambil daftar channel paling aktif (jumlah pesan terbanyak) di server. Panggil buat jawab 'channel mana paling ramai' atau aktivitas per-channel.",
parameters: {
type: "object",
properties: {
guildId: {
type: "string",
description: "ID server (opsional).",
},
limit: {
type: "number",
description: "Jumlah channel teratas (default 5, max 10).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_recent_activity",
description:
"Ambil aktivitas/pesan terbaru di server: siapa yang baru ngomong, di channel mana, jam berapa. Panggil buat jawaban soal 'lagi ngapain' / aktivitas terbaru di server.",
parameters: {
type: "object",
properties: {
guildId: {
type: "string",
description: "ID server (opsional).",
},
limit: {
type: "number",
description: "Jumlah pesan terakhir (default 5).",
},
},
},
},
},
{
type: "function",
function: {
name: "get_top_flagged",
description:
"Ambil pesan yang paling sering di-flag atau kena warning. Panggil buat jawab soal pesan bermasalah / moderator.",
parameters: {
type: "object",
properties: {
guildId: {
type: "string",
description: "ID server (opsional).",
},
limit: {
type: "number",
description: "Jumlah pesan (default 5).",
},
},
},
},
},
];
/** Executes a tool call against the real DB and returns a readable result. */
export async function executeTool(
name: string,
@@ -122,9 +31,11 @@ export async function executeTool(
typeof args.channelId === "string" && args.channelId
? args.channelId
: undefined;
const userId =
typeof args.userId === "string" && args.userId ? args.userId : undefined;
const limitRaw =
typeof args.limit === "number" ? args.limit : Number(args.limit) || 5;
const limit = Math.min(Math.max(1, Math.round(limitRaw)), 10);
const limit = Math.min(Math.max(1, Math.round(limitRaw)), 20);
try {
switch (name) {
@@ -133,9 +44,48 @@ export async function executeTool(
case "get_top_channels":
return await topChannels(guildId, limit);
case "get_recent_activity":
return await recentActivity(guildId, limit);
return await recentActivity(guildId, channelId, limit);
case "get_top_flagged":
return await topFlagged(guildId, limit);
return await topFlagged(guildId, channelId, limit);
case "search_messages":
return await searchMessages(
String(args.query ?? ""),
guildId,
channelId,
limit,
);
case "get_user_messages":
return await userMessages(userId, guildId, channelId, limit);
case "get_user_profile":
return await userProfile(userId, guildId);
case "get_user_reputation":
return await userReputation(userId, guildId);
case "get_channel_culture":
return await channelCulture(
typeof args.channelId === "string" ? args.channelId : undefined,
);
case "get_message_detail":
return await messageDetail(
typeof args.messageId === "string" ? args.messageId : undefined,
);
case "get_message_reviews":
return await messageReviews(
guildId,
typeof args.status === "string" ? args.status : undefined,
limit,
);
case "get_voice_recordings":
return await voiceRecordings(userId, channelId, guildId, limit);
case "get_moderation_timeline":
return await moderationTimeline(
guildId,
channelId,
typeof args.days === "number"
? Math.min(Math.max(1, args.days), 60)
: 14,
);
case "get_corrections":
return await corrections(guildId, limit);
default:
return `Unknown tool: ${name}`;
}
@@ -145,6 +95,23 @@ export async function executeTool(
}
}
// ── Query helpers ──────────────────────────────────────────
function scopeMessages(
guildId?: string,
channelId?: string,
): ReturnType<typeof and> | undefined {
const conds = [];
if (guildId) conds.push(eq(pgMessagesTable.guild_id, guildId));
if (channelId) conds.push(eq(pgMessagesTable.channel_id, channelId));
return conds.length ? and(...conds) : undefined;
}
/** Escape LIKE wildcards so user input can't break the pattern. */
function likePattern(q: string): string {
return q.replace(/[\\%_]/g, (c) => `\\${c}`);
}
// ── Tool executors ──────────────────────────────────────────
async function serverStats(
@@ -152,81 +119,330 @@ async function serverStats(
channelId?: string,
): Promise<string> {
const db = getDatabase();
const conditions: string[] = [];
if (guildId) conditions.push(`guild_id = '${guildId}'`);
if (channelId) conditions.push(`channel_id = '${channelId}'`);
const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
const [result] = await db
.select({
total_messages: sql<number>`COUNT(*)::int`,
active_users: sql<number>`COUNT(DISTINCT ${pgMessagesTable.user_id})::int`,
flagged: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'flagged')::int`,
warned: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'warn')::int`,
clean: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'clean')::int`,
})
.from(pgMessagesTable)
.where(scopeMessages(guildId, channelId));
const result = await db.execute(
sql.raw(
`SELECT COUNT(*)::int AS total_messages,
COUNT(DISTINCT user_id)::int AS active_users,
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned
FROM messages ${cond}`,
),
);
const rows =
(result as unknown as { rows: Record<string, unknown>[] }).rows ?? [];
const r = rows[0] ?? {};
return JSON.stringify({
total_messages: r.total_messages ?? 0,
active_users: r.active_users ?? 0,
flagged: r.flagged ?? 0,
warned: r.warned ?? 0,
});
const r = result ?? {
total_messages: 0,
active_users: 0,
flagged: 0,
warned: 0,
clean: 0,
};
return JSON.stringify(r);
}
async function topChannels(guildId?: string, limit = 5): Promise<string> {
const db = getDatabase();
const conditions: string[] = [];
if (guildId) conditions.push(`guild_id = '${guildId}'`);
const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
const result = await db.execute(
sql.raw(
`SELECT channel_id,
COUNT(*)::int AS count
FROM messages ${cond}
GROUP BY channel_id
ORDER BY count DESC
LIMIT ${limit}`,
),
);
const rows = (result as unknown as { rows: unknown[] }).rows ?? [];
return JSON.stringify(rows.slice(0, limit));
const rows = await db
.select({
channel_id: pgMessagesTable.channel_id,
count: sql<number>`COUNT(*)::int`,
})
.from(pgMessagesTable)
.where(scopeMessages(guildId))
.groupBy(pgMessagesTable.channel_id)
.orderBy(desc(sql`COUNT(*)`))
.limit(limit);
return JSON.stringify(rows);
}
async function recentActivity(guildId?: string, limit = 5): Promise<string> {
async function recentActivity(
guildId?: string,
channelId?: string,
limit = 5,
): Promise<string> {
const db = getDatabase();
const conditions: string[] = [];
if (guildId) conditions.push(`guild_id = '${guildId}'`);
const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
const result = await db.execute(
sql.raw(
`SELECT username, content, channel_id, created_at
FROM messages ${cond}
ORDER BY created_at DESC
LIMIT ${limit}`,
),
);
return JSON.stringify((result as unknown as { rows: unknown[] }).rows ?? []);
const rows = await db
.select({
id: pgMessagesTable.id,
username: pgMessagesTable.username,
user_id: pgMessagesTable.user_id,
channel_id: pgMessagesTable.channel_id,
content: pgMessagesTable.content,
created_at: pgMessagesTable.created_at,
ai_status: pgMessagesTable.ai_status,
})
.from(pgMessagesTable)
.where(scopeMessages(guildId, channelId))
.orderBy(desc(pgMessagesTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function topFlagged(guildId?: string, limit = 5): Promise<string> {
async function topFlagged(
guildId?: string,
channelId?: string,
limit = 5,
): Promise<string> {
const db = getDatabase();
const conditions = ["ai_status IN ('flagged', 'warn')"];
if (guildId) conditions.push(`guild_id = '${guildId}'`);
const cond = `WHERE ${conditions.join(" AND ")}`;
const result = await db.execute(
sql.raw(
`SELECT username, content, channel_id, ai_status, created_at
FROM messages ${cond}
ORDER BY created_at DESC
LIMIT ${limit}`,
),
);
return JSON.stringify((result as unknown as { rows: unknown[] }).rows ?? []);
const rows = await db
.select({
id: pgMessagesTable.id,
username: pgMessagesTable.username,
channel_id: pgMessagesTable.channel_id,
content: pgMessagesTable.content,
ai_status: pgMessagesTable.ai_status,
ai_severity: pgMessagesTable.ai_severity,
ai_moderation_flags: pgMessagesTable.ai_moderation_flags,
ai_analysis: pgMessagesTable.ai_analysis,
created_at: pgMessagesTable.created_at,
})
.from(pgMessagesTable)
.where(
and(
scopeMessages(guildId, channelId),
eq(pgMessagesTable.ai_status, "flagged"),
),
)
.orderBy(desc(pgMessagesTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function searchMessages(
query: string,
guildId?: string,
channelId?: string,
limit = 5,
): Promise<string> {
const db = getDatabase();
if (!query.trim()) return JSON.stringify({ error: "query kosong" });
const rows = await db
.select({
id: pgMessagesTable.id,
username: pgMessagesTable.username,
channel_id: pgMessagesTable.channel_id,
content: pgMessagesTable.content,
created_at: pgMessagesTable.created_at,
ai_status: pgMessagesTable.ai_status,
})
.from(pgMessagesTable)
.where(
and(
scopeMessages(guildId, channelId),
like(pgMessagesTable.content, `%${likePattern(query)}%`),
),
)
.orderBy(desc(pgMessagesTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function userMessages(
userId?: string,
guildId?: string,
channelId?: string,
limit = 10,
): Promise<string> {
const db = getDatabase();
if (!userId) return JSON.stringify({ error: "userId wajib" });
const conds = [eq(pgMessagesTable.user_id, userId)];
if (guildId) conds.push(eq(pgMessagesTable.guild_id, guildId));
if (channelId) conds.push(eq(pgMessagesTable.channel_id, channelId));
const rows = await db
.select({
id: pgMessagesTable.id,
channel_id: pgMessagesTable.channel_id,
content: pgMessagesTable.content,
created_at: pgMessagesTable.created_at,
ai_status: pgMessagesTable.ai_status,
})
.from(pgMessagesTable)
.where(and(...conds))
.orderBy(desc(pgMessagesTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function userProfile(userId?: string, guildId?: string): Promise<string> {
const db = getDatabase();
if (!userId) return JSON.stringify({ error: "userId wajib" });
const conds = [eq(pgUserProfilesTable.user_id, userId)];
if (guildId) conds.push(eq(pgUserProfilesTable.guild_id, guildId));
const rows = await db
.select({
user_id: pgUserProfilesTable.user_id,
guild_id: pgUserProfilesTable.guild_id,
profile_summary: pgUserProfilesTable.profile_summary,
last_analyzed_at: pgUserProfilesTable.last_analyzed_at,
})
.from(pgUserProfilesTable)
.where(and(...conds))
.limit(1);
return JSON.stringify(rows[0] ?? { error: "profil tidak ditemukan" });
}
async function userReputation(
userId?: string,
guildId?: string,
): Promise<string> {
const db = getDatabase();
if (!userId) return JSON.stringify({ error: "userId wajib" });
const conds = [eq(pgUserReputationsTable.user_id, userId)];
if (guildId) conds.push(eq(pgUserReputationsTable.guild_id, guildId));
const rows = await db
.select({
user_id: pgUserReputationsTable.user_id,
guild_id: pgUserReputationsTable.guild_id,
trust_score: pgUserReputationsTable.trust_score,
clean_message_streak: pgUserReputationsTable.clean_message_streak,
total_infractions: pgUserReputationsTable.total_infractions,
last_infraction_at: pgUserReputationsTable.last_infraction_at,
})
.from(pgUserReputationsTable)
.where(and(...conds))
.limit(1);
return JSON.stringify(rows[0] ?? { error: "reputasi tidak ditemukan" });
}
async function channelCulture(channelId?: string): Promise<string> {
const db = getDatabase();
if (!channelId) return JSON.stringify({ error: "channelId wajib" });
const rows = await db
.select({
channel_id: pgChannelCulturesTable.channel_id,
culture_summary: pgChannelCulturesTable.culture_summary,
last_analyzed_at: pgChannelCulturesTable.last_analyzed_at,
})
.from(pgChannelCulturesTable)
.where(eq(pgChannelCulturesTable.channel_id, channelId))
.limit(1);
return JSON.stringify(rows[0] ?? { error: "culture tidak ditemukan" });
}
async function messageDetail(messageId?: string): Promise<string> {
const db = getDatabase();
if (!messageId) return JSON.stringify({ error: "messageId wajib" });
const rows = await db
.select({
id: pgMessagesTable.id,
guild_id: pgMessagesTable.guild_id,
channel_id: pgMessagesTable.channel_id,
user_id: pgMessagesTable.user_id,
username: pgMessagesTable.username,
content: pgMessagesTable.content,
created_at: pgMessagesTable.created_at,
ai_status: pgMessagesTable.ai_status,
ai_moderation_flags: pgMessagesTable.ai_moderation_flags,
ai_moderation_score: pgMessagesTable.ai_moderation_score,
ai_severity: pgMessagesTable.ai_severity,
ai_categories: pgMessagesTable.ai_categories,
ai_analysis: pgMessagesTable.ai_analysis,
ai_recommended_action: pgMessagesTable.ai_recommended_action,
ai_confidence: pgMessagesTable.ai_confidence,
})
.from(pgMessagesTable)
.where(eq(pgMessagesTable.id, messageId))
.limit(1);
return JSON.stringify(rows[0] ?? { error: "pesan tidak ditemukan" });
}
async function messageReviews(
guildId?: string,
status?: string,
limit = 10,
): Promise<string> {
const db = getDatabase();
const conds = [];
if (guildId) conds.push(eq(pgMessageReviewsTable.guild_id, guildId));
if (status) conds.push(eq(pgMessageReviewsTable.status, status as never));
const rows = await db
.select({
id: pgMessageReviewsTable.id,
message_id: pgMessageReviewsTable.message_id,
reviewer_id: pgMessageReviewsTable.reviewer_id,
status: pgMessageReviewsTable.status,
notes: pgMessageReviewsTable.notes,
created_at: pgMessageReviewsTable.created_at,
reviewed_at: pgMessageReviewsTable.reviewed_at,
})
.from(pgMessageReviewsTable)
.where(conds.length ? and(...conds) : undefined)
.orderBy(desc(pgMessageReviewsTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function voiceRecordings(
userId?: string,
channelId?: string,
guildId?: string,
limit = 10,
): Promise<string> {
const db = getDatabase();
const conds = [];
if (userId) conds.push(eq(pgVoiceRecordingsTable.user_id, userId));
if (channelId) conds.push(eq(pgVoiceRecordingsTable.channel_id, channelId));
if (guildId) conds.push(eq(pgVoiceRecordingsTable.guild_id, guildId));
const rows = await db
.select({
id: pgVoiceRecordingsTable.id,
username: pgVoiceRecordingsTable.username,
channel_name: pgVoiceRecordingsTable.channel_name,
filename: pgVoiceRecordingsTable.filename,
size_bytes: pgVoiceRecordingsTable.size_bytes,
upload_status: pgVoiceRecordingsTable.upload_status,
transcription: pgVoiceRecordingsTable.transcription,
created_at: pgVoiceRecordingsTable.created_at,
})
.from(pgVoiceRecordingsTable)
.where(conds.length ? and(...conds) : undefined)
.orderBy(desc(pgVoiceRecordingsTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
async function moderationTimeline(
guildId?: string,
channelId?: string,
days = 14,
): Promise<string> {
const db = getDatabase();
const day = sql<string>`to_char(to_timestamp(${pgMessagesTable.created_at} / 1000), 'YYYY-MM-DD')`;
const rows = await db
.select({
day,
total: sql<number>`COUNT(*)::int`,
flagged: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'flagged')::int`,
warned: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'warn')::int`,
clean: sql<number>`COUNT(*) FILTER (WHERE ${pgMessagesTable.ai_status} = 'clean')::int`,
})
.from(pgMessagesTable)
.where(
and(
scopeMessages(guildId, channelId),
// only the last N days
sql`${pgMessagesTable.created_at} >= extract(epoch FROM now() - (${days} || ' days')::interval) * 1000`,
),
)
.groupBy(day)
.orderBy(day);
return JSON.stringify(rows);
}
async function corrections(_guildId?: string, limit = 10): Promise<string> {
const db = getDatabase();
const rows = await db
.select({
id: pgCorrectedModerationsTable.id,
message_id: pgCorrectedModerationsTable.message_id,
original_flags: pgCorrectedModerationsTable.original_flags,
corrected_flags: pgCorrectedModerationsTable.corrected_flags,
correction_notes: pgCorrectedModerationsTable.correction_notes,
content_snippet: pgCorrectedModerationsTable.content_snippet,
created_at: pgCorrectedModerationsTable.created_at,
})
.from(pgCorrectedModerationsTable)
.orderBy(desc(pgCorrectedModerationsTable.created_at))
.limit(limit);
return JSON.stringify(rows);
}
@@ -1 +0,0 @@
export { createChatbotRouter } from "./chatbot.routes.js";
@@ -1,28 +0,0 @@
import type { Router } from "express";
import express from "express";
import { config } from "../../shared/config/index.js";
export function createConfigRouter(): Router {
const router = express.Router();
// GET /api/config
router.get("/config", (_req, res) => {
res.json({
monitorGuildId: config.MONITOR_GUILD_ID || null,
webserverPort: config.WEBSERVER_PORT,
nodeEnv: config.NODE_ENV,
backlogSyncHours: config.BACKLOG_SYNC_HOURS,
backlogSyncBatchSize: config.BACKLOG_SYNC_BATCH_SIZE,
retentionMessagesDays: config.RETENTION_MESSAGES_DAYS,
retentionAttachmentsDays: config.RETENTION_ATTACHMENTS_DAYS,
retentionVoiceDays: config.RETENTION_VOICE_DAYS,
autoDeleteFlaggedEnabled: config.AUTO_DELETE_FLAGGED_ENABLED,
aiAnalysisEnabled: config.AI_ANALYSIS_ENABLED,
voiceGuildId: config.VOICE_GUILD_ID || null,
voiceChannelId: config.VOICE_CHANNEL_ID || null,
logLevel: config.LOG_LEVEL,
});
});
return router;
}
@@ -1 +0,0 @@
export { createConfigRouter } from "./config.routes.js";
@@ -1,111 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { dashboardService } from "./dashboard.service.js";
const logger = createChildLogger("dashboard.routes");
export function createDashboardRouter(): Router {
const router = express.Router();
// GET /api/dashboard/stats — aggregated server statistics
router.get(
"/dashboard/stats",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Fetching dashboard stats");
const stats = await dashboardService.getStats();
res.json(stats);
}),
);
// GET /api/dashboard/activity?days=14 — message volume over time
router.get(
"/dashboard/activity",
asyncHandler(async (req: Request, res: Response) => {
const days = Math.min(Math.max(Number(req.query.days) || 14, 1), 90);
const activity = await dashboardService.getActivity(days);
res.json(activity);
}),
);
// GET /api/dashboard/users — paginated user list with profiles
router.get(
"/dashboard/users",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 20;
const cursor =
typeof req.query.cursor === "string" ? req.query.cursor : undefined;
const search =
typeof req.query.search === "string" ? req.query.search : undefined;
const result = await dashboardService.listUsers({
limit,
cursor,
search,
});
res.json(result);
}),
);
// GET /api/dashboard/users/:userId — single user detail
router.get(
"/dashboard/users/:userId",
asyncHandler(async (req: Request, res: Response) => {
const userId = String(req.params.userId);
const detail = await dashboardService.getUserDetail(userId);
res.json(detail);
}),
);
// GET /api/dashboard/channels — paginated channel list with culture summaries
router.get(
"/dashboard/channels",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 20;
const search =
typeof req.query.search === "string" ? req.query.search : undefined;
const guildId =
typeof req.query.guild_id === "string" ? req.query.guild_id : undefined;
const result = await dashboardService.listChannels({
limit,
search,
guildId,
});
res.json(result);
}),
);
// GET /api/dashboard/channels/:channelId — single channel detail
router.get(
"/dashboard/channels/:channelId",
asyncHandler(async (req: Request, res: Response) => {
const channelId = String(req.params.channelId);
const detail = await dashboardService.getChannelDetail(channelId);
res.json(detail);
}),
);
// GET /api/dashboard/reactions — top reacted messages
router.get(
"/dashboard/reactions",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 20;
const reactions = await dashboardService.getTopReactions(limit);
res.json(reactions);
}),
);
// GET /api/dashboard/reactors — top users by reactions given
router.get(
"/dashboard/reactors",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 20;
const reactors = await dashboardService.getTopReactors(limit);
res.json(reactors);
}),
);
return router;
}
@@ -1 +0,0 @@
export { createDashboardRouter } from "./dashboard.routes.js";
@@ -67,9 +67,9 @@ export const moderationErrors = new Counter({
labelNames: ["type"] as const,
});
export const searxngCalls = new Counter({
name: "moderation_searxng_calls_total",
help: "SearXNG search calls",
export const webSearchCalls = new Counter({
name: "moderation_websearch_calls_total",
help: "Wikipedia web-search calls",
labelNames: ["status"] as const,
});
@@ -1 +0,0 @@
export { createMediaRouter } from "./media.routes.js";
@@ -1,71 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler, validateBody } from "../../shared/middlewares/index.js";
import { mediaLoopSchema, mediaQueueSchema } from "./media.schema.js";
import { getStatus, queue, setLoop, skip, stop } from "./media.service.js";
const logger = createChildLogger("media.routes");
export function createMediaRouter(): Router {
const router = express.Router();
// GET /api/media/status
router.get(
"/media/status",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Media status requested");
const status = await getStatus();
res.json(status);
}),
);
// POST /api/media/queue
router.post(
"/media/queue",
validateBody(mediaQueueSchema),
asyncHandler(async (req: Request, res: Response) => {
const { source, mode } = req.body as {
source: string;
mode: "music" | "screen";
};
logger.debug({ source, mode }, "Media queue requested");
const state = await queue(source, mode);
res.json(state);
}),
);
// POST /api/media/skip
router.post(
"/media/skip",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Media skip requested");
const state = await skip();
res.json(state);
}),
);
// POST /api/media/stop
router.post(
"/media/stop",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Media stop requested");
const state = await stop();
res.json(state);
}),
);
// POST /api/media/loop
router.post(
"/media/loop",
validateBody(mediaLoopSchema),
asyncHandler(async (req: Request, res: Response) => {
const { loop } = req.body as { loop: boolean };
logger.debug({ loop }, "Media loop requested");
const state = await setLoop(loop);
res.json(state);
}),
);
return router;
}
@@ -0,0 +1,43 @@
import { config } from "@/shared/config/index.js";
import { createChildLogger } from "@/shared/logger/index.js";
const logger = createChildLogger("messages-embed");
/**
* Embed a search query with the configured OpenAI-compatible embedding model.
* Uses raw fetch (the backend has no openai SDK dependency) and returns null
* when embeddings are not configured (search unavailable).
*
* encoding_format: "float" is REQUIRED — Nvidia-backed models reject base64.
*/
export async function embedQuery(text: string): Promise<number[] | null> {
if (!config.AI_LLM_API_KEY || !config.AI_LLM_EMBEDDING_MODEL) return null;
try {
const res = await fetch(`${config.AI_LLM_BASE_URL}/embeddings`, {
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${config.AI_LLM_API_KEY}`,
},
body: JSON.stringify({
model: config.AI_LLM_EMBEDDING_MODEL,
input: text,
encoding_format: "float",
}),
});
if (!res.ok) {
logger.warn({ status: res.status }, "query embed HTTP error");
return null;
}
const json = (await res.json()) as {
data?: Array<{ embedding?: number[] }>;
};
return json.data?.[0]?.embedding ?? null;
} catch (error) {
logger.warn(
{ error: error instanceof Error ? error.message : String(error) },
"query embed failed",
);
return null;
}
}
@@ -1 +0,0 @@
export { createMessagesRouter } from "./messages.routes.js";
@@ -1,74 +0,0 @@
import type { Request, Response } from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { messageQuerySchema } from "./messages.schema.js";
import { messagesService } from "./messages.service.js";
const logger = createChildLogger("messages.controller");
export const handleListMessages = asyncHandler(
async (req: Request, res: Response) => {
const query = messageQuerySchema.parse(req.query);
logger.debug({ query }, "Handling list messages request");
const result = await messagesService.listMessages(query);
res.json(result);
},
);
export const handleGetMessagesByChannel = asyncHandler(
async (req: Request, res: Response) => {
if (!req.params.channelId) {
res.status(400).json({ error: "Missing route parameter: channelId" });
return;
}
const channelId = req.params.channelId as string;
const query = messageQuerySchema.parse(req.query);
logger.debug({ channelId, query }, "Handling get messages by channel");
const result = await messagesService.getMessagesByChannel(channelId, query);
res.json(result);
},
);
export const handleGetMessageById = asyncHandler(
async (req: Request, res: Response) => {
if (!req.params.id) {
res.status(400).json({ error: "Missing route parameter: id" });
return;
}
const id = req.params.id as string;
logger.debug({ id }, "Handling get message by ID");
const result = await messagesService.getMessageById(id);
res.json(result);
},
);
export const handleGetImageMessages = asyncHandler(
async (req: Request, res: Response) => {
const guildId = req.query.guildId as string | undefined;
if (!guildId) {
res.status(400).json({ error: "Missing query parameter: guildId" });
return;
}
const limit = Number(req.query.limit) || 50;
logger.debug({ guildId, limit }, "Handling get image messages");
const result = await messagesService.getImageMessages(guildId, limit);
res.json(result);
},
);
export const handleGetAttachmentsByChannel = asyncHandler(
async (req: Request, res: Response) => {
if (!req.params.channelId) {
res.status(400).json({ error: "Missing route parameter: channelId" });
return;
}
const channelId = req.params.channelId as string;
const query = messageQuerySchema.parse(req.query);
logger.debug({ channelId, query }, "Handling get attachments by channel");
const result = await messagesService.getAttachmentsByChannel(
channelId,
query,
);
res.json(result);
},
);
@@ -172,6 +172,71 @@ export class MessagesRepository {
return { data, nextCursor };
}
/**
* Async generator that yields messages ONE AT A TIME for WS streaming.
* Each `.next()` runs its own bounded DB query (limit+1) advancing on the
* `created_at` cursor, so memory stays flat and the caller can emit one WS
* frame per message (no 50-row batch). Stops when a page returns < limit.
*/
async *streamMany(
query: MessageQuery,
pageSize = 50,
): AsyncGenerator<ReturnType<typeof mapMessageRow>, void, unknown> {
const conditions: SQL[] = [];
if (query.guildId) {
conditions.push(eq(pgMessagesTable.guild_id, query.guildId));
}
if (query.channelId) {
conditions.push(eq(pgMessagesTable.channel_id, query.channelId));
}
if (query.userId) {
conditions.push(eq(pgMessagesTable.user_id, query.userId));
}
if (query.status) {
conditions.push(eq(pgMessagesTable.ai_status, query.status));
}
if (EXCLUDED_THREAD_IDS.length > 0) {
const excludeThreads = or(
isNull(pgMessagesTable.thread_id),
notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS),
);
if (excludeThreads) conditions.push(excludeThreads);
}
const where = conditions.length > 0 ? and(...conditions) : undefined;
let cursor: string | undefined = query.cursor;
while (true) {
const pageConditions = where ? [where] : [];
if (cursor) {
pageConditions.push(lt(pgMessagesTable.created_at, Number(cursor)));
}
const pageWhere =
pageConditions.length > 0 ? and(...pageConditions) : undefined;
const db = getDatabase();
const rows = await db
.select()
.from(pgMessagesTable)
.where(pageWhere)
.orderBy(desc(pgMessagesTable.created_at))
.limit(pageSize + 1);
if (rows.length === 0) return;
const hasMore = rows.length > pageSize;
const pageRows = hasMore ? rows.slice(0, pageSize) : rows;
for (const r of pageRows) {
yield mapMessageRow(r as Record<string, unknown>);
}
if (!hasMore) return;
cursor = String(rows[pageSize - 1].created_at);
}
}
async create(data: MessageCreate) {
const db = getDatabase();
const id = crypto.randomUUID();
@@ -1,51 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import {
handleGetAttachmentsByChannel,
handleGetImageMessages,
handleGetMessageById,
handleGetMessagesByChannel,
handleListMessages,
} from "./messages.controller.js";
import { messagesService } from "./messages.service.js";
const logger = createChildLogger("messages.routes");
export function createMessagesRouter(): Router {
const router = express.Router();
// GET /api/messages/images - Get messages with image attachments
// MUST be registered BEFORE /messages/:channelId so "images" is not
// captured as a channelId param.
router.get("/messages/images", handleGetImageMessages);
// GET /api/messages - List messages
router.get("/messages", handleListMessages);
// GET /api/messages/:channelId - Get messages by channel
router.get("/messages/:channelId", handleGetMessagesByChannel);
// GET /api/messages/:channelId/attachments - Get attachments by channel
router.get("/messages/:channelId/attachments", handleGetAttachmentsByChannel);
// GET /api/messages/detail/:id - Get single message by ID
// (uses /detail/ prefix to avoid collision with :channelId route above)
router.get("/messages/detail/:id", handleGetMessageById);
// GET /api/review - Get flagged/warned messages for review
router.get(
"/review",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 20;
const channelId = (req.query.channelId as string) || undefined;
const rows = await messagesService.getReviewMessages(channelId, limit);
logger.debug({ limit, channelId }, "Review query executed");
res.json({ results: rows, limit, cursor: null });
}),
);
return router;
}
@@ -41,3 +41,11 @@ export const messageUpdateSchema = z.object({
export type MessageQuery = z.infer<typeof messageQuerySchema>;
export type MessageCreate = z.infer<typeof messageCreateSchema>;
export type MessageUpdate = z.infer<typeof messageUpdateSchema>;
export const semanticSearchSchema = z.object({
query: z.string().min(1).max(500),
limit: z.coerce.number().int().positive().max(50).default(10),
guildId: z.string().optional(),
});
export type SemanticSearchQuery = z.infer<typeof semanticSearchSchema>;
@@ -1,7 +1,9 @@
import { NotFoundError, ValidationError } from "@/shared/errors/index";
import { createChildLogger } from "@/shared/logger/index";
import { embedQuery } from "./embed.js";
import { messagesRepository } from "./messages.repository.js";
import type { MessageQuery } from "./messages.schema.js";
import type { MessageQuery, SemanticSearchQuery } from "./messages.schema.js";
import { searchArchive } from "./qdrant.js";
const logger = createChildLogger("messages.service");
@@ -15,6 +17,14 @@ export class MessagesService {
return messagesRepository.findMany(query);
}
/**
* Stream messages one at a time (no 50-row batch). The WS handler iterates
* this generator and emits one `message_snapshot` frame per message.
*/
streamMessages(query: MessageQuery, pageSize = 50) {
return messagesRepository.streamMany(query, pageSize);
}
async getMessagesByChannel(channelId: string, query: MessageQuery) {
if (!channelId) {
throw new ValidationError("channelId is required");
@@ -70,6 +80,40 @@ export class MessagesService {
logger.debug({ channelId, limit }, "Getting review messages");
return messagesRepository.getReviewMessages(channelId, limit);
}
/**
* Public, read-only semantic search over the persistent message archive.
* Embeds the query, searches Qdrant, returns text + metadata. Best-effort:
* if embeddings/Qdrant are unavailable, returns an empty result set.
*/
async semanticSearch(
input: SemanticSearchQuery,
): Promise<{ results: ReturnType<typeof mapSearchHit>[]; nextCursor: null }> {
const vector = await embedQuery(input.query);
if (!vector) {
logger.debug(
{ query: input.query },
"semantic search skipped: no embedder",
);
return { results: [], nextCursor: null };
}
const hits = await searchArchive(vector, input.limit, 0.6);
const results = hits.map((h) => mapSearchHit(h));
return { results, nextCursor: null };
}
}
/** Shape returned to the frontend (text + metadata from the archive payload). */
function mapSearchHit(hit: {
score: number;
payload: { text: string; content_hash?: string; analyzed_at: number };
}) {
return {
message_id: hit.payload.content_hash ?? null,
content: hit.payload.text,
score: hit.score,
created_at: hit.payload.analyzed_at,
};
}
export const messagesService = new MessagesService();
@@ -0,0 +1,95 @@
import { config } from "@/shared/config/index.js";
import { createChildLogger } from "@/shared/logger/index.js";
const logger = createChildLogger("messages-qdrant");
export interface ArchiveHit {
score: number;
payload: {
text: string;
content_hash?: string;
analyzed_at: number;
expires_at: number;
};
}
function baseUrl(): string {
return (config.QDRANT_URL ?? "http://100.121.180.82:6333").replace(
/\/+$/,
"",
);
}
function headers(): Record<string, string> {
const h: Record<string, string> = { "Content-Type": "application/json" };
if (config.QDRANT_API_KEY) h["api-key"] = config.QDRANT_API_KEY;
return h;
}
export const ARCHIVE_COLLECTION =
config.QDRANT_ARCHIVE_COLLECTION ?? "gmw_message_archive";
async function request(
method: string,
path: string,
body?: unknown,
timeoutMs = 10_000,
): Promise<unknown> {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), timeoutMs);
try {
const res = await fetch(`${baseUrl()}${path}`, {
method,
headers: headers(),
body: body === undefined ? undefined : JSON.stringify(body),
signal: controller.signal,
});
const text = await res.text();
if (!res.ok) {
throw new Error(
`Qdrant ${method} ${path} -> ${res.status}: ${text.slice(0, 200)}`,
);
}
return text ? JSON.parse(text) : null;
} finally {
clearTimeout(timer);
}
}
/** Search the archive collection for the nearest vectors to `vector`. */
export async function searchArchive(
vector: number[],
limit: number,
scoreThreshold: number,
): Promise<ArchiveHit[]> {
if (!config.QDRANT_URL) return [];
try {
const json = (await request(
"POST",
`/collections/${ARCHIVE_COLLECTION}/points/search`,
{
vector,
limit,
score_threshold: scoreThreshold,
with_payload: true,
},
)) as {
result?: Array<{
score?: number;
payload?: ArchiveHit["payload"];
}>;
};
return (json.result ?? [])
.filter((h) => h.payload?.text)
.map((h) => ({
score: h.score ?? 0,
payload: h.payload as ArchiveHit["payload"],
}));
} catch (error) {
logger.warn(
{ error: error instanceof Error ? error.message : String(error) },
"archive search failed",
);
return [];
}
}
@@ -1 +0,0 @@
export { createModerationRouter } from "./moderation.routes.js";
@@ -17,6 +17,20 @@ const ACTION_TYPES = [
] as const;
const STATUSES = ["pending", "executed", "failed"] as const;
/** Parse a JSON-stringified array column (e.g. flags/categories/evidence).
* Returns null on empty/malformed input so the FE can treat it as "no data". */
function parseJsonArray(value: unknown): string[] | null {
if (value == null) return null;
const str = typeof value === "string" ? value : String(value);
if (str.length === 0) return null;
try {
const parsed = JSON.parse(str);
return Array.isArray(parsed) ? (parsed as string[]) : null;
} catch {
return null;
}
}
export class ModerationRepository {
async getStats() {
const db = getDatabase();
@@ -103,6 +117,13 @@ export class ModerationRepository {
a.error,
a.created_at,
a.executed_at,
a.flags,
a.categories,
a.severity,
a.confidence,
a.score,
a.evidence,
a.policy_version,
m.username,
LEFT(m.content, 300) AS content
FROM moderation_actions a
@@ -126,6 +147,13 @@ export class ModerationRepository {
error: r.error ? String(r.error) : null,
created_at: r.created_at ? Number(r.created_at) : null,
executed_at: r.executed_at ? Number(r.executed_at) : null,
flags: parseJsonArray(r.flags),
categories: parseJsonArray(r.categories),
severity: r.severity ? String(r.severity) : null,
confidence: r.confidence != null ? Number(r.confidence) : null,
score: r.score != null ? Number(r.score) : null,
evidence: parseJsonArray(r.evidence),
policy_version: r.policy_version ? String(r.policy_version) : null,
username: r.username ? String(r.username) : null,
content: r.content ? String(r.content) : null,
}));
@@ -1,43 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "../../shared/logger/index.js";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { moderationService } from "./moderation.service.js";
const logger = createChildLogger("moderation.routes");
export function createModerationRouter(): Router {
const router = express.Router();
// GET /api/moderation/stats — moderation action summary
router.get(
"/moderation/stats",
asyncHandler(async (_req: Request, res: Response) => {
const stats = await moderationService.getStats();
res.json(stats);
}),
);
// GET /api/moderation/actions — paginated moderation action log
router.get(
"/moderation/actions",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 50;
const status = req.query.status as string | undefined;
const actionType = req.query.actionType as string | undefined;
const cursor = req.query.cursor as string | undefined;
const result = await moderationService.listActions({
limit,
status,
actionType,
cursor: cursor ? Number(cursor) : undefined,
});
logger.debug({ count: result.data.length }, "Moderation actions listed");
res.json(result);
}),
);
return router;
}
@@ -1 +0,0 @@
export { createRecordingsRouter } from "./recordings.routes.js";
@@ -1,41 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { recordingsService } from "./recordings.service.js";
const logger = createChildLogger("recordings.routes");
export function createRecordingsRouter(): Router {
const router = express.Router();
// GET /api/recordings
router.get(
"/recordings",
asyncHandler(async (req: Request, res: Response) => {
const limit = Number(req.query.limit) || 50;
const channelId = req.query.channelId as string | undefined;
const userId = req.query.userId as string | undefined;
const cursor = req.query.cursor as string | undefined;
logger.debug({ limit, channelId, userId, cursor }, "Fetching recordings");
const result = await recordingsService.getRecent(limit, {
channelId,
userId,
cursor,
});
res.json(result);
}),
);
// DELETE /api/recordings/:id
router.delete(
"/recordings/:id",
asyncHandler(async (req: Request, res: Response) => {
const id = req.params.id as string;
await recordingsService.deleteById(id);
res.json({ ok: true });
}),
);
return router;
}
@@ -1 +0,0 @@
export { createUiStateRouter } from "./ui-state.routes.js";
@@ -1,34 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { uiStateService } from "./ui-state.service.js";
const logger = createChildLogger("ui-state.routes");
export function createUiStateRouter(): Router {
const router = express.Router();
// GET /api/ui-state
router.get(
"/ui-state",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Fetching UI state");
const state = await uiStateService.getState();
res.json(state);
}),
);
// POST /api/ui-state
router.post(
"/ui-state",
asyncHandler(async (req: Request, res: Response) => {
const updates = req.body as Record<string, unknown>;
logger.debug({ keys: Object.keys(updates) }, "Updating UI state");
const result = await uiStateService.updateState(updates);
res.json(result);
}),
);
return router;
}
@@ -1 +0,0 @@
export { createVoiceRouter } from "./voice.routes.js";
@@ -1,45 +0,0 @@
import type { Request, Response } from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler } from "../../shared/middlewares/index.js";
import { publishCommandNoReply } from "../../shared/redis/index.js";
import type { ConnectVoiceInput, VoiceCommandInput } from "./voice.schema.js";
import {
connectVoice,
disconnectVoice,
getVoiceStatus,
} from "./voice.service.js";
const logger = createChildLogger("voice.controller");
export const handleGetVoiceStatus = asyncHandler(
async (_req: Request, res: Response) => {
const status = await getVoiceStatus();
res.json(status);
},
);
export const handleConnectVoice = asyncHandler(
async (req: Request, res: Response) => {
const { guildId, channelId } = req.body as ConnectVoiceInput;
logger.debug({ guildId, channelId }, "Connecting to voice channel");
const status = await connectVoice(guildId, channelId);
res.json(status);
},
);
export const handleDisconnectVoice = asyncHandler(
async (_req: Request, res: Response) => {
logger.debug("Disconnecting from voice");
const status = await disconnectVoice();
res.json(status);
},
);
export const handleVoiceCommand = asyncHandler(
async (req: Request, res: Response) => {
const { command } = req.body as VoiceCommandInput;
logger.debug({ command }, "Publishing voice command");
await publishCommandNoReply(command);
res.json({ success: true, command });
},
);
@@ -1,80 +0,0 @@
import type { Request, Response, Router } from "express";
import express from "express";
import { createChildLogger } from "@/shared/logger/index";
import { asyncHandler, validateBody } from "../../shared/middlewares/index.js";
import {
handleConnectVoice,
handleDisconnectVoice,
handleGetVoiceStatus,
handleVoiceCommand,
} from "./voice.controller.js";
import { connectVoiceSchema, voiceCommandSchema } from "./voice.schema.js";
import {
getGuilds,
getTextChannels,
getVoiceChannels,
} from "./voice.service.js";
const logger = createChildLogger("voice.routes");
export function createVoiceRouter(): Router {
const router = express.Router();
// ── Guilds ──────────────────────────────────────────────────────────────
// GET /api/guilds
router.get(
"/guilds",
asyncHandler(async (_req: Request, res: Response) => {
logger.debug("Fetching guilds");
const guilds = await getGuilds();
res.json(guilds);
}),
);
// GET /api/guilds/:guildId/channels
router.get(
"/guilds/:guildId/channels",
asyncHandler(async (req: Request, res: Response) => {
const guildId = req.params.guildId as string;
logger.debug({ guildId }, "Fetching text channels");
const channels = await getTextChannels(guildId);
res.json(channels);
}),
);
// GET /api/guilds/:guildId/voice-channels
router.get(
"/guilds/:guildId/voice-channels",
asyncHandler(async (req: Request, res: Response) => {
const guildId = req.params.guildId as string;
logger.debug({ guildId }, "Fetching voice channels");
const channels = await getVoiceChannels(guildId);
res.json(channels);
}),
);
// ── Voice connection ────────────────────────────────────────────────────
// GET /api/voice/status
router.get("/voice/status", handleGetVoiceStatus);
// POST /api/voice/connect
router.post(
"/voice/connect",
validateBody(connectVoiceSchema),
handleConnectVoice,
);
// POST /api/voice/disconnect
router.post("/voice/disconnect", handleDisconnectVoice);
// POST /api/voice/command — send arbitrary voice command (transmit start/stop)
router.post(
"/voice/command",
validateBody(voiceCommandSchema),
handleVoiceCommand,
);
return router;
}
+350
View File
@@ -0,0 +1,350 @@
import { os } from "@orpc/server";
import { z } from "zod";
import { analysisService } from "../modules/analysis/analysis.service";
import { chatRequestSchema } from "../modules/chatbot/chatbot.schema";
import { chatbotService } from "../modules/chatbot/chatbot.service";
// ── Service imports ──────────────────────────────────────────────
import { dashboardService } from "../modules/dashboard/dashboard.service";
import {
mediaLoopSchema,
mediaQueueSchema,
} from "../modules/media/media.schema";
import {
getStatus,
queue,
setLoop,
skip,
stop,
} from "../modules/media/media.service";
import {
messageQuerySchema,
semanticSearchSchema,
} from "../modules/messages/messages.schema";
import { messagesService } from "../modules/messages/messages.service";
import { moderationService } from "../modules/moderation/moderation.service";
import { recordingsService } from "../modules/recordings/recordings.service";
import { uiStateService } from "../modules/ui-state/ui-state.service";
import {
connectVoice,
disconnectVoice,
getGuilds,
getTextChannels,
getVoiceChannels,
getVoiceStatus,
} from "../modules/voice/voice.service";
import { config } from "../shared/config/index";
import { publishCommandNoReply } from "../shared/redis/index";
// ── Dashboard ────────────────────────────────────────────────────
const dashboardRouter = {
stats: os.handler(() => dashboardService.getStats()),
activity: os
.input(
z.object({ days: z.coerce.number().int().min(1).max(90).default(14) }),
)
.handler(({ input }) => dashboardService.getActivity(input.days)),
users: os
.input(
z.object({
limit: z.coerce.number().int().positive().default(20),
cursor: z.string().optional(),
search: z.string().optional(),
}),
)
.handler(({ input }) =>
dashboardService.listUsers({
limit: input.limit,
cursor: input.cursor,
search: input.search,
}),
),
userDetail: os
.input(z.object({ userId: z.string() }))
.handler(({ input }) => dashboardService.getUserDetail(input.userId)),
channels: os
.input(
z.object({
limit: z.coerce.number().int().positive().default(20),
search: z.string().optional(),
guildId: z.string().optional(),
}),
)
.handler(({ input }) =>
dashboardService.listChannels({
limit: input.limit,
search: input.search,
guildId: input.guildId,
}),
),
channelDetail: os
.input(z.object({ channelId: z.string() }))
.handler(({ input }) => dashboardService.getChannelDetail(input.channelId)),
reactions: os
.input(z.object({ limit: z.coerce.number().int().positive().default(20) }))
.handler(({ input }) => dashboardService.getTopReactions(input.limit)),
reactors: os
.input(z.object({ limit: z.coerce.number().int().positive().default(20) }))
.handler(({ input }) => dashboardService.getTopReactors(input.limit)),
};
// ── Messages ─────────────────────────────────────────────────────
const messagesRouter = {
list: os
.input(messageQuerySchema)
.handler(({ input }) => messagesService.listMessages(input)),
byChannel: os
.input(
z.object({
channelId: z.string(),
query: messageQuerySchema,
}),
)
.handler(({ input }) =>
messagesService.getMessagesByChannel(input.channelId, input.query),
),
detail: os
.input(z.object({ id: z.string() }))
.handler(({ input }) => messagesService.getMessageById(input.id)),
images: os
.input(
z.object({
guildId: z.string(),
limit: z.coerce.number().int().positive().default(50),
}),
)
.handler(({ input }) =>
messagesService.getImageMessages(input.guildId, input.limit),
),
attachmentsByChannel: os
.input(
z.object({
channelId: z.string(),
query: messageQuerySchema,
}),
)
.handler(({ input }) =>
messagesService.getAttachmentsByChannel(input.channelId, input.query),
),
review: os
.input(
z.object({
limit: z.coerce.number().int().positive().default(20),
channelId: z.string().optional(),
}),
)
.handler(async ({ input }) => {
const rows = await messagesService.getReviewMessages(
input.channelId,
input.limit,
);
return { results: rows, limit: input.limit, cursor: null };
}),
// Public, read-only semantic search over the message archive.
semanticSearch: os
.input(semanticSearchSchema)
.handler(({ input }) => messagesService.semanticSearch(input)),
};
// ── Moderation ───────────────────────────────────────────────────
const moderationRouter = {
stats: os.handler(() => moderationService.getStats()),
actions: os
.input(
z.object({
limit: z.coerce.number().int().positive().default(50),
status: z.string().optional(),
actionType: z.string().optional(),
cursor: z.coerce.number().int().optional(),
}),
)
.handler(({ input }) =>
moderationService.listActions({
limit: input.limit,
status: input.status,
actionType: input.actionType,
cursor: input.cursor,
}),
),
};
// ── Media ────────────────────────────────────────────────────────
const mediaRouter = {
status: os.handler(() => getStatus()),
queue: os.input(mediaQueueSchema).handler(async ({ input }) => {
await queue(input.source, input.mode);
return getStatus();
}),
skip: os.handler(async () => {
await skip();
return getStatus();
}),
stop: os.handler(async () => {
await stop();
return getStatus();
}),
loop: os.input(mediaLoopSchema).handler(async ({ input }) => {
await setLoop(input.loop);
return getStatus();
}),
};
// ── Voice ─────────────────────────────────────────────────────────
const voiceRouter = {
guilds: os.handler(() => getGuilds()),
textChannels: os
.input(z.object({ guildId: z.string() }))
.handler(({ input }) => getTextChannels(input.guildId)),
voiceChannels: os
.input(z.object({ guildId: z.string() }))
.handler(({ input }) => getVoiceChannels(input.guildId)),
status: os.handler(() => getVoiceStatus()),
connect: os
.input(z.object({ guildId: z.string(), channelId: z.string() }))
.handler(async ({ input }) => {
await connectVoice(input.guildId, input.channelId);
return getVoiceStatus();
}),
disconnect: os.handler(async () => {
await disconnectVoice();
return getVoiceStatus();
}),
command: os
.input(z.object({ command: z.string().min(1) }))
.handler(async ({ input }) => {
await publishCommandNoReply(input.command);
return { success: true, command: input.command };
}),
};
// ── Recordings ───────────────────────────────────────────────────
const recordingsRouter = {
list: os
.input(
z.object({
limit: z.coerce.number().int().positive().default(50),
channelId: z.string().optional(),
userId: z.string().optional(),
cursor: z.string().optional(),
}),
)
.handler(({ input }) =>
recordingsService.getRecent(input.limit, {
channelId: input.channelId,
userId: input.userId,
cursor: input.cursor,
}),
),
delete: os.input(z.object({ id: z.string() })).handler(async ({ input }) => {
await recordingsService.deleteById(input.id);
return { ok: true };
}),
};
// ── Analysis (search) ──────────────────────────────────────────────
const analysisRouter = {
search: os
.input(
z.object({
q: z.string().default(""),
channelId: z.string().optional(),
limit: z.coerce.number().int().positive().default(20),
}),
)
.handler(({ input }) =>
analysisService.search({
q: input.q,
channelId: input.channelId,
limit: input.limit,
}),
),
};
// ── Chatbot ───────────────────────────────────────────────────────
const chatbotRouter = {
chat: os
.input(
chatRequestSchema.extend({
// Per-device actor id; the old REST layer used an X-User-Id header.
// Anonymous sessions use a stable "anonymous" id.
userId: z.string().optional(),
}),
)
.handler(async ({ input }) => {
const userId = input.userId ?? "anonymous";
const response = await chatbotService.processMessage(
input.message,
input.context,
userId,
);
await chatbotService.saveConversation({
userId,
userMessage: input.message,
botResponse: response,
context: input.context,
timestamp: new Date(),
});
return { response, timestamp: new Date().toISOString() };
}),
history: os
.input(
z.object({
limit: z.coerce.number().int().positive().max(100).default(50),
userId: z.string().optional(),
}),
)
.handler(async ({ input }) => {
const userId = input.userId ?? "anonymous";
const history = await chatbotService.getChatHistory(userId, input.limit);
return { history, total: history.length };
}),
clearHistory: os
.input(z.object({ userId: z.string().optional() }))
.handler(async ({ input }) => {
const userId = input.userId ?? "anonymous";
await chatbotService.clearChatHistory(userId);
return { ok: true };
}),
};
// ── Config (public dashboard config snapshot) ──────────────────────
const configRouter = {
get: os.handler(() => ({
monitorGuildId: config.MONITOR_GUILD_ID || null,
webserverPort: config.WEBSERVER_PORT,
nodeEnv: config.NODE_ENV,
backlogSyncHours: config.BACKLOG_SYNC_HOURS,
backlogSyncBatchSize: config.BACKLOG_SYNC_BATCH_SIZE,
retentionMessagesDays: config.RETENTION_MESSAGES_DAYS,
retentionAttachmentsDays: config.RETENTION_ATTACHMENTS_DAYS,
retentionVoiceDays: config.RETENTION_VOICE_DAYS,
autoDeleteFlaggedEnabled: config.AUTO_DELETE_FLAGGED_ENABLED,
aiAnalysisEnabled: config.AI_ANALYSIS_ENABLED,
voiceGuildId: config.VOICE_GUILD_ID || null,
voiceChannelId: config.VOICE_CHANNEL_ID || null,
logLevel: config.LOG_LEVEL,
})),
};
// ── UI State ──────────────────────────────────────────────────────
const uiStateRouter = {
get: os.handler(() => uiStateService.getState()),
update: os
.input(z.record(z.string(), z.unknown()))
.handler(({ input }) => uiStateService.updateState(input)),
};
// ── Root router ───────────────────────────────────────────────────
export const appRouter = {
dashboard: dashboardRouter,
messages: messagesRouter,
moderation: moderationRouter,
media: mediaRouter,
voice: voiceRouter,
recordings: recordingsRouter,
analysis: analysisRouter,
chatbot: chatbotRouter,
config: configRouter,
uiState: uiStateRouter,
};
export type AppRouter = typeof appRouter;
+43
View File
@@ -0,0 +1,43 @@
import type { IncomingMessage, Server } from "node:http";
import type { Duplex } from "node:stream";
import { onError } from "@orpc/server";
import { RPCHandler } from "@orpc/server/ws";
import { WebSocketServer } from "ws";
import { createChildLogger } from "@/shared/logger/index";
import { appRouter } from "./router";
const logger = createChildLogger("orpc.ws");
/**
* Attach the oRPC WebSocket handler to the shared HTTP server, on a path
* SEPARATE from the voice/binary WebSocket (`/ws`). All structured data RPCs
* (dashboard, messages, moderation, media, voice control, recordings,
* analysis, chatbot, config, ui-state) flow over this `/trpc` socket; the
* `/ws` socket is left untouched for Discord PCM audio + gateway events.
*
* We use `noServer` + a manual `upgrade` router (instead of
* `new WebSocketServer({ server, path: "/trpc" })`) because two `ws` servers
* mounted with the `server` option on the SAME http.Server both register
* `upgrade` listeners, and `ws`'s path-guarded listener can reject (400) the
* other server's path. Routing the upgrade ourselves by URL keeps `/trpc`
* and `/ws` fully isolated.
*/
export function createORPCWebSocketServer(server: Server): WebSocketServer {
const handler = new RPCHandler(appRouter, {
interceptors: [
onError((error) => logger.error({ error }, "oRPC WS error")),
],
});
const wss = new WebSocketServer({ noServer: true, perMessageDeflate: false });
server.on("upgrade", (req: IncomingMessage, socket: Duplex, head: Buffer) => {
if (!req.url?.startsWith("/trpc")) return; // let the /ws server handle it
wss.handleUpgrade(req, socket, head, (ws) => {
handler.upgrade(ws, { context: {} });
});
});
logger.info({ path: "/trpc" }, "oRPC WebSocket server attached");
return wss;
}
@@ -134,6 +134,7 @@ export const configSchema = z
.default("https://9router.asepharyana.my.id/v1"),
AI_LLM_MODEL: z.string().default("text"),
AI_LLM_VISION_MODEL: z.string().optional(),
AI_LLM_EMBEDDING_MODEL: z.string().optional(),
AI_LLM_MAX_CONCURRENT: z.coerce.number().int().positive().default(5),
AI_LLM_IMAGE_MAX_DIMENSION: z.coerce
.number()
@@ -208,6 +209,11 @@ export const configSchema = z
.default("https://api.openai.com/v1"),
OPENAI_MODERATION_MODEL: z.string().default("omni-moderation-latest"),
// ── Qdrant (message archive for semantic search) ──────────────────
QDRANT_URL: z.string().optional(),
QDRANT_API_KEY: z.string().optional(),
QDRANT_ARCHIVE_COLLECTION: z.string().default("gmw_message_archive"),
// ── Auto Delete ─────────────────────────────────────────────────────
AUTO_DELETE_FLAGGED_ENABLED: z
.string()
+82 -2
View File
@@ -1,5 +1,7 @@
import type { Server } from "node:http";
import type { IncomingMessage, Server } from "node:http";
import type { Duplex } from "node:stream";
import { WebSocket, WebSocketServer } from "ws";
import { messagesService } from "../modules/messages/messages.service.js";
import { config } from "../shared/config/index.js";
import { BACKEND_COMMAND, BACKEND_VOICE_TRANSMIT } from "../shared/index.js";
import { createChildLogger } from "../shared/logger/index.js";
@@ -96,9 +98,20 @@ export function createWebSocketServer(server: Server): WebSocketServer {
const frontendClients = new Set<WebSocket>();
const gatewayClients = new Set<WebSocket>();
const wss = new WebSocketServer({ server, path: "/ws" });
const wss = new WebSocketServer({ noServer: true, perMessageDeflate: true });
_wss = wss;
// Manual upgrade routing: without this, two `ws` servers bound to the same
// http.Server via the `server` option both register `upgrade` listeners and
// the path-guarded one destructively rejects the other's path (400). We own
// the upgrade event and dispatch by URL instead.
server.on("upgrade", (req: IncomingMessage, socket: Duplex, head: Buffer) => {
if (!req.url?.startsWith("/ws")) return;
wss.handleUpgrade(req, socket, head, (ws) => {
wss.emit("connection", ws, req);
});
});
// Map-based dispatcher for JSON WebSocket message types
const jsonHandlers = new Map<string, MessageHandler>();
@@ -128,6 +141,73 @@ export function createWebSocketServer(server: Server): WebSocketServer {
);
});
// Stream historical messages one-by-one over WS (no 50-row batch).
// The frontend requests it once per channel switch; the backend emits one
// `message_snapshot` frame per message so the UI renders progressively.
jsonHandlers.set("stream_messages", async (ws, message) => {
if (ws.readyState !== WebSocket.OPEN) return;
const payload = (message.payload ?? {}) as {
guildId?: string;
channelId?: string;
cursor?: string;
limit?: number;
};
const guildId = payload.guildId;
const channelId = payload.channelId;
if (!guildId && !channelId) {
logger.warn({ payload }, "stream_messages requires guildId or channelId");
return;
}
const pageSize = 50; // internal DB page size; still emitted one frame at a time
const maxFrames = Math.min(payload.limit ?? 200, 500);
let sent = 0;
let nextCursor: string | null = null;
try {
for await (const msg of messagesService.streamMessages(
{
guildId,
channelId,
cursor: payload.cursor,
} as never,
pageSize,
)) {
if (ws.readyState !== WebSocket.OPEN) break;
// Streamed DESC (newest first); the oldest emitted carries the smallest
// created_at, which is exactly the next-page cursor for "load older".
const createdAt = (msg as { created_at?: number }).created_at;
if (createdAt !== undefined) nextCursor = String(createdAt);
ws.send(
JSON.stringify({
type: "message_snapshot",
data: msg,
}),
);
sent++;
if (sent >= maxFrames) break;
}
if (ws.readyState === WebSocket.OPEN) {
ws.send(
JSON.stringify({
type: "message_snapshot_end",
data: { sent, nextCursor },
}),
);
}
} catch (err) {
logger.error({ err }, "stream_messages failed");
if (ws.readyState === WebSocket.OPEN) {
ws.send(
JSON.stringify({
type: "message_snapshot_end",
data: { sent, nextCursor, error: true },
}),
);
}
}
});
wss.on("connection", (ws: WebSocket, req) => {
// Parse auth token from query string
const rawUrl = req.url ?? "/";
@@ -0,0 +1,57 @@
import { describe, expect, it } from "vitest";
import { tools } from "../src/modules/chatbot/chatbot.toolDefs.js";
const names = tools.map((t) => t.function.name);
describe("chatbot tool definitions", () => {
it("exposes a stable, non-empty tool set", () => {
expect(tools.length).toBeGreaterThanOrEqual(10);
expect(new Set(names).size).toBe(names.length); // no dup names
});
it("every tool declares a name, description, and object parameters", () => {
for (const t of tools) {
expect(t.type).toBe("function");
expect(typeof t.function.name).toBe("string");
expect(t.function.description.length).toBeGreaterThan(10);
expect(t.function.parameters.type).toBe("object");
}
});
it("required-only tools declare required args", () => {
const byName = new Map(tools.map((t) => [t.function.name, t]));
for (const [name, required] of [
["search_messages", "query"],
["get_user_messages", "userId"],
["get_user_profile", "userId"],
["get_user_reputation", "userId"],
["get_channel_culture", "channelId"],
["get_message_detail", "messageId"],
] as const) {
const tool = byName.get(name);
expect(tool, `missing tool ${name}`).toBeDefined();
expect(tool?.function.parameters.required).toContain(required);
}
});
it("covers the core server-watcher situations", () => {
for (const required of [
"get_server_stats",
"get_top_channels",
"get_recent_activity",
"get_top_flagged",
"search_messages",
"get_user_messages",
"get_user_profile",
"get_user_reputation",
"get_channel_culture",
"get_message_detail",
"get_message_reviews",
"get_voice_recordings",
"get_moderation_timeline",
"get_corrections",
]) {
expect(names, `missing ${required}`).toContain(required);
}
});
});
+114
View File
@@ -0,0 +1,114 @@
import { describe, expect, it } from "vitest";
/**
* Lock the contract that the WS `stream_messages` handler + frontend
* `useMessagesStream` depend on.
*
* Real behavior (src/modules/messages/messages.repository.ts → streamMany, and
* src/ws/server.ts stream_messages handler):
* - ONE `stream_messages` request streams the WHOLE history for the scope,
* internally paging `limit+1` at a time (cursor = oldest created_at of the
* page) until exhausted or maxFrames is hit.
* - Messages are emitted ONE AT A TIME, DESC (newest first).
* - The final `message_snapshot_end` carries `nextCursor` = the OLDEST emitted
* row's `created_at`, so the FE's next "load older" request pages forward.
*
* We replicate streamMany's pagination algorithm over an in-memory array so the
* test needs no DB.
*/
type Row = { id: string; created_at: number; guild_id: string };
function makeStream(
rows: Row[],
query: { guildId?: string; channelId?: string; cursor?: string },
pageSize = 50,
): () => Generator<Row, void, unknown> {
return function* () {
let cursor = query.cursor;
while (true) {
const page = rows
.filter((r) => (query.guildId ? r.guild_id === query.guildId : true))
.filter((r) => (cursor ? r.created_at < Number(cursor) : true))
.sort((a, b) => b.created_at - a.created_at)
.slice(0, pageSize + 1);
if (page.length === 0) return;
const hasMore = page.length > pageSize;
const pageRows = hasMore ? page.slice(0, pageSize) : page;
for (const r of pageRows) yield r;
if (!hasMore) return;
cursor = String(page[pageSize - 1].created_at);
}
};
}
function streamAll(
rows: Row[],
query: { guildId?: string; channelId?: string; cursor?: string },
pageSize = 50,
maxFrames = Infinity,
): { data: Row[]; nextCursor: string | null } {
const data: Row[] = [];
let nextCursor: string | null = null;
for (const r of makeStream(rows, query, pageSize)()) {
nextCursor = String(r.created_at);
data.push(r);
if (data.length >= maxFrames) break;
}
return { data, nextCursor };
}
const mk = (id: string, created_at: number, guild_id = "g1"): Row => ({
id,
created_at,
guild_id,
});
describe("messages.streamMany contract", () => {
it("emits newest-first and sets nextCursor to oldest created_at", () => {
const rows = [mk("a", 300), mk("b", 200), mk("c", 100)];
const { data, nextCursor } = streamAll(rows, { guildId: "g1" });
expect(data.map((r) => r.id)).toEqual(["a", "b", "c"]);
expect(nextCursor).toBe("100"); // oldest emitted
});
it("streams the entire history in one request, one frame at a time", () => {
// 120 rows; one request must yield all 120 (no 50-row batch boundary).
const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i));
const { data, nextCursor } = streamAll(rows, { guildId: "g1" }, 50);
expect(data).toHaveLength(120);
expect(data[0].id).toBe("m0"); // newest first
expect(nextCursor).toBe("881"); // oldest = m119 (1000-119)
});
it("honors a frame cap and leaves nextCursor mid-history", () => {
const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i));
const { data, nextCursor } = streamAll(rows, { guildId: "g1" }, 50, 60);
expect(data).toHaveLength(60);
// nextCursor = 60th oldest = m59 (1000-59=941)
expect(nextCursor).toBe("941");
});
it("paginates correctly across subsequent load-older requests", () => {
const rows = Array.from({ length: 120 }, (_, i) => mk(`m${i}`, 1000 - i));
const first = streamAll(rows, { guildId: "g1" }, 50, 50);
expect(first.data).toHaveLength(50);
expect(first.nextCursor).toBe("951"); // 50th oldest = m49
const older = streamAll(
rows,
{ guildId: "g1", cursor: first.nextCursor ?? undefined },
50,
50,
);
expect(older.data[0].id).toBe("m50"); // continues right after m49
expect(older.nextCursor).toBe("901"); // 100th oldest
});
it("filters by guild", () => {
const rows = [mk("x", 500, "g1"), mk("y", 400, "g2")];
const { data } = streamAll(rows, { guildId: "g2" });
expect(data.map((r) => r.id)).toEqual(["y"]);
});
});
@@ -0,0 +1,16 @@
-- Migration: 0015_add_moderation_explainability.sql
-- Date: 2026-08-18
-- Description: Add structured explainability columns to moderation_actions.
-- These are READ (surfaced read-only to the public web) so they are never
-- "written but never read" — they back the public moderation transparency view.
-- Idempotent: no-ops on databases that already carry the columns.
ALTER TABLE IF EXISTS "moderation_actions"
ADD COLUMN IF NOT EXISTS "flags" text,
ADD COLUMN IF NOT EXISTS "categories" text,
ADD COLUMN IF NOT EXISTS "severity" text
CHECK ("severity" IS NULL OR "severity" IN ('none','low','medium','high','critical')),
ADD COLUMN IF NOT EXISTS "confidence" real,
ADD COLUMN IF NOT EXISTS "score" real,
ADD COLUMN IF NOT EXISTS "evidence" text,
ADD COLUMN IF NOT EXISTS "policy_version" text;
@@ -106,6 +106,13 @@
"when": 1785621600000,
"tag": "0014_add_term_glossary_cache",
"breakpoints": true
},
{
"idx": 15,
"version": "7",
"when": 1787184000000,
"tag": "0015_add_moderation_explainability",
"breakpoints": true
}
]
}
Loaded 50 of 130 files, more files were not shown because too many files have changed in this diff. Show more