docs(gateway): consolidate README/ARCHITECTURE, drop stale MODULE_STRUCTURE

README.md was the extraction-era document (referenced winston, mock-crc.ts,
llmModerationClient.ts, indonesianTextNormalizer.ts — all long gone) and
duplicated ARCHITECTURE.md. Rewritten as a short run-the-service guide;
layout/design lives only in ARCHITECTURE.md.

MODULE_STRUCTURE.md deleted: it was a stale duplicate of ARCHITECTURE.md,
referenced by nothing but itself.

ARCHITECTURE.md updated to the post-refactor reality: app/ lifecycle split
(bootstrap/lifecycle/process-guards/metrics-collector), ai-moderation
recovery-worker + cache-prune, per-module index.ts facades, one-way
dependency rule, corrected init/shutdown/observability sections.

services/discord-gateway/ARCHITECTURE.md

@@ -5,10 +5,9 @@ messages/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
+> NOTE: this doc is the source of truth for the module layout. The old
-> `MODULE_STRUCTURE.md` was stale (referenced `winston`, `mock-crc.ts`,
+> `MODULE_STRUCTURE.md` was a stale duplicate and has been removed. `README.md`
-> `indonesianTextNormalizer.ts`, and `aiAnalysisWorker.ts`/`llmModerationClient.ts`
+> only covers how to run the service.
-> which were renamed/merged). If they disagree, this file wins.
 
 ## Top-level layout
 
@@ -16,33 +15,41 @@ consume. The backend serves the HTTP/WS API to the frontend.
 services/discord-gateway/
 ├── src/
 │   ├── index.ts                     # Entry point → initializeDiscordGateway()
-│   ├── app/
+│   ├── app/                         # Process lifecycle
-│   │   ├── bootstrap.ts             # Wires client, DB, Redis, workers, schedulers
+│   │   ├── bootstrap.ts             # Startup order: config → DB → services → metrics → login
-│   │   ├── shutdown.ts              # Graceful shutdown (SIGINT/SIGTERM + transient errors)
+│   │   ├── lifecycle.ts             # Everything wired on the Discord 'ready' hook
+│   │   ├── process-guards.ts        # SIGINT/SIGTERM + uncaught-error policy
+│   │   ├── metrics-collector.ts     # AI pipeline Prometheus gauges
+│   │   ├── shutdown.ts              # Graceful shutdown sequence
 │   │   └── retention.ts             # Expired-record cleanup scheduler
-│   ├── shared/
+│   ├── shared/                      # Infrastructure — never imports from modules/
 │   │   ├── 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, meta, analytics
 │   │   ├── logger/                  # pino wrapper + createChildLogger()
-│   │   ├── errors/                  # AppError / ConfigError ...
+│   │   ├── errors/                  # AppError / ConfigError ... + errorMessage()
+│   │   │                            #   + isTransientStreamError()
 │   │   ├── 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
+│   │   ├── redis-channels.ts        # Redis channel + command constants
 │   │   └── moderation-types.ts      # Shared AI analysis domain types
-│   └── modules/
+│   └── modules/                     # Feature modules, each with an index.ts facade
 │       ├── message-capture/         # Discord event listeners + DB store
 │       ├── ai-moderation/           # LLM moderation pipeline (see below)
 │       ├── attachment-upload/       # Download + (sharp) resize + upload
-│       ├── event-broadcaster/        # RedisEventPublisher + EventBroadcaster
+│       ├── event-broadcaster/       # RedisEventPublisher + EventBroadcaster
 │       ├── command-handler/         # Redis-subscribed backend→gateway commands
 │       ├── reaction-tracking/ thread-tracking/ user-presence/
-│       ├── channel-topic/ guild-member-events/
+│       ├── channel-topic/ guild-member-events/ monitor/
 │       └── gateway-metrics/         # Prometheus /metrics endpoint (port 4016)
 ```
 
+Dependency direction is one-way: `index.ts` → `app/` → `modules/` → `shared/`.
+Code outside a module imports its `index.ts` facade, never an internal file;
+deep imports stay valid inside the module itself.
+
 ## AI moderation pipeline (`ai-moderation/`)
 
 LLM-only judge — no regex/heuristic classification. One orchestrator call
@@ -54,8 +61,15 @@ concurrency semaphores. The text lane frees its lock and saves+broadcasts the
 moment text analysis finishes — it never waits on a slow vision/media batch
 of the same conversation, and vice versa.
 
-- `aiAnalyzer.ts` — public API: `queueMessageAnalysis`, `getAnalysisQueueStatus`,
+- `aiAnalyzer.ts` — public API: `queueMessageAnalysis`, `queueConversationAnalysis`,
-  `startPendingAIAnalysisWorker` (recovery worker + cache-prune).
+  `getAnalysisQueueStatus`, `startPendingAIAnalysisWorker`. Short-circuits
+  age-restricted and skip-list messages before any LLM work.
+- `recovery-worker.ts` — periodic sweep for stranded `pending` messages
+  (re-scheduled per lane) and `error`/`analysis_incomplete` messages
+  (individual fallback queue); prunes stale lane locks, per-conversation CB
+  counters and individual in-flight markers.
+- `cache-prune.ts` — throttled (6h) expired-verdict sweep across Postgres and
+  Qdrant, driven from the recovery interval.
 - `batchScheduler.ts` — per-conversation per-LANE debounce → `processBatch`
   (lane-aware). `splitMessagesByLane` / `laneOfMessage` live in
   `analysisLanes.ts` (pure, unit-testable).
@@ -123,28 +137,51 @@ See `src/shared/redis-channels.ts` for the canonical names.
 
 ## Initialization flow
 
+`bootstrap.ts` runs these steps in order (each is a named function):
+
 1. Validate env (Zod). Refuse to start if `AI_ANALYSIS_ENABLED` but no key.
-2. `AUTO_MIGRATE_ON_STARTUP` → run pending Drizzle migrations.
+   → `assertConfigIsUsable()`
-3. `initializeDatabase()` (pg Pool, min 0).
+2. Build long-lived services: Discord client, `RedisEventPublisher` +
-4. Create discord.js-selfbot-v13 client; register listeners on `ready`.
+   `EventBroadcaster`, `CommandHandler`; install the shutdown handler.
-5. Start `gmw-discord-gateway` metrics server (port `METRICS_PORT`, default 4016).
+3. Connect infrastructure → `connectDatabase()`:
-6. `client.login(token)`.
+   `AUTO_MIGRATE_ON_STARTUP` runs pending Drizzle migrations, then
+   `initializeDatabase()` (pg Pool, min 0).
+4. `registerClientDebugLogging()` — only client debug lines carrying signal.
+5. Install process guards (`registerProcessGuards`).
+6. Register pipeline gauges + start the metrics server (port `METRICS_PORT`,
+   default 4016).
+7. `client.login(token)`.
+
+On the Discord `ready` event, `lifecycle.ts` runs `startGatewayLifecycle()`:
+
+1. Inject the event broadcaster into message-capture and moderation-actions
+   (before any listener can fire).
+2. Register Discord listeners: message-capture, reaction, thread, presence,
+   channel-topic, guild-member.
+3. Start background work: AI analysis worker + recovery worker, command
+   handler, retention cleanup, weekly digest.
 
 ## Graceful shutdown
 
-`SIGINT`/`SIGTERM` (and uncaught transient stream errors: EPIPE / ECONNRESET /
+`process-guards.ts` owns the policy. `SIGINT`/`SIGTERM` and non-transient
-ERR_STREAM_DESTROYED / ERR_STREAM_WRITE_AFTER_END are treated as non-fatal):
+uncaught exceptions/rejections run `shutdown.ts`; transient stream errors
-stop metrics → close event broadcaster (Redis) → close command handler →
+(EPIPE / ECONNRESET / ERR_STREAM_DESTROYED / ERR_STREAM_WRITE_AFTER_END, see
-close DB → destroy client → exit.
+`isTransientStreamError()`) are logged and IGNORED so the bot stays online.
+
+Shutdown order: stop metrics → close event broadcaster (Redis) → close command
+handler → close DB → destroy client → exit.
 
 ## Observability
 
 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`,
+pipeline gauges registered by `app/metrics-collector.ts` —
-`ai_analysis_active_batch_requests`, `ai_analysis_active_individual_requests`,
+`ai_analysis_queued_conversations`, `ai_analysis_active_batch_requests`,
-`ai_analysis_individual_in_flight`, `ai_analysis_individual_circuit_breaker_active`,
+`ai_analysis_active_text_requests`, `ai_analysis_active_media_requests`,
-`ai_analysis_worker_threads`, `ai_analysis_worker_threads_active`.
+`ai_analysis_active_individual_requests`, `ai_analysis_individual_in_flight`,
+`ai_analysis_individual_circuit_breaker_active`,
+`ai_analysis_worker_threads_{text,media}`,
+`ai_analysis_worker_threads_active_{text,media}`.
 
 ## Key invariants (do not break)
 

services/discord-gateway/MODULE_STRUCTURE.md

