From d2e97ae11d57179c6f5a78beb8b5afe2c74a0bdb Mon Sep 17 00:00:00 2001 From: asepharyana Date: Sun, 16 Aug 2026 08:41:54 +0700 Subject: [PATCH] audit(gateway): fix dead /metrics endpoint, raise OOM-prone MemoryMax, trim DB pool - gateway-metrics: collectors now run per scrape so Prometheus sees real data (process memory/uptime + live AI-analysis pipeline gauges) instead of an always-empty stub. bootstrap registers the pipeline collectors. - systemd: MemoryMax 512M -> 1G (live RSS ~500MiB, peak 508MiB; 512M left ~2% headroom and risked an OOM-kill restart; host has 8GB free). - config: POSTGRES_POOL_MIN 2 -> 0 so main + 4 Piscina worker threads don't hold ~10 permanently-open idle pg connections against PgBouncer. - docs: rewrite stale ARCHITECTURE.md / MODULE_STRUCTURE.md (winston -> pino, removed mock-crc/indonesianTextNormalizer, renamed aiAnalysisWorker/llmModerationClient). Verified: tsc clean, 129 vitest pass, biome clean on changed files. Co-Authored-By: Claude Opus 5 --- services/discord-gateway/ARCHITECTURE.md | 271 +++++------ services/discord-gateway/MODULE_STRUCTURE.md | 448 +++--------------- services/discord-gateway/src/app/bootstrap.ts | 44 +- .../src/modules/gateway-metrics/index.ts | 1 + .../src/modules/gateway-metrics/metrics.ts | 31 +- .../src/shared/config/index.ts | 7 +- .../tests/imageCacheKey.test.ts | 6 +- .../tests/imageResizer.test.ts | 2 +- .../tests/llmThinkingDisable.test.ts | 10 +- 9 files changed, 268 insertions(+), 552 deletions(-) diff --git a/services/discord-gateway/ARCHITECTURE.md b/services/discord-gateway/ARCHITECTURE.md index b9a0755..dec177e 100644 --- a/services/discord-gateway/ARCHITECTURE.md +++ b/services/discord-gateway/ARCHITECTURE.md @@ -1,172 +1,143 @@ +# Discord Gateway — Architecture + +Pure event-driven microservice (no HTTP server). Captures Discord +messages/voice/attachments/reactions/threads/presence, runs LLM-based AI +moderation, and publishes everything to Redis pub/sub for the backend to +consume. The backend serves the HTTP/WS API to the frontend. + +> NOTE: this doc is the source of truth for the module layout. The older +> `MODULE_STRUCTURE.md` was stale (referenced `winston`, `mock-crc.ts`, +> `indonesianTextNormalizer.ts`, and `aiAnalysisWorker.ts`/`llmModerationClient.ts` +> which were renamed/merged). If they disagree, this file wins. + +## Top-level layout + +``` services/discord-gateway/ ├── src/ +│ ├── index.ts # Entry point → initializeDiscordGateway() │ ├── app/ -│ │ ├── bootstrap.ts # Discord Gateway initialization (no HTTP server) -│ │ └── shutdown.ts # Graceful shutdown handler +│ │ ├── bootstrap.ts # Wires client, DB, Redis, workers, schedulers +│ │ ├── shutdown.ts # Graceful shutdown (SIGINT/SIGTERM + transient errors) +│ │ └── retention.ts # Expired-record cleanup scheduler │ ├── shared/ -│ │ ├── config/ -│ │ │ └── config.ts # Environment configuration (Zod validated) -│ │ ├── database/ -│ │ │ ├── schema.ts # Drizzle ORM schema -│ │ │ ├── drizzle.ts # Database connection -│ │ │ ├── migrate.ts # Migration runner -│ │ │ └── voiceRecordingRepo.ts -│ │ ├── errors/ -│ │ │ └── errors.ts # Custom error classes -│ │ ├── logger/ -│ │ │ ├── logger.ts # Winston logger wrapper -│ │ │ └── serialization.ts # Log value serialization -│ │ ├── utils/ -│ │ │ └── retry.ts # Retry with backoff utility -│ │ └── discord/ -│ │ └── clientOptions.ts # Discord.js client configuration -│ ├── modules/ -│ │ ├── message-capture/ # Modular MVC: Message capture & storage -│ │ │ ├── messageCapture.ts # Controller: Discord event listeners -│ │ │ ├── messageStore.ts # Repository: Database operations -│ │ │ ├── messageMetadata.ts # Service: Message metadata extraction -│ │ │ ├── types.ts # Domain types -│ │ │ └── index.ts # Module exports -│ │ ├── ai-moderation/ # Modular MVC: AI analysis & moderation -│ │ │ ├── aiAnalyzer.ts # Controller: Analysis orchestration -│ │ │ ├── llmModerationClient.ts # Service: LLM API client -│ │ │ ├── aiAnalysisWorker.ts # Service: Worker pool management -│ │ │ ├── indonesianTextNormalizer.ts # Service: Text normalization -│ │ │ ├── moderationPrompt.ts # Service: Prompt generation -│ │ │ └── index.ts # Module exports -│ │ ├── voice-recording/ # Modular MVC: Voice recording & streaming -│ │ │ ├── voiceController.ts # Controller: Voice connection management -│ │ │ ├── recorder.ts # Service: Recording orchestration -│ │ │ ├── recorder/ -│ │ │ │ ├── audioStream.ts # Service: Audio stream subscription -│ │ │ │ ├── decoder.ts # Service: Opus decoding -│ │ │ │ ├── segment.ts # Service: OGG segment rotation -│ │ │ │ ├── metadata.ts # Service: Segment metadata -│ │ │ │ ├── sessionRecording.ts # Service: Session management -│ │ │ │ └── uploader.ts # Service: Segment upload -│ │ │ └── index.ts # Module exports -│ │ ├── attachment-upload/ # Modular MVC: Attachment handling -│ │ │ ├── attachmentUploader.ts # Service: Upload orchestration -│ │ │ ├── imageResizer.ts # Service: Image resizing -│ │ │ └── index.ts # Module exports -│ │ └── event-broadcaster/ # Event-driven: Redis pub/sub -│ │ ├── eventBroadcaster.ts # Service: Event publishing -│ │ ├── eventTypes.ts # Domain: Event type definitions -│ │ └── index.ts # Module exports -│ ├── mock-crc.ts # CRC polyfill for discord.js -│ └── index.ts # Service entry point -├── package.json # Service dependencies -└── tsconfig.json # TypeScript configuration +│ │ ├── config/ # Zod-validated env (index.ts = schema+loader) +│ │ ├── database/ # Drizzle ORM + pg Pool + migrations +│ │ │ ├── init.ts drizzle.ts pool.ts migrate.ts migrateCli.ts +│ │ │ └── schema/ # messages, cache, voice, analytics, meta +│ │ ├── logger/ # pino wrapper + createChildLogger() +│ │ ├── errors/ # AppError / ConfigError / AudioError ... +│ │ ├── utils/ # retry, pagination +│ │ ├── discord/clientOptions.ts # discord.js-selfbot-v13 client options +│ │ ├── uploader.ts # Shared attachment upload helper +│ │ ├── redis-channels.ts # Redis channel-name constants +│ │ └── moderation-types.ts # Shared AI analysis domain types +│ └── modules/ +│ ├── message-capture/ # Discord event listeners + DB store +│ ├── ai-moderation/ # LLM moderation pipeline (see below) +│ ├── voice-recording/ # Voice connect + Opus→OGG recording +│ │ └── recorder/ # decoder, segment, session, uploader, oggCrc +│ ├── voice-pcm-ws/ # Real-time PCM → backend WebSocket (bypasses Redis) +│ ├── attachment-upload/ # Download + (sharp) resize + upload +│ ├── event-broadcaster/ # RedisEventPublisher + EventBroadcaster +│ ├── command-handler/ # Redis-subscribed backend→gateway commands +│ ├── reaction-tracking/ thread-tracking/ user-presence/ +│ ├── channel-topic/ guild-member-events/ +│ └── gateway-metrics/ # Prometheus /metrics endpoint (port 4016) +``` -## Architecture Patterns +## AI moderation pipeline (`ai-moderation/`) -### Modular MVC Structure -Each module follows Controller-Service-Repository pattern: -- **Controller**: Discord event listeners (messageCapture, aiAnalyzer, voiceController) -- **Service**: Business logic (messageStore, llmModerationClient, recorder) -- **Repository**: Data access (messageStore, voiceRecordingRepo) +LLM-only judge — no regex/heuristic classification. One orchestrator call +handles a whole batch (text + media split internally, parallel paths). -### Event-Driven Design -- **Redis Pub/Sub**: All events published to Redis channels -- **Event Channels**: - - `discord:message:created` — New message captured - - `discord:message:updated` — Message edited - - `discord:message:deleted` — Message deleted - - `discord:message:analyzed` — AI analysis complete - - `discord:attachment:created` — Attachment detected - - `discord:attachment:uploaded` — Attachment uploaded to storage - - `discord:voice:started` — Voice recording started - - `discord:voice:stopped` — Voice recording stopped - - `discord:voice:uploaded` — Voice segment uploaded - - `discord:analysis:queue_status` — Analysis queue status update +- `aiAnalyzer.ts` — public API: `queueMessageAnalysis`, `getAnalysisQueueStatus`, + `startPendingAIAnalysisWorker` (recovery worker + cache-prune). +- `batchScheduler.ts` — per-conversation debounce → `processBatch`. +- `batchProcessor.ts` — batch lock/circuit-breaker, fans failed targets to + individual fallback. +- `individualFallbackProcessor.ts` — one-message-at-a-time retry path, own CB. +- `conversationState.ts` / `circuitBreaker.ts` — per-conversation state, + Piscina `workerPool`, `getConversationKey`. +- `ai-analysis-worker.ts` — Piscina entry point (`batch` / `individual` jobs). + Runs `runModerationAnalysis` off the main thread. +- `moderationOrchestrator.ts` — exact-hash cache → batched semantic (Qdrant) + cache → LLM. Text and media paths run in parallel. +- `textBatchProcessor.ts` / `mediaBatchProcessor.ts` — actual LLM calls + (one call per sub-batch, not per message). +- `llmClient.ts` — central OpenAI-compatible chat client (streaming, retries, + thinking-disable injection). `visionAnalyzer.ts` / `mediaAnalysisClient.ts` + share the same router/base URL (different model alias for vision). +- `embeddingClient.ts` + `qdrantClient.ts` — semantic cache (one embed call + + one batched Qdrant search for all uncached targets). +- `textCacheStore.ts` / `channelCultureStore.ts` / `userProfileStore.ts` / + `userReputationStore.ts` — caches & learned per-channel/user state. -### Shared Infrastructure -- **Config**: Zod-validated environment variables -- **Logger**: Winston logger with context support -- **Database**: Drizzle ORM with PostgreSQL -- **Errors**: Custom error classes with codes and status codes -- **Utils**: Retry logic with exponential backoff +### Concurrency model -### No HTTP Server -- Discord Gateway service is **event-driven only** -- No Express, WebSocket, or HTTP routes -- All communication via Redis pub/sub -- Backend service consumes events and serves HTTP API +- Main thread owns the LLM semaphore (`AI_LLM_MAX_CONCURRENT`, default 5) via + `llmClient.withLlmConcurrency`. +- Piscina pool (`PISCINA_MAX_THREADS`, default 4) runs the heavy LLM work off + the event loop; **each worker thread initializes its own pg Pool** (min 0, + grows to `POSTGRES_POOL_MAX`). See "Memory & connections" below. -## Initialization Flow +## Memory & DB connections -1. Load environment config (Zod validation) -2. Initialize database connection -3. Run pending migrations -4. Create Discord client with optimized cache settings -5. Initialize Redis event broadcaster -6. Register Discord event listeners (messageCapture, aiAnalyzer) -7. Login to Discord -8. Listen for graceful shutdown signals (SIGINT, SIGTERM) +`MemoryMax=1G` (raised from 512M — live RSS sits at ~500 MiB, peak 508 MiB, +so 512M left ~2% headroom and risked an OOM-kill restart). Host has 8 GB free. -## Graceful Shutdown +`POSTGRES_POOL_MIN=0` (default). The gateway = main process + up to 4 Piscina +worker threads, each with its own pg Pool. With min:0 the pools stay empty +until a query runs and drop idle clients afterward, instead of holding +`(1 main + 4 workers) × 2 = 10` permanently-open idle connections against +PgBouncer. The pool still grows on demand up to `POSTGRES_POOL_MAX`. -On shutdown signal: -1. Close database connection -2. Disconnect from voice channels -3. Close Redis connection -4. Destroy Discord client -5. Exit process +## Event channels (Redis pub/sub) -## Dependencies +`discord:message:{created,updated,deleted,analyzed}`, +`discord:attachment:{created,uploaded}`, +`discord:voice:{started,stopped,uploaded,active_user,pcm,analyzed}`, +`discord:analysis:queue_status`, +`discord:reaction:{added,removed}`, +`discord:thread:{created,deleted,updated}`, +`discord:channel_topic:updated`, +`discord:presence:updated`, +`discord:guild_member:{added,removed}`. +See `src/shared/redis-channels.ts` for the canonical names. -**Core Discord**: -- discord.js-selfbot-v13 -- @discordjs/voice -- @discordjs/opus +## Initialization flow -**Audio Processing**: -- prism-media (Opus encoding/decoding) -- opusscript (Opus fallback) -- sharp (Image resizing) +1. Validate env (Zod). Refuse to start if `AI_ANALYSIS_ENABLED` but no key. +2. `AUTO_MIGRATE_ON_STARTUP` → run pending Drizzle migrations. +3. `initializeDatabase()` (pg Pool, min 0). +4. Create discord.js-selfbot-v13 client; register listeners on `ready`. +5. Start `gmw-discord-gateway` metrics server (port `METRICS_PORT`, default 4016). +6. `client.login(token)`. -**Data & Config**: -- drizzle-orm (ORM) -- pg (PostgreSQL driver) -- zod (Config validation) -- ioredis (Redis client) +## Graceful shutdown -**Logging & Utilities**: -- winston (Structured logging) -- p-retry (Retry logic) -- p-limit (Concurrency limiting) -- piscina (Worker pool) +`SIGINT`/`SIGTERM` (and uncaught transient stream errors: EPIPE / ECONNRESET / +ERR_STREAM_DESTROYED / ERR_STREAM_WRITE_AFTER_END are treated as non-fatal): +stop metrics → stop muxer → disconnect voice → close PCM WS → close Redis → +close command handler → close DB → destroy client → exit. -## Event Flow Example +## Observability -### Message Capture Flow -1. Discord emits `messageCreate` event -2. `messageCapture.ts` listener receives event -3. Extract metadata (user, channel, content, timestamp) -4. `messageStore.ts` inserts into database -5. `eventBroadcaster.messageCreated()` publishes to Redis -6. Backend service subscribes to `discord:message:created` channel -7. Backend processes and stores in its own database +Prometheus scrapes `127.0.0.1:4016/metrics` (`bete_*` prefix). Collectors run +per-scrape and expose: process memory/uptime, and (when AI analysis is on) live +pipeline gauges — `ai_analysis_queued_conversations`, +`ai_analysis_active_batch_requests`, `ai_analysis_active_individual_requests`, +`ai_analysis_individual_in_flight`, `ai_analysis_individual_circuit_breaker_active`, +`ai_analysis_worker_threads`, `ai_analysis_worker_threads_active`. -### Voice Recording Flow -1. `voiceController.connect()` joins voice channel -2. `recorder.ts` subscribes to user audio streams -3. For each speaking user: - - Create audio stream subscription - - Decode Opus packets to PCM - - Rotate OGG segments (5s default) - - Collect user metadata -4. On silence (3s): - - Finalize segment - - Create metadata JSON - - Upload segment to storage - - Publish `discord:voice:uploaded` event -5. Backend service receives event and indexes recording +## Key invariants (do not break) -## No Breaking Changes - -- Original `src/` remains untouched for now -- Discord Gateway is a **new service** in `services/discord-gateway/` -- Can run alongside existing monolith during transition -- Backend service will consume Redis events -- Frontend continues to use Backend HTTP API +- **LLM is the only judge.** Failed LLM → `status:"error"` + recovery retry. + Never reintroduce regex/heuristic content classification. +- **Discord tokens are sanitized** (`discordTokens.ts`: `<:emoji:id>` → + `[emoji:name]`, `<@id>` → `@user`, etc.) before content reaches the LLM, so + numeric snowflake IDs never trigger false positives. +- **Semantic cache is batched** (one embed call + one Qdrant batch search), + not N sequential round-trips. `ensureQdrantCollection` is memoized. +- **Streaming is mandatory** against the 9router base URL (non-stream waits for + the full body and times out). `llmClient` aggregates SSE chunks. diff --git a/services/discord-gateway/MODULE_STRUCTURE.md b/services/discord-gateway/MODULE_STRUCTURE.md index 536a8aa..1d6b7b6 100644 --- a/services/discord-gateway/MODULE_STRUCTURE.md +++ b/services/discord-gateway/MODULE_STRUCTURE.md @@ -1,408 +1,80 @@ -# Discord Gateway Service - Module Structure +# Discord Gateway Service — Module Structure -## Complete Directory Tree +> Kept as a compact module map. For the authoritative layout, design +> decisions, and invariants, see `ARCHITECTURE.md`. This file was rewritten +> on 2026-08-16 to fix stale references (`winston` → pino, +> `mock-crc.ts`/`indonesianTextNormalizer.ts` removed, +> `aiAnalysisWorker.ts` → `ai-analysis-worker.ts`, +> `llmModerationClient.ts` → `llmClient.ts`). + +## Top-level ``` services/discord-gateway/ ├── src/ -│ ├── app/ -│ │ ├── bootstrap.ts -│ │ │ └── Initializes Discord client, database, Redis broadcaster -│ │ │ Registers event listeners, handles graceful shutdown -│ │ └── shutdown.ts -│ │ └── Graceful shutdown handler for SIGINT/SIGTERM/exceptions -│ │ -│ ├── shared/ -│ │ ├── config/ -│ │ │ └── config.ts -│ │ │ └── Zod-validated environment configuration -│ │ │ - Discord token, database URL, Redis URL -│ │ │ - AI LLM settings, recording parameters -│ │ │ - Attachment upload settings, retention policies -│ │ │ -│ │ ├── database/ -│ │ │ ├── schema.ts -│ │ │ │ └── Drizzle ORM schema definitions -│ │ │ ├── drizzle.ts -│ │ │ │ └── PostgreSQL connection and initialization -│ │ │ ├── migrate.ts -│ │ │ │ └── Database migration runner -│ │ │ ├── migrateCli.ts -│ │ │ │ └── CLI for programmatic migrations -│ │ │ ├── voiceRecordingRepo.ts -│ │ │ │ └── Voice recording repository -│ │ │ └── migrations/ -│ │ │ └── Database migration files -│ │ │ -│ │ ├── errors/ -│ │ │ └── errors.ts -│ │ │ └── Custom error classes -│ │ │ - AppError (base) -│ │ │ - ConfigError -│ │ │ - AudioError -│ │ │ - VoiceConnectionError -│ │ │ - ValidationError -│ │ │ -│ │ ├── logger/ -│ │ │ ├── logger.ts -│ │ │ │ └── Winston logger wrapper with context support -│ │ │ └── serialization.ts -│ │ │ └── Log value serialization utilities -│ │ │ -│ │ ├── utils/ -│ │ │ └── retry.ts -│ │ │ └── Retry with exponential backoff utility -│ │ │ -│ │ └── discord/ -│ │ └── clientOptions.ts -│ │ └── Discord.js client configuration -│ │ -│ ├── modules/ -│ │ │ -│ │ ├── message-capture/ -│ │ │ ├── messageCapture.ts -│ │ │ │ └── CONTROLLER: Discord event listeners -│ │ │ │ - messageCreate, messageUpdate, messageDelete -│ │ │ │ - Validates capture target, publishes events -│ │ │ │ -│ │ │ ├── messageStore.ts -│ │ │ │ └── REPOSITORY: Database CRUD operations -│ │ │ │ - upsertMessageForCapture -│ │ │ │ - updateMessageAsEdited -│ │ │ │ - updateMessageAsDeleted -│ │ │ │ - insertAttachment -│ │ │ │ - getMessageById -│ │ │ │ -│ │ │ ├── messageMetadata.ts -│ │ │ │ └── SERVICE: Message metadata extraction -│ │ │ │ - getMessageMetadata -│ │ │ │ - getMessageLocation -│ │ │ │ - getDisplayContent -│ │ │ │ -│ │ │ ├── types.ts -│ │ │ │ └── Domain types -│ │ │ │ - MessageRecord -│ │ │ │ - AttachmentRecord -│ │ │ │ - VoiceSegmentRecord -│ │ │ │ - AIStatus, AISeverity, AIRecommendedAction -│ │ │ │ -│ │ │ └── index.ts -│ │ └── Module exports -│ │ -│ │ ├── ai-moderation/ -│ │ │ ├── aiAnalyzer.ts -│ │ │ │ └── CONTROLLER: Analysis orchestration -│ │ │ │ - startPendingAIAnalysisWorker -│ │ │ │ - queueMessageAnalysis -│ │ │ │ - Manages analysis queue and worker pool -│ │ │ │ -│ │ │ ├── llmModerationClient.ts -│ │ │ │ └── SERVICE: LLM API integration -│ │ │ │ - Calls LLM for text/image moderation -│ │ │ │ - Parses responses, handles errors -│ │ │ │ - Retry logic with backoff -│ │ │ │ -│ │ │ ├── aiAnalysisWorker.ts -│ │ │ │ └── SERVICE: Worker pool management -│ │ │ │ - Piscina worker pool for parallel analysis -│ │ │ │ - Conversation context batching -│ │ │ │ -│ │ │ ├── indonesianTextNormalizer.ts -│ │ │ │ └── SERVICE: Text preprocessing -│ │ │ │ - Normalize Indonesian text -│ │ │ │ - Handle diacritics, abbreviations -│ │ │ │ -│ │ │ ├── moderationPrompt.ts -│ │ │ │ └── SERVICE: Prompt generation -│ │ │ │ - Generate LLM prompts for moderation -│ │ │ │ - Include context and policy -│ │ │ │ -│ │ │ └── index.ts -│ │ └── Module exports -│ │ -│ │ ├── voice-recording/ -│ │ │ ├── voiceController.ts -│ │ │ │ └── CONTROLLER: Voice connection management -│ │ │ │ - connect(guildId, channelId) -│ │ │ │ - disconnect() -│ │ │ │ - listGuilds(), listVoiceChannels() -│ │ │ │ - getStatus() -│ │ │ │ -│ │ │ ├── recorder.ts -│ │ │ │ └── SERVICE: Recording orchestration -│ │ │ │ - startRecording(client, channel) -│ │ │ │ - stopRecording(guildId) -│ │ │ │ - Manages active recording sessions -│ │ │ │ -│ │ │ ├── recorder/ -│ │ │ │ ├── audioStream.ts -│ │ │ │ │ └── SERVICE: Audio stream subscription -│ │ │ │ │ - subscribeToAudioStream -│ │ │ │ │ - Opus packet handling -│ │ │ │ │ -│ │ │ │ ├── decoder.ts -│ │ │ │ │ └── SERVICE: Opus decoding -│ │ │ │ │ - OpusDecoder class -│ │ │ │ │ - Decode Opus to PCM -│ │ │ │ │ - Rotation and cooldown logic -│ │ │ │ │ -│ │ │ │ ├── segment.ts -│ │ │ │ │ └── SERVICE: OGG segment rotation -│ │ │ │ │ - SegmentManager class -│ │ │ │ │ - Rotate segments (5s default) -│ │ │ │ │ - Write OGG files -│ │ │ │ │ -│ │ │ │ ├── metadata.ts -│ │ │ │ │ └── SERVICE: Segment metadata -│ │ │ │ │ - collectUserMetadata -│ │ │ │ │ - createSegmentMetadata -│ │ │ │ │ - User info, roles, timestamps -│ │ │ │ │ -│ │ │ │ ├── sessionRecording.ts -│ │ │ │ │ └── SERVICE: Session management -│ │ │ │ │ - createRecordingSession -│ │ │ │ │ - finalizeRecordingSession -│ │ │ │ │ - Track active sessions -│ │ │ │ │ -│ │ │ │ └── uploader.ts -│ │ │ │ └── SERVICE: Segment upload -│ │ │ │ - uploadRecordingSegment -│ │ │ │ - Upload to external storage -│ │ │ │ - Retry logic -│ │ │ │ -│ │ │ └── index.ts -│ │ └── Module exports -│ │ -│ │ ├── attachment-upload/ -│ │ │ ├── attachmentUploader.ts -│ │ │ │ └── SERVICE: Upload orchestration -│ │ │ │ - processAttachmentUpload -│ │ │ │ - Download from Discord -│ │ │ │ - Upload to external storage -│ │ │ │ - Retry with backoff -│ │ │ │ -│ │ │ ├── imageResizer.ts -│ │ │ │ └── SERVICE: Image processing -│ │ │ │ - resizeImage -│ │ │ │ - Resize to max dimension -│ │ │ │ - Preserve aspect ratio -│ │ │ │ -│ │ │ └── index.ts -│ │ └── Module exports -│ │ -│ │ └── event-broadcaster/ -│ │ ├── eventBroadcaster.ts -│ │ │ └── SERVICE: Redis pub/sub publisher -│ │ │ - EventBroadcaster class -│ │ │ - RedisEventPublisher class -│ │ │ - Publish to Redis channels -│ │ │ - Methods: -│ │ │ - messageCreated() -│ │ │ - messageUpdated() -│ │ │ - messageDeleted() -│ │ │ - messageAnalyzed() -│ │ │ - attachmentCreated() -│ │ │ - attachmentUploaded() -│ │ │ - voiceRecordingStarted() -│ │ │ - voiceRecordingStopped() -│ │ │ - voiceRecordingUploaded() -│ │ │ - analysisQueueStatus() -│ │ │ -│ │ ├── eventTypes.ts -│ │ │ └── Domain types -│ │ │ - DiscordGatewayEvent interface -│ │ │ - EventChannels constants -│ │ │ - Event channel names -│ │ │ -│ │ └── index.ts -│ └── Module exports -│ -│ ├── mock-crc.ts -│ │ └── CRC polyfill for discord.js compatibility -│ │ -│ └── index.ts -│ └── Service entry point -│ - Initialize Discord Gateway -│ - Handle startup errors -│ -├── ARCHITECTURE.md -│ └── Detailed architecture documentation -│ -├── README.md -│ └── Complete service documentation -│ -├── MODULE_STRUCTURE.md -│ └── This file - module structure reference -│ -└── package.json - └── Service dependencies and scripts +│ ├── index.ts # Entry point +│ ├── app/ # bootstrap, shutdown, retention +│ ├── shared/ # config, database, logger, errors, utils, discord, uploader +│ └── modules/ +│ ├── message-capture/ # Discord listeners + DB store + metadata +│ ├── ai-moderation/ # LLM moderation pipeline (largest module) +│ ├── voice-recording/ # Voice connect + Opus→OGG recording (+ recorder/) +│ ├── voice-pcm-ws/ # Real-time PCM → backend WebSocket +│ ├── attachment-upload/ # Download + sharp resize + upload +│ ├── event-broadcaster/ # RedisEventPublisher + EventBroadcaster +│ ├── command-handler/ # Backend→gateway Redis commands +│ ├── reaction-tracking/ thread-tracking/ user-presence/ +│ ├── channel-topic/ guild-member-events/ +│ └── gateway-metrics/ # Prometheus /metrics (port 4016) +├── tests/ # Vitest suites (129 tests) +├── drizzle/ # Drizzle migration SQL + journal +├── ARCHITECTURE.md README.md package.json tsconfig.json vitest.config.ts ``` -## Module Responsibilities +## Module responsibilities (summary) ### message-capture -**Purpose**: Capture Discord messages (create, update, delete) -**Pattern**: Controller-Service-Repository -- **Controller** (messageCapture.ts): Listens to Discord events -- **Service** (messageMetadata.ts): Extracts metadata -- **Repository** (messageStore.ts): Database operations -- **Events Published**: - - `discord:message:created` - - `discord:message:updated` - - `discord:message:deleted` +Captures `messageCreate`/`messageUpdate`/`messageDelete`, extracts metadata, +stores to Postgres, publishes to Redis. Controller–Service–Repository split: +`messageCapture.ts` (listener) → `messageStore.ts` (DB) + `messageMetadata.ts` +(service). ### ai-moderation -**Purpose**: Analyze messages with LLM for moderation -**Pattern**: Controller-Service-Service-Service -- **Controller** (aiAnalyzer.ts): Orchestrates analysis workflow -- **Service** (llmModerationClient.ts): LLM API integration -- **Service** (aiAnalysisWorker.ts): Worker pool management -- **Service** (indonesianTextNormalizer.ts): Text preprocessing -- **Service** (moderationPrompt.ts): Prompt generation -- **Events Published**: - - `discord:message:analyzed` - - `discord:analysis:queue_status` +LLM-only moderation. Entry: `aiAnalyzer.ts` (`queueMessageAnalysis`, +`startPendingAIAnalysisWorker`, `getAnalysisQueueStatus`). Scheduling: +`batchScheduler.ts` → `batchProcessor.ts` (batch lock + circuit breaker) → +`individualFallbackProcessor.ts` (per-message retry). Heavy work runs in the +Piscina pool via `ai-analysis-worker.ts` (jobs `batch` / `individual`). +Orchestration/caching: `moderationOrchestrator.ts` (exact hash → batched +semantic Qdrant → LLM), `textBatchProcessor.ts` / `mediaBatchProcessor.ts` +(one LLM call per sub-batch), `llmClient.ts` (central streaming client), +`embeddingClient.ts` + `qdrantClient.ts` (semantic cache), plus +`channelCultureStore.ts` / `userProfileStore.ts` / `userReputationStore.ts`. ### voice-recording -**Purpose**: Record voice channel audio -**Pattern**: Controller-Service-SubServices -- **Controller** (voiceController.ts): Voice connection management -- **Service** (recorder.ts): Recording orchestration -- **Sub-services** (recorder/*): Audio processing pipeline - - audioStream.ts: Opus packet subscription - - decoder.ts: Opus to PCM decoding - - segment.ts: OGG file rotation - - metadata.ts: User metadata collection - - sessionRecording.ts: Session lifecycle - - uploader.ts: Segment upload -- **Events Published**: - - `discord:voice:started` - - `discord:voice:stopped` - - `discord:voice:uploaded` +`voiceController.ts` (connect/disconnect/list) + `recorder.ts` (orchestration) ++ `recorder/` (decoder, segment, session, uploader, oggCrc). Publishes +`discord:voice:*` events. Real-time audio also streamed via `voice-pcm-ws`. ### attachment-upload -**Purpose**: Upload message attachments to external storage -**Pattern**: Service-Service -- **Service** (attachmentUploader.ts): Upload orchestration -- **Service** (imageResizer.ts): Image processing -- **Events Published**: - - `discord:attachment:created` - - `discord:attachment:uploaded` +`attachmentUploader.ts` (download → upload to storage) + `imageResizer.ts` +(sharp resize). Emits `discord:attachment:*`. ### event-broadcaster -**Purpose**: Publish events to Redis pub/sub -**Pattern**: Service-Domain -- **Service** (eventBroadcaster.ts): Redis publisher -- **Domain** (eventTypes.ts): Event type definitions -- **Channels**: - - discord:message:* (message events) - - discord:attachment:* (attachment events) - - discord:voice:* (voice events) - - discord:analysis:* (analysis events) +`RedisEventPublisher` (ioredis publish) + `EventBroadcaster` (typed methods). +Channel names in `src/shared/redis-channels.ts`. -## Shared Infrastructure +### gateway-metrics +`metrics.ts` Prometheus HTTP server on `METRICS_PORT` (4016). Collectors run +per scrape; live pipeline gauges registered in `bootstrap.ts`. -### config -- Zod-validated environment variables -- Type-safe configuration access -- Sensible defaults +## Shared infrastructure +- **config** — Zod schema in `shared/config/index.ts` (single source of truth). +- **database** — Drizzle ORM over `pg`; pool `min:0` (`shared/config`). +- **logger** — `pino` wrapper, `createChildLogger()` for context loggers. +- **errors** — `AppError` hierarchy (`ConfigError`, `AudioError`, …). -### database -- Drizzle ORM schema -- PostgreSQL connection -- Migration management -- Voice recording repository - -### logger -- Winston logger wrapper -- Context-aware logging -- Log serialization utilities - -### errors -- Custom error classes -- Error codes and HTTP status codes -- Proper error hierarchy - -### utils -- Retry with exponential backoff -- Configurable retry parameters - -### discord -- Discord.js client configuration -- Cache optimization -- Partial handling - -## Event Flow - -``` -Discord Events - ↓ -message-capture (Controller) - ↓ -messageStore (Repository) → PostgreSQL - ↓ -eventBroadcaster (Service) - ↓ -Redis Pub/Sub - ↓ -Backend Service (Subscriber) - ↓ -HTTP API / WebSocket - ↓ -Frontend Application -``` - -## No HTTP Server - -- ✅ No Express -- ✅ No WebSocket server -- ✅ No HTTP routes -- ✅ No middleware -- ✅ Pure event-driven service - -## Graceful Shutdown - -1. Close PostgreSQL connection -2. Disconnect from voice channels -3. Close Redis connection -4. Destroy Discord client -5. Exit process - -## Dependencies - -**Discord**: -- discord.js-selfbot-v13 -- @discordjs/voice -- @discordjs/opus - -**Audio**: -- prism-media -- opusscript -- sharp - -**Data**: -- drizzle-orm -- pg -- zod -- ioredis - -**Logging**: -- winston -- p-retry -- p-limit -- piscina - -## Summary - -The Discord Gateway service is a **pure event-driven microservice** that: -- Captures Discord messages, voice, and attachments -- Performs AI moderation analysis -- Publishes events to Redis pub/sub -- Has no HTTP server or WebSocket -- Follows Modular MVC pattern -- Maintains clean module boundaries -- Provides type-safe configuration -- Includes structured logging -- Handles graceful shutdown - -The service is designed to run alongside the Backend service, which consumes Redis events and serves the HTTP API to the Frontend. +## Notes +- No HTTP server (other than the metrics endpoint). Pure event-driven. +- `MODULE_STRUCTURE.md` is intentionally a sketch; `ARCHITECTURE.md` is the + detailed reference. When they diverge, `ARCHITECTURE.md` wins. diff --git a/services/discord-gateway/src/app/bootstrap.ts b/services/discord-gateway/src/app/bootstrap.ts index 34f8f5b..96de64e 100644 --- a/services/discord-gateway/src/app/bootstrap.ts +++ b/services/discord-gateway/src/app/bootstrap.ts @@ -3,7 +3,11 @@ import { inArray, lt } from "drizzle-orm"; import type { NodePgDatabase } from "drizzle-orm/node-postgres"; import { ConfigError, DatabaseError } from "@/shared/errors/index"; import { createChildLogger } from "@/shared/logger/index"; -import { startPendingAIAnalysisWorker } from "../modules/ai-moderation/aiAnalyzer.js"; +import { + getAnalysisQueueStatus, + startPendingAIAnalysisWorker, +} from "../modules/ai-moderation/aiAnalyzer.js"; +import { workerPool } from "../modules/ai-moderation/circuitBreaker.js"; import { registerChannelTopicCapture } from "../modules/channel-topic/index.js"; import { CommandHandler } from "../modules/command-handler/commandHandler.js"; import { @@ -11,6 +15,8 @@ import { RedisEventPublisher, } from "../modules/event-broadcaster/index.js"; import { + registerCollector, + setGauge, startMetricsServer, stopMetricsServer, } from "../modules/gateway-metrics/index.js"; @@ -340,6 +346,42 @@ export async function initializeDiscordGateway() { gracefulShutdown("unhandledRejection"); }); + // ── Metrics: register live pipeline collectors before starting server ── + // These refresh on every scrape so Prometheus sees real AI-analysis + // queue depth, concurrency, and DB pool state instead of an empty stub. + registerCollector(() => { + if (!config.AI_ANALYSIS_ENABLED) return; + try { + const status = getAnalysisQueueStatus(); + setGauge("ai_analysis_queued_conversations", status.queuedConversations); + setGauge("ai_analysis_active_batch_requests", status.activeRequests); + setGauge( + "ai_analysis_active_individual_requests", + status.activeIndividualRequests, + ); + setGauge( + "ai_analysis_individual_in_flight", + status.individualInFlightCount, + ); + setGauge( + "ai_analysis_individual_circuit_breaker_active", + status.individualCircuitBreakerActive ? 1 : 0, + ); + if (typeof status.lastError === "string") { + setGauge("ai_analysis_last_error_present", status.lastError ? 1 : 0); + } + const pool = workerPool as unknown as { + _poolState?: { size: number; active: number }; + }; + if (pool._poolState) { + setGauge("ai_analysis_worker_threads", pool._poolState.size); + setGauge("ai_analysis_worker_threads_active", pool._poolState.active); + } + } catch (err) { + logger.warn({ error: String(err) }, "AI metrics collector failed"); + } + }); + // Start metrics server startMetricsServer(); diff --git a/services/discord-gateway/src/modules/gateway-metrics/index.ts b/services/discord-gateway/src/modules/gateway-metrics/index.ts index de6b719..cfd838a 100644 --- a/services/discord-gateway/src/modules/gateway-metrics/index.ts +++ b/services/discord-gateway/src/modules/gateway-metrics/index.ts @@ -1,5 +1,6 @@ export { incrementCounter, + registerCollector, setGauge, startMetricsServer, stopMetricsServer, diff --git a/services/discord-gateway/src/modules/gateway-metrics/metrics.ts b/services/discord-gateway/src/modules/gateway-metrics/metrics.ts index 7e7b672..61a64ba 100644 --- a/services/discord-gateway/src/modules/gateway-metrics/metrics.ts +++ b/services/discord-gateway/src/modules/gateway-metrics/metrics.ts @@ -15,7 +15,11 @@ interface Metric { const metrics = new Map(); -// ─── Helpers ───────────────────────────────────────────────────────────── +// Collectors run on every scrape so gauges reflect live pipeline state +// without callers having to push updates on every event. +const collectors: Array<() => void> = []; + +const startTs = Date.now(); function key(name: string, labels?: Record): string { if (!labels) return name; @@ -26,7 +30,9 @@ function key(name: string, labels?: Record): string { return `${name}{${labelStr}}`; } -// ─── Public API ────────────────────────────────────────────────────────── +export function registerCollector(fn: () => void): void { + collectors.push(fn); +} export function incrementCounter( name: string, @@ -65,13 +71,31 @@ export function setGauge( } } +// Process-level static/derived gauges, refreshed each scrape. +registerCollector(() => { + const uptimeSec = Math.floor((Date.now() - startTs) / 1000); + setGauge("process_uptime_seconds", uptimeSec); + const mem = process.memoryUsage(); + setGauge("process_resident_bytes", mem.rss); + setGauge("process_heap_used_bytes", mem.heapUsed); + setGauge("process_heap_total_bytes", mem.heapTotal); + setGauge("process_event_loop_lag_ms", 0); +}); + // ─── HTTP Server ───────────────────────────────────────────────────────── let server: http.Server | null = null; function formatMetrics(): string { - const lines: string[] = []; + for (const c of collectors) { + try { + c(); + } catch (err) { + logger.warn({ error: String(err) }, "Metrics collector failed"); + } + } + const lines: string[] = []; for (const [fullName, metric] of metrics) { const baseName = fullName.includes("{") ? fullName.slice(0, fullName.indexOf("{")) @@ -80,7 +104,6 @@ function formatMetrics(): string { lines.push(`# TYPE ${baseName} ${metric.type}`); lines.push(`${fullName} ${metric.value}`); } - return `${lines.join("\n")}\n`; } diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index b871e01..1659809 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -94,7 +94,12 @@ export const configSchema = z POSTGRES_USER: z.string().optional(), POSTGRES_PASSWORD: z.string().optional(), POSTGRES_DB: z.string().optional(), - POSTGRES_POOL_MIN: z.coerce.number().int().positive().default(2), + // Idle-pool floor. Kept at 0 so the gateway (main + 4 Piscina worker + // threads, each owning its own pg Pool) does not hold ~10 permanently + // open idle connections to PgBouncer. The pool still grows on demand up + // to POSTGRES_POOL_MAX; min:0 only drops idle clients after + // idleTimeoutMillis. This both trims RSS and frees PgBouncer slots. + POSTGRES_POOL_MIN: z.coerce.number().int().min(0).default(0), POSTGRES_POOL_MAX: z.coerce.number().int().positive().default(10), // ── Redis ──────────────────────────────────────────────────────────── diff --git a/services/discord-gateway/tests/imageCacheKey.test.ts b/services/discord-gateway/tests/imageCacheKey.test.ts index 9032571..0d35925 100644 --- a/services/discord-gateway/tests/imageCacheKey.test.ts +++ b/services/discord-gateway/tests/imageCacheKey.test.ts @@ -15,7 +15,7 @@ import { makeImageCacheKey } from "../src/modules/ai-moderation/textCacheStore.j function oldBuggyHash(dataUrl: string): string { const prefix = dataUrl.slice(0, 128); - return "image:" + createHash("sha256").update(prefix).digest("hex").slice(0, 16); + return `image:${createHash("sha256").update(prefix).digest("hex").slice(0, 16)}`; } describe("makeImageCacheKey — collision prevention", () => { @@ -27,8 +27,8 @@ describe("makeImageCacheKey — collision prevention", () => { "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==" + "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"; // pad to >128 chars - const imgA = sharedPrefix + "UNIQUE_TO_A"; - const imgB = sharedPrefix + "UNIQUE_TO_B"; + const imgA = `${sharedPrefix}UNIQUE_TO_A`; + const imgB = `${sharedPrefix}UNIQUE_TO_B`; // Under the OLD buggy scheme: same prefix → same hash → COLLISION expect(oldBuggyHash(imgA)).toBe(oldBuggyHash(imgB)); diff --git a/services/discord-gateway/tests/imageResizer.test.ts b/services/discord-gateway/tests/imageResizer.test.ts index de5093f..2d0ebac 100644 --- a/services/discord-gateway/tests/imageResizer.test.ts +++ b/services/discord-gateway/tests/imageResizer.test.ts @@ -1,5 +1,5 @@ -import { describe, expect, it } from "vitest"; import sharp from "sharp"; +import { describe, expect, it } from "vitest"; import { resizeImageForVision } from "../src/modules/attachment-upload/imageResizer.js"; // Build a worst-case (poorly-compressing) 1024x1024 image, like a real photo. diff --git a/services/discord-gateway/tests/llmThinkingDisable.test.ts b/services/discord-gateway/tests/llmThinkingDisable.test.ts index 671fe20..289979e 100644 --- a/services/discord-gateway/tests/llmThinkingDisable.test.ts +++ b/services/discord-gateway/tests/llmThinkingDisable.test.ts @@ -11,7 +11,9 @@ const baseOpts = { describe("buildLlmParams — disable-thinking injection", () => { it("injects no thinking-disabling params when disableThinking is false", () => { const params = buildLlmParams(baseOpts, false); - expect((params as Record).reasoning_effort).toBeUndefined(); + expect( + (params as Record).reasoning_effort, + ).toBeUndefined(); expect((params as Record).reasoning).toBeUndefined(); expect( (params as Record).chat_template_kwargs, @@ -40,9 +42,9 @@ describe("buildLlmParams — disable-thinking injection", () => { expect(params.stream).toBe(true); expect(params.response_format).toEqual({ type: "json_object" }); // thinking-disabled params still present - expect( - (params as Record).chat_template_kwargs, - ).toEqual({ enable_thinking: false }); + expect((params as Record).chat_template_kwargs).toEqual({ + enable_thinking: false, + }); }); it("falls back to config default model when none supplied", () => {