@@ -1,73 +0,0 @@
-# Discord Gateway Service — Module Structure
-
-> 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/
-│   ├── 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)
-│       ├── 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 (summary)
-
-### message-capture
-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
-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`.
-
-### attachment-upload
-`attachmentUploader.ts` (download → upload to storage) + `imageResizer.ts`
-(sharp resize). Emits `discord:attachment:*`.
-
-### event-broadcaster
-`RedisEventPublisher` (ioredis publish) + `EventBroadcaster` (typed methods).
-Channel names in `src/shared/redis-channels.ts`.
-
-### gateway-metrics
-`metrics.ts` Prometheus HTTP server on `METRICS_PORT` (4016). Collectors run
-per scrape; live pipeline gauges registered in `bootstrap.ts`.
-
-## 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`, …).
-
-## 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.

services/discord-gateway/README.md

@@ -1,319 +1,63 @@
-# Discord Gateway Service - Extraction Complete
+# Discord Gateway
 
-## Overview
+Event-driven selfbot service: captures Discord events, runs LLM moderation,
+publishes everything to Redis for the backend to consume.
 
-Successfully extracted Discord Gateway service with **Modular MVC + Event-Driven Architecture** using Redis pub/sub for inter-service communication.
+> Architecture, invariants and the AI pipeline are documented in
+> **`ARCHITECTURE.md`** — that file is the source of truth. This README only
+> covers how to run it.
 
-## Directory Structure
+## Commands
 
-```
+```bash
-services/discord-gateway/
+pnpm install
-├── src/
+pnpm typecheck     # tsc --noEmit
-│   ├── app/
+pnpm lint          # biome check --diagnostic-level=error .
-│   │   ├── bootstrap.ts          # Service initialization (Discord client, DB, Redis)
+pnpm test          # vitest run (138 tests)
-│   │   └── shutdown.ts           # Graceful shutdown handler
+pnpm build         # tsc — CI/prod builds run this inside nix, which also
-│   ├── shared/                   # Shared infrastructure layer
+                   # runs scripts/fix-imports.mjs to rewrite @/ aliases and
-│   │   ├── config/
+                   # extensionless imports for Node ESM
-│   │   │   └── config.ts         # Zod-validated environment config
+pnpm dev           # tsx watch src/index.ts
-│   │   ├── database/
+pnpm start         # node dist/index.js
-│   │   │   ├── schema.ts         # Drizzle ORM schema
-│   │   │   ├── drizzle.ts        # PostgreSQL connection
-│   │   │   ├── migrate.ts        # Migration runner
-│   │   ├── errors/
-│   │   │   └── errors.ts         # Custom error classes
-│   │   ├── logger/
-│   │   │   ├── logger.ts         # Winston logger wrapper
-│   │   │   └── serialization.ts  # Log serialization
-│   │   ├── utils/
-│   │   │   └── retry.ts          # Retry with exponential backoff
-│   │   └── discord/
-│   │       └── clientOptions.ts  # Discord.js client config
-│   ├── modules/                  # Feature modules (Modular MVC)
-│   │   ├── message-capture/      # Controller-Service-Repository
-│   │   │   ├── messageCapture.ts # Controller: Discord event listeners
-│   │   │   ├── messageStore.ts   # Repository: DB operations
-│   │   │   ├── messageMetadata.ts # Service: Metadata extraction
-│   │   │   ├── types.ts          # Domain types
-│   │   │   └── index.ts          # Module exports
-│   │   ├── ai-moderation/        # Controller-Service-Repository
-│   │   │   ├── aiAnalyzer.ts     # Controller: Analysis orchestration
-│   │   │   ├── llmModerationClient.ts # Service: LLM API client
-│   │   │   ├── aiAnalysisWorker.ts # Service: Worker pool
-│   │   │   ├── indonesianTextNormalizer.ts # Service: Text normalization
-│   │   │   ├── moderationPrompt.ts # Service: Prompt generation
-│   │   │   └── index.ts          # Module exports
-│   │   ├── attachment-upload/    # Controller-Service-Repository
-│   │   │   ├── attachmentUploader.ts # Service: Upload orchestration
-│   │   │   ├── imageResizer.ts   # Service: Image resizing
-│   │   │   └── index.ts          # Module exports
-│   │   └── event-broadcaster/    # Event-driven layer
-│   │       ├── eventBroadcaster.ts # Service: Redis pub/sub publisher
-│   │       ├── eventTypes.ts     # Domain: Event type definitions
-│   │       └── index.ts          # Module exports
-│   ├── mock-crc.ts               # CRC polyfill for discord.js
-│   └── index.ts                  # Service entry point
-├── ARCHITECTURE.md               # Detailed architecture documentation
-├── package.json                  # Service dependencies
-└── tsconfig.json                 # TypeScript configuration (inherited)
 ```
 
-## Architecture Patterns
+Deployment is CI-only: `nix build .#discord-gateway` → Attic cache → systemd
+restart on the VPS. Do not build/hand-copy the artifact.
 
-### 1. Modular MVC Structure
+## Layout
-Each feature module follows **Controller-Service-Repository** pattern:
-
-**Message Capture Module**:
-- **Controller** (`messageCapture.ts`): Listens to Discord events (messageCreate, messageUpdate, messageDelete)
-- **Service** (`messageMetadata.ts`): Extracts and normalizes message metadata
-- **Repository** (`messageStore.ts`): Database CRUD operations
-
-**AI Moderation Module**:
-- **Controller** (`aiAnalyzer.ts`): Orchestrates analysis workflow
-- **Service** (`llmModerationClient.ts`): LLM API integration
-- **Service** (`aiAnalysisWorker.ts`): Worker pool management
-- **Service** (`indonesianTextNormalizer.ts`): Text preprocessing
-
-**Attachment Upload Module**:
-- **Service** (`attachmentUploader.ts`): Upload orchestration
-- **Service** (`imageResizer.ts`): Image processing
-
-### 2. Event-Driven Architecture
-**Redis Pub/Sub** replaces WebSocket broadcaster:
 
 ```
-Discord Events → Discord Gateway Service → Redis Pub/Sub → Backend Service
+src/
-                                         ↓
+├── index.ts                 # entry → initializeDiscordGateway()
-                                    Event Channels:
+├── app/                     # process lifecycle
-                                    - discord:message:created
+│   ├── bootstrap.ts         #   startup order: config → DB → services → metrics → login
-                                    - discord:message:updated
+│   ├── lifecycle.ts         #   everything wired on the Discord 'ready' hook
-                                    - discord:message:deleted
+│   ├── process-guards.ts    #   SIGINT/SIGTERM + uncaught error policy
-                                    - discord:message:analyzed
+│   ├── metrics-collector.ts #   AI pipeline Prometheus gauges
-                                    - discord:attachment:created
+│   ├── shutdown.ts          #   graceful shutdown sequence
-                                    - discord:attachment:uploaded
+│   └── retention.ts         #   expired-record cleanup scheduler
-                                    - discord:analysis:queue_status
+├── shared/                  # infrastructure — never imports from modules/
+│   ├── config/ database/ logger/ errors/ utils/
+│   ├── discord/clientOptions.ts
+│   ├── redis-channels.ts    #   canonical Redis channel + command constants
+│   └── moderation-types.ts  #   domain types shared across services
+└── modules/                 # feature modules (each exposes an index.ts facade)
+    ├── ai-moderation/       #   LLM moderation pipeline (largest module)
+    ├── message-capture/     #   Discord listeners + message/attachment DB
+    ├── attachment-upload/   #   download → resize → upload
+    ├── event-broadcaster/   #   Redis pub/sub publisher
+    ├── command-handler/     #   backend → gateway commands over Redis
+    ├── gateway-metrics/     #   Prometheus /metrics (port 4016)
+    ├── monitor/             #   weekly digest scheduler
+    └── reaction-tracking/ thread-tracking/ user-presence/
+        channel-topic/ guild-member-events/
 ```
 
-### 3. Shared Infrastructure Layer
+Dependency direction is one-way: `index.ts` → `app/` → `modules/` → `shared/`.
-Centralized, reusable components:
+Callers outside a module import its `index.ts` facade, never an internal file.
-- **Config**: Zod-validated environment variables
-- **Logger**: Winston logger with context support
-- **Database**: Drizzle ORM with PostgreSQL
-- **Errors**: Custom error classes with codes and HTTP status codes
-- **Utils**: Retry logic with exponential backoff
-- **Discord**: Client configuration and options
 
-### 4. No HTTP Server
+## Testing
-- **Event-driven only**: No Express, WebSocket, or HTTP routes
-- **Redis pub/sub**: All inter-service communication via Redis
-- **Backend service**: Consumes events and serves HTTP API
-- **Frontend**: Continues to use Backend HTTP API
 
-## Key Features
+Vitest, tests in `tests/`. Config supplies dummy env vars so the suite runs
-
+without live Postgres/Redis/Qdrant; external services are mocked. `llmE2e.test.ts`
-### Message Capture
+is skipped by default and needs real credentials (`pnpm test:e2e:live`).
-1. Discord emits `messageCreate`, `messageUpdate`, `messageDelete` events
-2. `messageCapture.ts` listener receives and validates event
-3. Extract metadata: user, channel, content, timestamp, attachments
-4. `messageStore.ts` inserts into PostgreSQL
-5. `eventBroadcaster.messageCreated()` publishes to Redis
-6. Backend service subscribes and processes
-
-### AI Moderation
-1. `aiAnalyzer.ts` queues messages for analysis
-2. `llmModerationClient.ts` calls LLM API with context
-3. `indonesianTextNormalizer.ts` preprocesses text
-4. Results stored in database
-5. `eventBroadcaster.messageAnalyzed()` publishes results
-6. Backend service receives and updates UI
-
-### Attachment Upload
-1. `messageCapture.ts` detects attachments
-2. `attachmentUploader.ts` downloads from Discord
-3. `imageResizer.ts` resizes images if needed
-4. Upload to external storage with retry logic
-5. `eventBroadcaster.attachmentUploaded()` publishes
-6. Backend service stores metadata
-
-## Initialization Flow
-
-```
-1. Load environment config (Zod validation)
-   ↓
-2. Initialize PostgreSQL connection
-   ↓
-3. Run pending database migrations
-   ↓
-4. Create Discord client with optimized cache
-   ↓
-5. Initialize Redis event broadcaster
-   ↓
-6. Register Discord event listeners
-   - messageCapture (message events)
-   - aiAnalyzer (analysis worker)
-   ↓
-7. Login to Discord
-   ↓
-8. Listen for graceful shutdown signals
-```
-
-## Graceful Shutdown
-
-On SIGINT/SIGTERM/uncaughtException/unhandledRejection:
-1. Close PostgreSQL connection
-2. Close Redis connection
-3. Destroy Discord client
-4. Exit process (code 0 for clean, 1 for error)
-
-## Dependencies
-
-**Core Discord**:
-- `discord.js-selfbot-v13` — Discord client (selfbot variant)
-
-**Media Processing**:
-- `sharp` — Image resizing
-
-**Data & Config**:
-- `drizzle-orm` — Type-safe ORM
-- `pg` — PostgreSQL driver
-- `zod` — Config validation
-- `ioredis` — Redis client
-
-**Logging & Utilities**:
-- `winston` — Structured logging
-- `p-retry` — Retry with backoff
-- `p-limit` — Concurrency limiting
-- `piscina` — Worker pool
-
-## No Breaking Changes
-
-- Original `src/` remains untouched
-- 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
-
-## Next Steps
-
-1. **Create Backend service** (`services/backend/`)
-   - HTTP API endpoints
-   - Redis event subscribers
-   - Database models
-   - WebSocket broadcaster
-
-2. **Update Frontend** (`frontend/`)
-   - Connect to Backend HTTP API
-   - Subscribe to WebSocket events
-
-3. **Nix & CI/CD**
-   - flake.nix package for Discord Gateway
-   - systemd services (gmw-backend, gmw-discord-gateway)
-   - GitHub Actions for build/deploy (nix copy → systemctl restart)
-
-4. **Documentation**
-   - API documentation
-   - Event schema documentation
-   - Deployment guide
-
-## Files Created
-
-**Total: 43 files**
-
-### Shared Infrastructure (9 files)
-- `src/shared/config/config.ts`
-- `src/shared/database/` (5 files)
-- `@bete/shared/errors` (shared package)
-- `src/shared/logger/logger.ts`
-- `src/shared/logger/serialization.ts`
-- `src/shared/utils/retry.ts`
-- `src/shared/discord/clientOptions.ts`
-
-### Modules (28 files)
-- `src/modules/message-capture/` (5 files)
-- `src/modules/ai-moderation/` (6 files)
-- `src/modules/attachment-upload/` (3 files)
-- `src/modules/event-broadcaster/` (3 files)
-
-### App & Entry (4 files)
-- `src/app/bootstrap.ts`
-- `src/app/shutdown.ts`
-- `src/index.ts`
-- `src/mock-crc.ts`
-
-### Configuration (2 files)
-- `package.json`
-- `ARCHITECTURE.md`
-
-## Verification Checklist
-
-✅ Directory structure created
-✅ Shared infrastructure migrated
-✅ Message capture module migrated
-✅ AI moderation module migrated
-✅ Attachment upload module migrated
-✅ Event broadcaster module created (Redis pub/sub)
-✅ Bootstrap and entry point created
-✅ Package.json with dependencies
-✅ No HTTP server code (Express, WebSocket removed)
-✅ Event-driven architecture implemented
-✅ Graceful shutdown handler
-✅ Module index files for clean exports
-✅ Architecture documentation
-
-## Event Flow Diagram
-
-```
-┌─────────────────────────────────────────────────────────────────┐
-│                     Discord Gateway Service                     │
-├─────────────────────────────────────────────────────────────────┤
-│                                                                 │
-│  ┌────────────────────────────┐  ┌────────────────────────────┐ │
-│  │      Message Capture       │  │       AI Moderation        │ │
-│  │        (Controller)        │  │        (Controller)        │ │
-│  └──────────────┬─────────────┘  └──────────────┬─────────────┘ │
-│                 │                               │               │
-│                 ├───────────────────────────────┤               │
-│                 │                               │               │
-│                 ▼                               ▼               │
-│  ┌───────────────────────────────────────────────────────────┐  │
-│  │             Event Broadcaster (Redis Pub/Sub)             │  │
-│  │  - discord:message:created                                │  │
-│  │  - discord:message:updated                                │  │
-│  │  - discord:message:deleted                                │  │
-│  │  - discord:message:analyzed                               │  │
-│  │  - discord:attachment:created                             │  │
-│  │  - discord:attachment:uploaded                            │  │
-│  │  - discord:analysis:queue_status                          │  │
-│  └───────────────────────────────────────────────────────────┘  │
-│                                │                                │
-└────────────────────────────────┼────────────────────────────────┘
-                                 │
-                                 │ Redis Pub/Sub
-                                 │
-                                 ▼
-┌─────────────────────────────────────────────────────────────────┐
-│                    Backend Service                              │
-│  (Subscribes to events, serves HTTP API, manages WebSocket)    │
-└─────────────────────────────────────────────────────────────────┘
-            │
-            │ HTTP API
-            │
-            ▼
-┌─────────────────────────────────────────────────────────────────┐
-│                    Frontend Application                         │
-│  (React SPA, real-time updates via WebSocket)                  │
-└─────────────────────────────────────────────────────────────────┘
-```
-
-## Summary
-
-The Discord Gateway service has been successfully extracted with:
-- **Modular MVC architecture** for clean separation of concerns
-- **Event-driven design** using Redis pub/sub for inter-service communication
-- **Shared infrastructure layer** for reusable components
-- **No HTTP server** — pure event-driven service
-- **Graceful shutdown** handling
-- **Type-safe configuration** with Zod validation
-- **Structured logging** with Winston
-- **PostgreSQL integration** with Drizzle ORM
-
-The service is ready for integration with the Backend service, which will consume Redis events and serve the HTTP API to the Frontend.

services/discord-gateway/src/app/bootstrap.ts

@@ -1,36 +1,19 @@
 import { Client } from "discord.js-selfbot-v13";
-import { ConfigError, DatabaseError } from "@/shared/errors/index";
-import { createChildLogger } from "@/shared/logger/index";
 import {
-  getAnalysisQueueStatus,
+  ConfigError,
-  startPendingAIAnalysisWorker,
+  DatabaseError,
-} from "../modules/ai-moderation/aiAnalyzer.js";
+  errorMessage,
-import {
+} from "@/shared/errors/index.js";
-  mediaWorkerPool,
+import { createChildLogger } from "@/shared/logger/index.js";
-  textWorkerPool,
-} from "../modules/ai-moderation/circuitBreaker.js";
-import { registerChannelTopicCapture } from "../modules/channel-topic/index.js";
 import { CommandHandler } from "../modules/command-handler/commandHandler.js";
 import {
   EventBroadcaster,
   RedisEventPublisher,
 } from "../modules/event-broadcaster/index.js";
 import {
-  registerCollector,
-  setGauge,
   startMetricsServer,
   stopMetricsServer,
 } from "../modules/gateway-metrics/index.js";
-import { registerGuildMemberEvents } from "../modules/guild-member-events/index.js";
-import {
-  registerMessageCapture,
-  setEventBroadcaster as setMessageCaptureEventBroadcaster,
-} from "../modules/message-capture/messageCapture.js";
-import { setModerationEventBroadcaster } from "../modules/message-capture/moderationActionsDb.js";
-import { startDigestScheduler } from "../modules/monitor/digestScheduler.js";
-import { registerReactionCapture } from "../modules/reaction-tracking/index.js";
-import { registerThreadCapture } from "../modules/thread-tracking/index.js";
-import { registerPresenceCapture } from "../modules/user-presence/index.js";
 import { config } from "../shared/config/index.js";
 import {
   closeDatabase,
@@ -38,45 +21,34 @@ import {
 } from "../shared/database/drizzle.js";
 import { runMigrations } from "../shared/database/migrate.js";
 import { createDiscordClientOptions } from "../shared/discord/clientOptions.js";
-import { startRetentionCleanup } from "./retention.js";
+import { startGatewayLifecycle } from "./lifecycle.js";
+import { registerPipelineMetrics } from "./metrics-collector.js";
+import { registerProcessGuards } from "./process-guards.js";
 import { createGracefulShutdown } from "./shutdown.js";
 
 const logger = createChildLogger("discord-gateway");
 
 // ─── Bootstrap ─────────────────────────────────────────────────────────────
+//
+// Startup order:
+//   1. validate config            (fail fast on missing AI credentials)
+//   2. connect infrastructure     (migrations → DB pool)
+//   3. build long-lived services  (Discord client, Redis publisher, command
+//                                  handler) + install shutdown/process guards
+//   4. start observability        (pipeline gauges → metrics server)
+//   5. log in                     (ready-hook wires listeners via lifecycle.ts)
 
-export async function initializeDiscordGateway() {
+/** Refuse to start when AI analysis is on but no LLM credentials exist. */
+function assertConfigIsUsable(): void {
   if (config.AI_ANALYSIS_ENABLED && !config.AI_LLM_API_KEY) {
     throw new ConfigError(
       "AI_ANALYSIS_ENABLED=true but AI_LLM_API_KEY is missing from environment. AI analysis cannot run without credentials.",
     );
   }
+}
 
-  const token = config.DISCORD_TOKEN;
+/** Run migrations (when enabled) then open the PostgreSQL pool. */
-  logger.info(
+async function connectDatabase(): Promise<void> {
-    { hasToken: token.length > 0, tokenLength: token.length },
-    "Config loaded",
-  );
-
-  logger.info("Creating Discord client");
-  const client = new Client(createDiscordClientOptions());
-
-  // Initialize Redis event broadcaster
-  const redisPublisher = new RedisEventPublisher(config.REDIS_URL, logger);
-  const eventBroadcaster = new EventBroadcaster(redisPublisher);
-
-  // Initialize Redis command handler for backend→gateway commands
-  const commandHandler = new CommandHandler();
-
-  const gracefulShutdown = createGracefulShutdown({
-    logger,
-    closeDatabase,
-    client,
-    eventBroadcaster,
-    commandHandler,
-    stopMetricsServer,
-  });
-
   try {
     if (config.AUTO_MIGRATE_ON_STARTUP) {
       logger.info(
@@ -90,183 +62,79 @@ export async function initializeDiscordGateway() {
     logger.info("PostgreSQL database initialized");
   } catch (err) {
     logger.error(
-      { err, errorMsg: err instanceof Error ? err.message : String(err) },
+      { err, errorMsg: errorMessage(err) },
       "Failed to initialize database",
     );
     throw new DatabaseError(
-      `Database initialization failed: ${err instanceof Error ? err.message : String(err)}`,
+      `Database initialization failed: ${errorMessage(err)}`,
     );
   }
+}
 
+/** Log only client debug lines that carry signal (errors/streams, or VERBOSE). */
+function registerClientDebugLogging(client: Client): void {
   client.on("debug", (msg) => {
-    if (
+    const lower = msg.toLowerCase();
-      msg.toLowerCase().includes("error") ||
+    if (lower.includes("error") || lower.includes("stream")) {
-      msg.toLowerCase().includes("stream")
-    ) {
       logger.info({ debugMsg: msg }, "Discord Client Debug");
     } else if (config.VERBOSE) {
       logger.debug({ debugMsg: msg }, "Discord Client Debug");
     }
   });
+}
 
-  client.on("ready", async () => {
+export async function initializeDiscordGateway() {
+  assertConfigIsUsable();
+
+  const token = config.DISCORD_TOKEN;
+  logger.info(
+    { hasToken: token.length > 0, tokenLength: token.length },
+    "Config loaded",
+  );
+
+  logger.info("Creating Discord client");
+  const client = new Client(createDiscordClientOptions());
+
+  // Long-lived services: Redis event broadcaster (gateway → backend) and the
+  // Redis command handler (backend → gateway).
+  const redisPublisher = new RedisEventPublisher(config.REDIS_URL, logger);
+  const eventBroadcaster = new EventBroadcaster(redisPublisher);
+  const commandHandler = new CommandHandler();
+
+  const gracefulShutdown = createGracefulShutdown({
+    logger,
+    closeDatabase,
+    client,
+    eventBroadcaster,
+    commandHandler,
+    stopMetricsServer,
+  });
+
+  await connectDatabase();
+
+  registerClientDebugLogging(client);
+
+  client.on("ready", () => {
     logger.info({ user: client.user?.tag }, "Bot logged in");
-    setMessageCaptureEventBroadcaster(eventBroadcaster);
+    startGatewayLifecycle({
-    setModerationEventBroadcaster(eventBroadcaster);
+      client,
-    registerMessageCapture(client);
+      eventBroadcaster,
-    startPendingAIAnalysisWorker(client, eventBroadcaster);
+      commandHandler,
-
+      logger,
-    // Register new event captures
+    });
-    registerReactionCapture(client, eventBroadcaster);
-    registerThreadCapture(client, eventBroadcaster);
-    registerPresenceCapture(client, eventBroadcaster);
-    registerChannelTopicCapture(client, eventBroadcaster);
-    registerGuildMemberEvents(client, eventBroadcaster);
-
-    // Start command handler after Discord is ready
-    commandHandler.start(client);
-    logger.info("Command handler started");
-
-    // Start retention cleanup scheduler
-    startRetentionCleanup();
-    // Start weekly moderation digest (public, automated)
-    startDigestScheduler();
   });
 
   client.on("error", (err) => {
-    logger.error(
+    logger.error({ err, errorMsg: errorMessage(err) }, "Client error");
-      { err, errorMsg: err instanceof Error ? err.message : String(err) },
-      "Client error",
-    );
   });
 
-  process.on("SIGINT", () => {
+  registerProcessGuards(logger, gracefulShutdown);
-    gracefulShutdown("SIGINT");
-  });
 
-  process.on("SIGTERM", () => {
+  // Metrics: register live pipeline collectors before starting the server.
-    gracefulShutdown("SIGTERM");
+  registerPipelineMetrics(logger);
-  });
-
-  process.on("uncaughtException", (err) => {
-    const code =
-      typeof (err as NodeJS.ErrnoException).code === "string"
-        ? (err as NodeJS.ErrnoException).code
-        : "";
-    // Transient stream-teardown errors (voice stop/disconnect races, child
-    // process stdin closed while we still write) are NOT fatal — crashing the
-    // gateway on EPIPE takes the whole bot offline mid-music. Log + continue.
-    if (
-      code === "EPIPE" ||
-      code === "ERR_STREAM_DESTROYED" ||
-      code === "ERR_STREAM_WRITE_AFTER_END" ||
-      code === "ECONNRESET"
-    ) {
-      logger.warn(
-        { error: err },
-        "Uncaught transient stream error — continuing",
-      );
-      return;
-    }
-    logger.error(
-      {
-        err,
-        errorMsg: err instanceof Error ? err.message : String(err),
-        stack: err?.stack,
-      },
-      "Uncaught exception",
-    );
-    gracefulShutdown("uncaughtException");
-  });
-
-  process.on("unhandledRejection", (reason) => {
-    const err =
-      reason instanceof Error ? reason : new Error(String(reason ?? "unknown"));
-    const code = (err as NodeJS.ErrnoException).code ?? "";
-    // Same transient-teardown policy as uncaughtException: a rejection that
-    // fires while a stream is being torn down (EPIPE after ffmpeg stdin
-    // closes, write-after-destroy, socket reset) must NOT take the whole
-    // gateway offline. Log detail + continue. Everything else still shuts
-    // down so real bugs surface.
-    if (
-      code === "EPIPE" ||
-      code === "ERR_STREAM_DESTROYED" ||
-      code === "ERR_STREAM_WRITE_AFTER_END" ||
-      code === "ECONNRESET"
-    ) {
-      logger.warn(
-        { error: err },
-        "Unhandled rejection transient stream error — continuing",
-      );
-      return;
-    }
-    logger.error({ error: err, reason: String(reason) }, "Unhandled rejection");
-    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_text_requests",
-        status.activeTextRequests ?? status.activeRequests,
-      );
-      setGauge(
-        "ai_analysis_active_media_requests",
-        status.activeMediaRequests ?? 0,
-      );
-      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);
-      }
-      type PoolState = { _poolState?: { size: number; active: number } };
-      const textPool = textWorkerPool as unknown as PoolState;
-      const mediaPool = mediaWorkerPool as unknown as PoolState;
-      // Reported per queue (2026-08-31 text/media pool split) so the text
-      // and media backlogs are distinguishable in dashboards/alerts instead
-      // of one combined "worker threads" number.
-      if (textPool._poolState) {
-        setGauge("ai_analysis_worker_threads_text", textPool._poolState.size);
-        setGauge(
-          "ai_analysis_worker_threads_active_text",
-          textPool._poolState.active,
-        );
-      }
-      if (mediaPool._poolState) {
-        setGauge("ai_analysis_worker_threads_media", mediaPool._poolState.size);
-        setGauge(
-          "ai_analysis_worker_threads_active_media",
-          mediaPool._poolState.active,
-        );
-      }
-    } catch (err) {
-      logger.warn({ error: String(err) }, "AI metrics collector failed");
-    }
-  });
-
-  // Start metrics server
   startMetricsServer();
 
   logger.info("Calling Discord client.login");
-
-  // Fix: use await + try/catch instead of .then().catch()
   try {
     await client.login(token);
     logger.info("Discord client logged in successfully");

services/discord-gateway/src/app/lifecycle.ts

@@ -0,0 +1,61 @@
+import type { Client } from "discord.js-selfbot-v13";
+import type { Logger } from "@/shared/logger/index.js";
+import { startPendingAIAnalysisWorker } from "../modules/ai-moderation/index.js";
+import { registerChannelTopicCapture } from "../modules/channel-topic/index.js";
+import type { CommandHandler } from "../modules/command-handler/commandHandler.js";
+import type { EventBroadcaster } from "../modules/event-broadcaster/index.js";
+import { registerGuildMemberEvents } from "../modules/guild-member-events/index.js";
+import {
+  registerMessageCapture,
+  setEventBroadcaster as setMessageCaptureEventBroadcaster,
+  setModerationEventBroadcaster,
+} from "../modules/message-capture/index.js";
+import { startDigestScheduler } from "../modules/monitor/digestScheduler.js";
+import { registerReactionCapture } from "../modules/reaction-tracking/index.js";
+import { registerThreadCapture } from "../modules/thread-tracking/index.js";
+import { registerPresenceCapture } from "../modules/user-presence/index.js";
+import { startRetentionCleanup } from "./retention.js";
+
+export interface GatewayLifecycleOptions {
+  client: Client;
+  eventBroadcaster: EventBroadcaster;
+  commandHandler: CommandHandler;
+  logger: Logger;
+}
+
+/**
+ * Wires everything that must start once Discord is connected.
+ *
+ * Ordering matters:
+ *  1. Inject the event broadcaster into the modules that publish events —
+ *     they must be able to publish before their listeners are registered.
+ *  2. Register the Discord event listeners (capture modules).
+ *  3. Start the background workers/schedulers.
+ */
+export function startGatewayLifecycle({
+  client,
+  eventBroadcaster,
+  commandHandler,
+  logger,
+}: GatewayLifecycleOptions): void {
+  // 1. Inject broadcaster first so no captured event is dropped.
+  setMessageCaptureEventBroadcaster(eventBroadcaster);
+  setModerationEventBroadcaster(eventBroadcaster);
+
+  // 2. Discord event listeners.
+  registerMessageCapture(client);
+  registerReactionCapture(client, eventBroadcaster);
+  registerThreadCapture(client, eventBroadcaster);
+  registerPresenceCapture(client, eventBroadcaster);
+  registerChannelTopicCapture(client, eventBroadcaster);
+  registerGuildMemberEvents(client, eventBroadcaster);
+
+  // 3. Background workers + schedulers.
+  startPendingAIAnalysisWorker(client, eventBroadcaster);
+  commandHandler.start(client);
+  logger.info("Command handler started");
+
+  startRetentionCleanup();
+  // Weekly moderation digest (public, automated)
+  startDigestScheduler();
+}

services/discord-gateway/src/app/metrics-collector.ts

@@ -0,0 +1,77 @@
+import type { Logger } from "@/shared/logger/index.js";
+import {
+  getAnalysisQueueStatus,
+  mediaWorkerPool,
+  textWorkerPool,
+} from "../modules/ai-moderation/index.js";
+import {
+  registerCollector,
+  setGauge,
+} from "../modules/gateway-metrics/index.js";
+import { config } from "../shared/config/index.js";
+
+/** Piscina exposes its live thread counters on `_poolState`. */
+type PoolState = { _poolState?: { size: number; active: number } };
+
+/**
+ * Registers the AI-pipeline Prometheus gauges.
+ *
+ * The collector refreshes on every scrape, so Prometheus sees real queue
+ * depth / concurrency / worker-thread state instead of an empty stub.
+ * Registered before the metrics server starts.
+ */
+export function registerPipelineMetrics(logger: Logger): void {
+  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_text_requests",
+        status.activeTextRequests ?? status.activeRequests,
+      );
+      setGauge(
+        "ai_analysis_active_media_requests",
+        status.activeMediaRequests ?? 0,
+      );
+      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);
+      }
+
+      // Reported per queue (2026-08-31 text/media pool split) so the text
+      // and media backlogs are distinguishable in dashboards/alerts instead
+      // of one combined "worker threads" number.
+      const textPool = textWorkerPool as unknown as PoolState;
+      const mediaPool = mediaWorkerPool as unknown as PoolState;
+      if (textPool._poolState) {
+        setGauge("ai_analysis_worker_threads_text", textPool._poolState.size);
+        setGauge(
+          "ai_analysis_worker_threads_active_text",
+          textPool._poolState.active,
+        );
+      }
+      if (mediaPool._poolState) {
+        setGauge("ai_analysis_worker_threads_media", mediaPool._poolState.size);
+        setGauge(
+          "ai_analysis_worker_threads_active_media",
+          mediaPool._poolState.active,
+        );
+      }
+    } catch (err) {
+      logger.warn({ error: String(err) }, "AI metrics collector failed");
+    }
+  });
+}

services/discord-gateway/src/app/process-guards.ts

@@ -0,0 +1,61 @@
+import { errorMessage, isTransientStreamError } from "@/shared/errors/index.js";
+import type { Logger } from "@/shared/logger/index.js";
+import type { GracefulShutdown } from "./shutdown.js";
+
+/**
+ * Process-level signal + error guards.
+ *
+ * Extracted from bootstrap so the "what keeps the gateway alive vs what
+ * shuts it down" policy lives in exactly one place.
+ *
+ * Policy: transient stream-teardown failures (EPIPE / ERR_STREAM_DESTROYED /
+ * ERR_STREAM_WRITE_AFTER_END / ECONNRESET) are logged and IGNORED — crashing
+ * the gateway on them (voice stop/disconnect races, a child process stdin
+ * closed while we still write) takes the whole bot offline mid-operation.
+ * Anything else is a real bug: log with stack and shut down cleanly.
+ */
+export function registerProcessGuards(
+  logger: Logger,
+  gracefulShutdown: GracefulShutdown,
+): void {
+  process.on("SIGINT", () => {
+    gracefulShutdown("SIGINT");
+  });
+
+  process.on("SIGTERM", () => {
+    gracefulShutdown("SIGTERM");
+  });
+
+  process.on("uncaughtException", (err) => {
+    if (isTransientStreamError(err)) {
+      logger.warn(
+        { error: err },
+        "Uncaught transient stream error — continuing",
+      );
+      return;
+    }
+    logger.error(
+      {
+        err,
+        errorMsg: errorMessage(err),
+        stack: err?.stack,
+      },
+      "Uncaught exception",
+    );
+    gracefulShutdown("uncaughtException");
+  });
+
+  process.on("unhandledRejection", (reason) => {
+    const err =
+      reason instanceof Error ? reason : new Error(String(reason ?? "unknown"));
+    if (isTransientStreamError(err)) {
+      logger.warn(
+        { error: err },
+        "Unhandled rejection transient stream error — continuing",
+      );
+      return;
+    }
+    logger.error({ error: err, reason: String(reason) }, "Unhandled rejection");
+    gracefulShutdown("unhandledRejection");
+  });
+}

services/discord-gateway/src/app/retention.ts

@@ -1,8 +1,5 @@
 import { inArray, lt } from "drizzle-orm";
-import type {
+import type { NodePgDatabase } from "drizzle-orm/node-postgres";
-  NodePgDatabase,
-  NodePgQueryResultHKT,
-} from "drizzle-orm/node-postgres";
 import { createChildLogger } from "@/shared/logger/index";
 import { config } from "../shared/config/index.js";
 import { getDatabase } from "../shared/database/drizzle.js";

services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts

@@ -12,28 +12,14 @@ import {
   buildSkipAnalysisUserResult,
   isAgeRestrictedMessage,
   isSkipAnalysisUser,
-  skipAgeRestrictedMessages,
-  skipAnalysisUserMessages,
 } from "./batchProcessor.js";
 import { scheduleConversationAnalysis } from "./batchScheduler.js";
 import { getConversationKey } from "./circuitBreaker.js";
-import {
+import { conversationDebounceTimers } from "./conversationState.js";
-  ANALYSIS_LANES,
-  type AnalysisLane,
-  clearConversationProcessing,
-  conversationConsecutiveErrors,
-  conversationDebounceTimers,
-  conversationErrorCooldown,
-  conversationProcessing,
-  isConversationProcessingLocked,
-} from "./conversationState.js";
 import {
   activeIndividualRequests,
-  enqueueIndividualFallbacks,
   individualCooldownUntil,
   individualInFlight,
-  individualInFlightByConversation,
-  individualInFlightLastTouched,
 } from "./individualFallbackProcessor.js";
 import {
   broadcastAnalysisCompleted,
@@ -41,31 +27,20 @@ import {
   setModerationClient,
   setSharedEventBroadcaster,
 } from "./moderationState.js";
-import { deleteExpiredQdrantPoints } from "./qdrantClient.js";
+import { startRecoveryWorker } from "./recovery-worker.js";
-import { pruneExpiredTexts } from "./textCacheStore.js";
 
 const logger = createChildLogger("ai-analyzer");
 
 // ---------------------------------------------------------------------------
-// Cache hygiene (expired verdict sweep)
+// Public API — queueing, status, worker startup
-// ---------------------------------------------------------------------------
-const CACHE_PRUNE_INTERVAL_MS = 6 * 60 * 60 * 1000; // every 6 hours
-let lastCachePruneAt = 0;
-
-// ---------------------------------------------------------------------------
-// Re-exports from sub-modules (preserving original public API)
-// ---------------------------------------------------------------------------
-
-export { pickBatchWithinBudget } from "./batchProcessor.js";
-export { getConversationKey } from "./circuitBreaker.js";
-export { onCircuitBreakerAlert } from "./conversationState.js";
-
-// ---------------------------------------------------------------------------
-// Public API
 // ---------------------------------------------------------------------------
 
 /**
  * Queues a message for analysis (debounced by conversation).
+ *
+ * Messages that never need an LLM call are short-circuited here and recorded
+ * with their skip verdict: age-restricted messages and configured skip-list
+ * users.
  */
 export async function queueMessageAnalysis(messageId: string): Promise<void> {
   if (!config.AI_ANALYSIS_ENABLED) return;
@@ -78,13 +53,7 @@ export async function queueMessageAnalysis(messageId: string): Promise<void> {
     }
 
     if (isAgeRestrictedMessage(message)) {
-      const updated = await messageStore.updateMessageAIAnalysis(
+      await recordSkip(message.id, buildAgeRestrictedSkipResult());
-        message.id,
-        buildAgeRestrictedSkipResult(),
-      );
-      if (updated) {
-        broadcastAnalysisCompleted(updated);
-      }
       logger.debug(
         { messageId },
         "Skipped AI analysis for age-restricted message",
@@ -93,13 +62,7 @@ export async function queueMessageAnalysis(messageId: string): Promise<void> {
     }
 
     if (isSkipAnalysisUser(message)) {
-      const updated = await messageStore.updateMessageAIAnalysis(
+      await recordSkip(message.id, buildSkipAnalysisUserResult());
-        message.id,
-        buildSkipAnalysisUserResult(),
-      );
-      if (updated) {
-        broadcastAnalysisCompleted(updated);
-      }
       logger.debug(
         { messageId, userId: message.user_id },
         "Skipped AI analysis for configured skip-list user",
@@ -119,6 +82,17 @@ export async function queueMessageAnalysis(messageId: string): Promise<void> {
   }
 }
 
+/** Persist a skip verdict and broadcast it so the dashboard reflects it. */
+async function recordSkip(
+  messageId: string,
+  result: Parameters<typeof messageStore.updateMessageAIAnalysis>[1],
+): Promise<void> {
+  const updated = await messageStore.updateMessageAIAnalysis(messageId, result);
+  if (updated) {
+    broadcastAnalysisCompleted(updated);
+  }
+}
+
 /**
  * Queues a conversation for analysis (debounced).
  */
@@ -144,11 +118,12 @@ export function getAnalysisQueueStatus(): AnalysisQueueStatus {
 }
 
 /**
- * Starts the periodic recovery worker.
+ * Starts the background workers behind the analysis pipeline:
+ *  - the recovery worker (stranded pending / incomplete messages + cache prune)
+ *  - the optional culture and user-profile learners.
  *
- * Now also recovers messages stuck in `error/analysis_incomplete`
+ * Also injects the Discord client and event broadcaster into the pipeline
- * state (not just `pending`), and skips conversations that already have
+ * state so downstream modules can act and publish.
- * individual fallback work in progress to avoid DB last-write-wins races.
  */
 export function startPendingAIAnalysisWorker(
   client?: Client,
@@ -167,151 +142,5 @@ export function startPendingAIAnalysisWorker(
       .catch(console.error);
   }
 
-  setInterval(() => {
+  startRecoveryWorker();
-    // [D] Periodic cache hygiene: purge expired moderation verdicts from
-    // Postgres and Qdrant. Expired entries are never reused (filters check
-    // expires_at) but accumulate forever without this sweep.
-    const now = Date.now();
-    if (now - lastCachePruneAt >= CACHE_PRUNE_INTERVAL_MS) {
-      lastCachePruneAt = now;
-      Promise.all([pruneExpiredTexts(), deleteExpiredQdrantPoints()])
-        .then(([pgDeleted, qdDeleted]) => {
-          if (pgDeleted > 0 || qdDeleted > 0) {
-            logger.info(
-              { pgDeleted, qdDeleted },
-              "Expired moderation cache pruned",
-            );
-          }
-        })
-        .catch((err: unknown) => {
-          logger.warn({ error: String(err) }, "Moderation cache prune failed");
-        });
-    }
-
-    // Only revert stuck processing messages if there's active processing.
-    // Avoids a DB query every recovery interval when the pipeline is idle.
-    if (conversationProcessing.size > 0) {
-      messageStore
-        .revertStuckProcessingMessages(300000)
-        .catch((err: unknown) => {
-          logger.error(
-            { error: String(err) },
-            "Failed to run stuck processing recovery",
-          );
-        });
-    }
-
-    Promise.all([
-      messageStore.getPendingConversationKeys(500),
-      messageStore.getConversationKeysWithIncompleteAnalysis(200),
-    ])
-      .then(([pendingKeys, incompleteKeys]) => {
-        const now = Date.now();
-
-        for (const [key, expiry] of conversationErrorCooldown) {
-          if (now >= expiry) conversationErrorCooldown.delete(key);
-        }
-        // conversationProcessing is now a Partial<Record<lane, startedAt>>.
-        // Prune stale lane slots individually so one stale lane never clears
-        // the other lane's healthy lock.
-        for (const [key, record] of conversationProcessing) {
-          for (const lane of ANALYSIS_LANES as readonly AnalysisLane[]) {
-            const startedAt = record?.[lane];
-            if (
-              startedAt &&
-              now - startedAt >= config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS
-            ) {
-              clearConversationProcessing(key, lane);
-            }
-          }
-        }
-
-        const staleThreshold = config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS * 2;
-        for (const [key, lastTouched] of individualInFlightLastTouched) {
-          if (now - lastTouched >= staleThreshold) {
-            individualInFlightLastTouched.delete(key);
-            individualInFlightByConversation.delete(key);
-            logger.warn(
-              { key },
-              "Pruned stale individualInFlightByConversation entry",
-            );
-          }
-        }
-
-        // Also prune stale per-conversation CB error counts that have cooled
-        // down so old conversations can be retried.
-        for (const [key] of conversationConsecutiveErrors) {
-          const cbExpire = conversationErrorCooldown.get(key) ?? 0;
-          if (cbExpire && now >= cbExpire) {
-            conversationConsecutiveErrors.delete(key);
-          }
-        }
-
-        const incompleteKeySet = new Set(incompleteKeys);
-
-        // --- Batch recovery for pending messages ---
-        for (const key of pendingKeys) {
-          if (
-            ANALYSIS_LANES.some((lane) =>
-              conversationDebounceTimers.has(`${key}::${lane}`),
-            )
-          ) {
-            continue;
-          }
-          // Batch recovery must not race ANY in-flight batch lane, so the
-          // lock check is lane-agnostic here (individual fallback handles
-          // error rows separately).
-          if (isConversationProcessingLocked(key)) continue;
-          if (individualInFlightByConversation.has(key)) continue;
-          if (incompleteKeySet.has(key)) continue;
-          const cooldownUntil = conversationErrorCooldown.get(key);
-          if (cooldownUntil && now < cooldownUntil) continue;
-          // No lane specified → schedule BOTH lanes; each fetches its own
-          // pending subset from the DB.
-          scheduleConversationAnalysis(key);
-        }
-
-        // --- Individual recovery for error/analysis_incomplete messages ---
-        // Circuit breaker check: no point iterating if individual CB is active.
-        if (now >= individualCooldownUntil) {
-          const promises: Promise<void>[] = [];
-          for (const key of incompleteKeys) {
-            // Skip if individual work is already running for this conversation.
-            if (individualInFlightByConversation.has(key)) continue;
-            // Skip if batch processing is running.
-            if (isConversationProcessingLocked(key)) continue;
-
-            promises.push(
-              messageStore
-                .getIncompleteMessagesByConversation(key, 500)
-                .then(async (msgs) => {
-                  const processableMessages = await skipAnalysisUserMessages(
-                    await skipAgeRestrictedMessages(msgs),
-                  );
-                  return processableMessages;
-                })
-                .then((msgs) => {
-                  if (msgs.length > 0) {
-                    enqueueIndividualFallbacks(msgs);
-                  }
-                })
-                .catch((err: unknown) => {
-                  logger.error(
-                    { key, error: String(err) },
-                    "Failed to fetch incomplete messages for recovery",
-                  );
-                }),
-            );
-          }
-          // Errors are handled per-key; return the combined promise for observability.
-          return Promise.all(promises);
-        }
-      })
-      .catch((err: unknown) => {
-        logger.error(
-          { error: err instanceof Error ? err.message : String(err) },
-          "Pending AI analysis recovery worker failed",
-        );
-      });
-  }, config.AI_ANALYSIS_RECOVERY_INTERVAL_MS);
 }

services/discord-gateway/src/modules/ai-moderation/cache-prune.ts

@@ -0,0 +1,41 @@
+import { createChildLogger } from "@/shared/logger/index";
+import { config } from "../../shared/config/index.js";
+import { deleteExpiredQdrantPoints } from "./qdrantClient.js";
+import { pruneExpiredTexts } from "./textCacheStore.js";
+
+const logger = createChildLogger("cache-prune");
+
+/** Expired-verdict sweep cadence. */
+const CACHE_PRUNE_INTERVAL_MS = 6 * 60 * 60 * 1000; // every 6 hours
+
+let lastCachePruneAt = 0;
+
+/**
+ * Cache hygiene: purge expired moderation verdicts from Postgres and Qdrant.
+ *
+ * Expired entries are never reused (read filters check `expires_at`) but they
+ * accumulate forever without a sweep. Called from the recovery interval; the
+ * 6-hour throttle keeps it to one sweep per window.
+ */
+export function runCachePruneIfDue(now: number = Date.now()): void {
+  if (now - lastCachePruneAt < CACHE_PRUNE_INTERVAL_MS) return;
+  lastCachePruneAt = now;
+
+  Promise.all([pruneExpiredTexts(), deleteExpiredQdrantPoints()])
+    .then(([pgDeleted, qdDeleted]) => {
+      if (pgDeleted > 0 || qdDeleted > 0) {
+        logger.info(
+          { pgDeleted, qdDeleted },
+          "Expired moderation cache pruned",
+        );
+      }
+    })
+    .catch((err: unknown) => {
+      logger.warn({ error: String(err) }, "Moderation cache prune failed");
+    });
+}
+
+/** Reset the throttle window (tests). */
+export function resetCachePruneState(): void {
+  lastCachePruneAt = 0;
+}

services/discord-gateway/src/modules/ai-moderation/index.ts

@@ -0,0 +1,35 @@
+/**
+ * Public surface of the AI-moderation module.
+ *
+ * The module has ~50 internal files; callers outside it (app/, tests, other
+ * modules) should import from THIS barrel so internal files can be moved
+ * without touching call sites.
+ *
+ * Deep imports remain valid inside the module itself.
+ */
+
+export type {
+  AIRecommendedAction,
+  AISeverity,
+  AIStatus,
+  AnalysisQueueStatus,
+  AnalysisResult,
+} from "../../shared/moderation-types.js";
+// ── Entry API: queueing, status, recovery worker ──────────────────────────
+export {
+  getAnalysisQueueStatus,
+  queueConversationAnalysis,
+  queueMessageAnalysis,
+  startPendingAIAnalysisWorker,
+} from "./aiAnalyzer.js";
+// ── Worker pools (app/metrics-collector reads their live thread counters) ──
+export {
+  getConversationKey,
+  mediaWorkerPool,
+  textWorkerPool,
+} from "./circuitBreaker.js";
+// ── Pipeline state hooks the bootstrap injects into ───────────────────────
+export {
+  setModerationClient,
+  setSharedEventBroadcaster,
+} from "./moderationState.js";

services/discord-gateway/src/modules/ai-moderation/recovery-worker.ts

@@ -0,0 +1,184 @@
+import { createChildLogger } from "@/shared/logger/index";
+import { config } from "../../shared/config/index.js";
+import { messageStore } from "../message-capture/messageStore.js";
+import {
+  skipAgeRestrictedMessages,
+  skipAnalysisUserMessages,
+} from "./batchProcessor.js";
+import { scheduleConversationAnalysis } from "./batchScheduler.js";
+import { runCachePruneIfDue } from "./cache-prune.js";
+import {
+  ANALYSIS_LANES,
+  type AnalysisLane,
+  clearConversationProcessing,
+  conversationConsecutiveErrors,
+  conversationDebounceTimers,
+  conversationErrorCooldown,
+  conversationProcessing,
+  isConversationProcessingLocked,
+} from "./conversationState.js";
+import {
+  enqueueIndividualFallbacks,
+  individualCooldownUntil,
+  individualInFlightByConversation,
+  individualInFlightLastTouched,
+} from "./individualFallbackProcessor.js";
+
+const logger = createChildLogger("ai-recovery");
+
+/** Revert messages stuck in `processing` for longer than this. */
+const STUCK_PROCESSING_AGE_MS = 300_000;
+
+/**
+ * Starts the periodic recovery worker.
+ *
+ * Recovers two classes of stranded work:
+ *  - `pending` messages → re-scheduled through the normal per-lane debounce.
+ *  - `error` / `analysis_incomplete` messages → individual fallback queue.
+ *
+ * Also prunes stale in-memory bookkeeping (lane locks, per-conversation
+ * circuit-breaker counters, individual-fallback in-flight markers) so a
+ * crashed batch cannot wedge a conversation forever, and triggers the
+ * throttled cache-prune sweep.
+ *
+ * Skips conversations that already have individual fallback work in progress
+ * to avoid DB last-write-wins races.
+ */
+export function startRecoveryWorker(): void {
+  setInterval(() => {
+    runCachePruneIfDue();
+
+    // Only revert stuck processing messages if there's active processing.
+    // Avoids a DB query every recovery interval when the pipeline is idle.
+    if (conversationProcessing.size > 0) {
+      messageStore
+        .revertStuckProcessingMessages(STUCK_PROCESSING_AGE_MS)
+        .catch((err: unknown) => {
+          logger.error(
+            { error: String(err) },
+            "Failed to run stuck processing recovery",
+          );
+        });
+    }
+
+    Promise.all([
+      messageStore.getPendingConversationKeys(500),
+      messageStore.getConversationKeysWithIncompleteAnalysis(200),
+    ])
+      .then(([pendingKeys, incompleteKeys]) => {
+        const now = Date.now();
+
+        pruneStaleConversationState(now);
+
+        const incompleteKeySet = new Set(incompleteKeys);
+
+        // --- Batch recovery for pending messages ---
+        for (const key of pendingKeys) {
+          if (
+            ANALYSIS_LANES.some((lane) =>
+              conversationDebounceTimers.has(`${key}::${lane}`),
+            )
+          ) {
+            continue;
+          }
+          // Batch recovery must not race ANY in-flight batch lane, so the
+          // lock check is lane-agnostic here (individual fallback handles
+          // error rows separately).
+          if (isConversationProcessingLocked(key)) continue;
+          if (individualInFlightByConversation.has(key)) continue;
+          if (incompleteKeySet.has(key)) continue;
+          const cooldownUntil = conversationErrorCooldown.get(key);
+          if (cooldownUntil && now < cooldownUntil) continue;
+          // No lane specified → schedule BOTH lanes; each fetches its own
+          // pending subset from the DB.
+          scheduleConversationAnalysis(key);
+        }
+
+        // --- Individual recovery for error/analysis_incomplete messages ---
+        // Circuit breaker check: no point iterating if individual CB is active.
+        if (now >= individualCooldownUntil) {
+          const promises: Promise<void>[] = [];
+          for (const key of incompleteKeys) {
+            // Skip if individual work is already running for this conversation.
+            if (individualInFlightByConversation.has(key)) continue;
+            // Skip if batch processing is running.
+            if (isConversationProcessingLocked(key)) continue;
+
+            promises.push(
+              recoverIncompleteConversation(key).catch((err: unknown) => {
+                logger.error(
+                  { key, error: String(err) },
+                  "Failed to fetch incomplete messages for recovery",
+                );
+              }),
+            );
+          }
+          // Errors are handled per-key; return the combined promise for observability.
+          return Promise.all(promises);
+        }
+      })
+      .catch((err: unknown) => {
+        logger.error(
+          { error: err instanceof Error ? err.message : String(err) },
+          "Pending AI analysis recovery worker failed",
+        );
+      });
+  }, config.AI_ANALYSIS_RECOVERY_INTERVAL_MS);
+}
+
+/** Fetch one conversation's incomplete messages and queue them individually. */
+async function recoverIncompleteConversation(key: string): Promise<void> {
+  const msgs = await messageStore.getIncompleteMessagesByConversation(key, 500);
+  const processable = await skipAnalysisUserMessages(
+    await skipAgeRestrictedMessages(msgs),
+  );
+  if (processable.length > 0) {
+    enqueueIndividualFallbacks(processable);
+  }
+}
+
+/**
+ * Drop stale in-memory bookkeeping:
+ *  - per-lane processing locks past the timeout (pruned PER LANE so one stale
+ *    lane never clears the other lane's healthy lock),
+ *  - individual-fallback in-flight markers that stopped being touched,
+ *  - per-conversation circuit-breaker error counts whose cooldown has lapsed.
+ */
+function pruneStaleConversationState(now: number): void {
+  for (const [key, expiry] of conversationErrorCooldown) {
+    if (now >= expiry) conversationErrorCooldown.delete(key);
+  }
+
+  for (const [key, record] of conversationProcessing) {
+    for (const lane of ANALYSIS_LANES as readonly AnalysisLane[]) {
+      const startedAt = record?.[lane];
+      if (
+        startedAt &&
+        now - startedAt >= config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS
+      ) {
+        clearConversationProcessing(key, lane);
+      }
+    }
+  }
+
+  const staleThreshold = config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS * 2;
+  for (const [key, lastTouched] of individualInFlightLastTouched) {
+    if (now - lastTouched >= staleThreshold) {
+      individualInFlightLastTouched.delete(key);
+      individualInFlightByConversation.delete(key);
+      logger.warn(
+        { key },
+        "Pruned stale individualInFlightByConversation entry",
+      );
+    }
+  }
+
+  // Also prune stale per-conversation CB error counts that have cooled
+  // down so old conversations can be retried.
+  for (const [key] of conversationConsecutiveErrors) {
+    const cbExpire = conversationErrorCooldown.get(key) ?? 0;
+    if (cbExpire && now >= cbExpire) {
+      conversationConsecutiveErrors.delete(key);
+    }
+  }
+}

services/discord-gateway/src/modules/command-handler/commandHandler.ts

@@ -1,12 +1,12 @@
 import type { Client } from "discord.js-selfbot-v13";
 import Redis from "ioredis";
 import { config } from "../../shared/config/index.js";
+import { createChildLogger } from "../../shared/logger/index.js";
 import {
   BACKEND_COMMAND,
   type CommandMessage,
   type CommandReply,
-} from "../../shared/index.js";
+} from "../../shared/redis-channels.js";
-import { createChildLogger } from "../../shared/logger/index.js";
 import { GuildHandler } from "./guild.handler.js";
 import {
   type CommandHandlerFn,

services/discord-gateway/src/modules/command-handler/guild.handler.ts

@@ -1,6 +1,9 @@
 import type { Client } from "discord.js-selfbot-v13";
-import type { CommandMessage, CommandReply } from "../../shared/index.js";
 import { createChildLogger } from "../../shared/logger/index.js";
+import type {
+  CommandMessage,
+  CommandReply,
+} from "../../shared/redis-channels.js";
 
 // ---------------------------------------------------------------------------
 // GuildHandler

services/discord-gateway/src/modules/command-handler/handler-registry.ts

@@ -4,7 +4,7 @@ import {
   COMMAND_MODERATION_ACTION,
   type CommandMessage,
   type CommandReply,
-} from "../../shared/index.js";
+} from "../../shared/redis-channels.js";
 import type { GuildHandler } from "./guild.handler.js";
 import type { ModerationHandler } from "./moderation.handler.js";
 

services/discord-gateway/src/modules/command-handler/moderation.handler.ts

@@ -1,6 +1,9 @@
 import type { Client } from "discord.js-selfbot-v13";
-import type { CommandMessage, CommandReply } from "../../shared/index.js";
 import { createChildLogger } from "../../shared/logger/index.js";
+import type {
+  CommandMessage,
+  CommandReply,
+} from "../../shared/redis-channels.js";
 import { messageStore } from "../message-capture/messageStore.js";
 
 // ---------------------------------------------------------------------------

services/discord-gateway/src/modules/event-broadcaster/eventBroadcaster.ts

@@ -1,9 +1,12 @@
 import Redis from "ioredis";
-import type { AttachmentRecord, MessageRecord } from "../../shared/index.js";
 import {
   type CustomLogger,
   createChildLogger,
 } from "../../shared/logger/index.js";
+import type {
+  AttachmentRecord,
+  MessageRecord,
+} from "../../shared/moderation-types.js";
 import { type DiscordGatewayEvent, EventChannels } from "./eventTypes.js";
 
 export class RedisEventPublisher {

services/discord-gateway/src/modules/event-broadcaster/eventTypes.ts

@@ -17,7 +17,7 @@ import {
   DISCORD_THREAD_DELETED,
   DISCORD_THREAD_UPDATED,
   type DiscordGatewayEvent,
-} from "../../shared/index.js";
+} from "../../shared/redis-channels.js";
 
 export type { DiscordGatewayEvent };
 

services/discord-gateway/src/modules/message-capture/index.ts

@@ -0,0 +1,41 @@
+/**
+ * Public surface of the message-capture module.
+ *
+ * Callers outside this module import from here instead of reaching into
+ * messageCapture.ts / moderationActionsDb.ts / messageStore.ts directly, so
+ * the internal file layout can change without touching call sites.
+ *
+ * Deep imports remain valid inside the module itself.
+ */
+
+// ── Discord listener registration (called from app/lifecycle.ts) ───────────
+export {
+  registerMessageCapture,
+  setEventBroadcaster,
+} from "./messageCapture.js";
+// ── Message persistence facade ────────────────────────────────────────────
+export { messageStore } from "./messageStore.js";
+// ── Live moderation-action publishing ─────────────────────────────────────
+export { setModerationEventBroadcaster } from "./moderationActionsDb.js";
+
+// ── Domain types ──────────────────────────────────────────────────────────
+export type {
+  AIRecommendedAction,
+  AISeverity,
+  AIStatus,
+  AnalysisQueueStatus,
+  AnalysisResult,
+  AttachmentRecord,
+  DashboardMessage,
+  MessageQuery,
+  MessageRecord,
+  MessageReview,
+  ModerationAction,
+  ModerationActionType,
+  ModerationWsEvent,
+  PageResult,
+  RetentionPolicy,
+  ReviewStatus,
+  RoleMetadata,
+  UserMetadata,
+} from "./types.js";

services/discord-gateway/src/modules/message-capture/messagesPagination.ts

@@ -2,8 +2,11 @@ import { and, desc, eq, inArray, type SQL, sql } from "drizzle-orm";
 import type { NodePgDatabase } from "drizzle-orm/node-postgres";
 import type * as schema from "../../shared/database/schema.js";
 import { messagesTable } from "../../shared/database/schema.js";
-import { buildCursorCondition, pageResult } from "../../shared/index.js";
 import { createChildLogger, type Logger } from "../../shared/logger/index.js";
+import {
+  buildCursorCondition,
+  pageResult,
+} from "../../shared/utils/pagination.js";
 import type {
   MessageQuery,
   MessageRecord,

services/discord-gateway/src/modules/message-capture/moderationActionsDb.ts

@@ -2,8 +2,11 @@ import { and, desc, eq, inArray, type SQL } from "drizzle-orm";
 import type { NodePgDatabase } from "drizzle-orm/node-postgres";
 import type * as schema from "../../shared/database/schema.js";
 import { moderationActionsTable } from "../../shared/database/schema.js";
-import { buildCursorCondition, pageResult } from "../../shared/index.js";
 import { createChildLogger, type Logger } from "../../shared/logger/index.js";
+import {
+  buildCursorCondition,
+  pageResult,
+} from "../../shared/utils/pagination.js";
 import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js";
 import type { ModerationAction, PageResult } from "../message-capture/types.js";
 

services/discord-gateway/src/modules/message-capture/reviewsDb.ts

@@ -2,8 +2,11 @@ import { and, desc, eq, inArray, type SQL } from "drizzle-orm";
 import type { NodePgDatabase } from "drizzle-orm/node-postgres";
 import type * as schema from "../../shared/database/schema.js";
 import { messageReviewsTable } from "../../shared/database/schema.js";
-import { buildCursorCondition, pageResult } from "../../shared/index.js";
 import { createChildLogger, type Logger } from "../../shared/logger/index.js";
+import {
+  buildCursorCondition,
+  pageResult,
+} from "../../shared/utils/pagination.js";
 import type { MessageReview, PageResult } from "../message-capture/types.js";
 
 // ─── ReviewsDb Class ────────────────────────────────────────────────────────

services/discord-gateway/src/modules/message-capture/types.ts

@@ -2,8 +2,7 @@ import type {
   AnalysisQueueStatus,
   AttachmentRecord,
   MessageRecord,
-  UserMetadata,
+} from "../../shared/moderation-types.js";
-} from "../../shared/index.js";
 
 // Re-export all shared types for backward compatibility
 export type {
@@ -26,7 +25,7 @@ export type {
   ReviewStatus,
   RoleMetadata,
   UserMetadata,
-} from "../../shared/index.js";
+} from "../../shared/moderation-types.js";
 
 // Local-only types (not shared across services)
 export type ModerationWsEvent =

services/discord-gateway/src/shared/errors/index.ts

@@ -46,3 +46,32 @@ export class ConfigError extends AppError {
     this.name = "ConfigError";
   }
 }
+
+// ---------------------------------------------------------------------------
+// Generic error helpers (shared by every module — avoids the repeated
+// `err instanceof Error ? err.message : String(err)` pattern, 18+ sites)
+// ---------------------------------------------------------------------------
+
+/** Normalize an unknown thrown value to a readable message. */
+export function errorMessage(err: unknown): string {
+  return err instanceof Error ? err.message : String(err);
+}
+
+/**
+ * True when an error code is a transient stream-teardown failure
+ * (EPIPE / stream destroyed / write-after-end / socket reset).
+ *
+ * These are NOT fatal: crashing the gateway on them (e.g. voice stop races,
+ * ffmpeg stdin closed while we still write) takes the whole bot offline
+ * mid-operation. Callers that install process-level handlers use this to
+ * log-and-continue instead of shutting down.
+ */
+export function isTransientStreamError(err: unknown): boolean {
+  const code = (err as NodeJS.ErrnoException)?.code ?? "";
+  return (
+    code === "EPIPE" ||
+    code === "ERR_STREAM_DESTROYED" ||
+    code === "ERR_STREAM_WRITE_AFTER_END" ||
+    code === "ECONNRESET"
+  );
+}

services/discord-gateway/src/shared/index.ts

@@ -1,9 +0,0 @@
-export * from "./config/index.js";
-export * from "./database/init.js";
-export * from "./database/pool.js";
-export * from "./database/schema.js";
-export * from "./errors/index.js";
-export * from "./logger/index.js";
-export * from "./moderation-types.js";
-export * from "./redis-channels.js";
-export * from "./utils/index.js";