# Real-Time BACEN Message Monitoring - Design Document **Date**: 2026-02-07 **Status**: Approved **Author**: Monetarie Engineering --- ## Overview Real-time monitoring dashboard for BACEN/SPB message traffic. Bloomberg-terminal style single-screen interface with live message feed, aggregate statistics, alert management, and full filtering. Covers 4 use cases: operations dashboard, alert-focused monitoring, traffic observatory, and compliance/audit stream. ## Architecture ### Data Flow ``` NATS JetStream → BacenGateway.Monitoring.Broadcaster (GenServer) → Phoenix.PubSub (topic: "monitoring") → MonitoringChannel (per-client server-side filtering) → WebSocket → Vue/Pinia monitoringStore → UI Components ``` ### Stage 1: NATS to Phoenix PubSub The `Broadcaster` GenServer subscribes to NATS topics (`spb.messages.>`) via the existing `:bacen_gateway_nats` connection. Every message event (created, state_changed, failed, confirmed) is broadcast to `Phoenix.PubSub` on the `"monitoring"` topic. The Broadcaster also maintains: - Rolling stats counter (messages/sec, state distribution) broadcast every 1 second - Message deduplication by `message_id` to handle NATS at-least-once delivery - Rate limiting: if >100 msgs/sec, batch into 100ms windows ### Stage 2: Phoenix Channel `MonitoringChannel` joins on `"monitoring:live"`. On join: 1. Authenticates via JWT token (URL param) 2. Sends snapshot: last 100 messages from DB, current stats, pending alerts 3. Stores per-client filter state in socket assigns On `"set_filters"` event from client: - Updates filter state in socket assigns - Only matching events are pushed to that client On PubSub broadcast: - Applies client's filters server-side - Drops non-matching messages silently ### Stage 3: Vue WebSocket Client Frontend connects via `monitoringSocket.ts` wrapper. Events dispatched to Pinia `monitoringStore`. Vue reactivity renders updates. ## UI Layout ### Single Unified Dashboard (Full Viewport) ``` +------------------------------------------------------------------+ | [*] Connected | Filters: STR, LPI | [!3] Alerts [Sound ON] | <- Top Bar (64px) +------------------------------------------------------------------+ | [== Filter Bar (collapsible) ==================================] | | Type: [STR v] Direction: [All v] State: [All v] ISPB: [____] | +------------------------------------------------------------------+ | | | | | STATS PANEL | LIVE MESSAGE FEED | ALERT PANEL | | (30% width) | (50% width) | (20% width) | | | | | | Today: 40,000 | 22:45:01 STR0001 -> ITAU | [!] CRITICAL | | Sent: 26,668 | 22:45:00 LPI0003 <- BCB | 3 STR failures | | Confirmed: 6,667 | 22:44:59 PAG0101 -> BRAD | in 2 minutes | | Failed: 6,666 | 22:44:58 SEL0002 <- SELIC | [Acknowledge] | | | 22:44:57 STR0004 -> NUBANK| | | [msgs/sec chart] | 22:44:56 CAM0001 <- BCB | [!] WARNING | | | ... | LPI queue depth | | [state donut] | | > 500 | | | [Click row for details] | [Acknowledge] | | Top Institutions: | | | | 1. MONETARIE 12,340 | | [Ack All] | | 2. ITAU 3,201 | | | | 3. BRADESCO 2,890 | | | | | | | | Clearings: | | | | STR [GREEN] | | | | LPI [GREEN] | | | | CIP [YELLOW] | | | | SELIC [GREEN] | | | +------------------------------------------------------------------+ ``` ### Filter Bar - **Message Type**: Multi-select dropdown (STR, LPI, GEN, PAG, CAM, SEL, LDL, DDA, CIR, CTP) - **Direction**: All / Inbound / Outbound - **State**: All / Pending / Sent / Received / Acknowledged / Confirmed / Failed - **Institution**: Searchable dropdown by ISPB or short name - **Priority**: All / 1 (High) / 2 (Medium) / 3 (Low) - **Search**: Text search for message ID (MSG0000000001) - **Clear Filters**: Reset all to default ### Message Detail Drawer Slide-over panel (right side, 500px) when clicking a message row: - Message metadata: ID, type, category, direction, institution, priority, timestamps - State history timeline (created → sent → confirmed, with timestamps) - Full XML content (syntax highlighted, collapsible) - JSONB content parsed view - Retry button (for failed messages, triggers re-processing via API) ## Backend Implementation ### New Files #### `lib/bacen_gateway_web/channels/monitoring_socket.ex` WebSocket endpoint. Authenticates via JWT token passed as URL query param. ```elixir defmodule BacenGatewayWeb.MonitoringSocket do use Phoenix.Socket channel "monitoring:*", BacenGatewayWeb.MonitoringChannel @impl true def connect(%{"token" => token}, socket, _connect_info) do case BacenGateway.Auth.JWT.verify_token(token) do {:ok, claims} -> {:ok, assign(socket, :user_id, claims["sub"])} {:error, _} -> :error end end def connect(_params, _socket, _connect_info), do: :error @impl true def id(socket), do: "monitoring:#{socket.assigns.user_id}" end ``` #### `lib/bacen_gateway_web/channels/monitoring_channel.ex` Channel with per-client filtering. Sends snapshot on join. Filters events server-side. Key behaviors: - `join("monitoring:live", _, socket)` → snapshot + subscribe - `handle_in("set_filters", filters, socket)` → update socket assigns - `handle_info({:monitoring_event, event}, socket)` → filter + push or drop - `handle_info({:stats_update, stats}, socket)` → always push (no filter) - `handle_info({:alert, alert}, socket)` → always push #### `lib/bacen_gateway/monitoring/broadcaster.ex` GenServer that bridges NATS → PubSub. - Subscribes to `spb.messages.>` on `:bacen_gateway_nats` - Maintains ETS table for deduplication (message_id → timestamp, TTL 60s) - Rolling stats: counters per state, messages/sec (sliding 60s window) - Broadcasts `{:monitoring_event, event}` to PubSub - Broadcasts `{:stats_update, stats}` every 1 second - Rate limiting: buffers messages if >100/sec, flushes every 100ms #### `lib/bacen_gateway/monitoring/alert_manager.ex` GenServer for alert detection. Alert conditions: - Message failure (any message transitions to "failed") - Consecutive failures (>3 failures from same institution in 5 minutes) - Stuck messages (state "pending" for >5 minutes) - Queue depth warning (IBM MQ queue >1000 messages) Storage: ETS table `{alert_id, severity, message, timestamp, acknowledged}` API: - `AlertManager.get_active_alerts/0` → list of unacknowledged alerts - `AlertManager.acknowledge/1` → mark alert as acknowledged - `AlertManager.acknowledge_all/0` → clear all alerts ### Modified Files #### `lib/bacen_gateway_web/endpoint.ex` Add socket mount: ```elixir socket "/monitoring/websocket", BacenGatewayWeb.MonitoringSocket, websocket: [timeout: 45_000], longpoll: false ``` #### `lib/bacen_gateway/application.ex` Add to children list: ```elixir {BacenGateway.Monitoring.Broadcaster, []}, {BacenGateway.Monitoring.AlertManager, []}, ``` #### `router.ex` No changes needed — WebSocket is mounted via endpoint, not router. ## Frontend Implementation ### New Files #### `src/services/monitoringSocket.ts` WebSocket client wrapping Phoenix Channel protocol. ```typescript interface MonitoringSocket { connect(token: string): void disconnect(): void setFilters(filters: MonitoringFilters): void acknowledgeAlert(alertId: string): void onMessage(callback: (msg: BacenMessage) => void): void onStats(callback: (stats: MonitoringStats) => void): void onAlert(callback: (alert: Alert) => void): void onConnectionChange(callback: (status: ConnectionStatus) => void): void } ``` Auto-reconnect with exponential backoff: 1s, 2s, 4s, 8s, max 30s. #### `src/stores/monitoring.ts` Pinia store: ```typescript interface MonitoringState { messages: BacenMessage[] // capped at 500 stats: MonitoringStats alerts: Alert[] filters: MonitoringFilters connectionStatus: 'connected' | 'reconnecting' | 'disconnected' soundEnabled: boolean // persisted to localStorage } ``` #### `src/views/MonitoringDashboardView.vue` Main page. Connects on mount, disconnects on unmount. Full-height 4-panel layout. #### `src/components/monitoring/LiveMessageFeed.vue` Virtualized table using Vuetify `v-virtual-scroll`. Columns: timestamp, message_id, type, direction, institution, state, priority. Row click opens detail drawer. New messages animate in from top. Failed messages have red left border. #### `src/components/monitoring/StatsPanel.vue` Left panel. Counters with Vuetify `v-sparkline` for trends. Donut chart for state distribution. Top institutions list. Clearing status indicators. #### `src/components/monitoring/AlertPanel.vue` Right panel. Alert list sorted by severity → timestamp. Acknowledge button per alert. "Acknowledge All" button. Audio beep on new critical alert (`new Audio('/alert.mp3').play()` when `soundEnabled`). #### `src/components/monitoring/FilterBar.vue` Collapsible bar. Vuetify `v-select` for each filter dimension. `v-text-field` for message ID search. "Clear Filters" button. Emits to store on change. #### `src/components/monitoring/MessageDetailDrawer.vue` Vuetify `v-navigation-drawer` (right, temporary, 500px). Message metadata, state timeline, XML content with `
` block, retry button for failed messages.

### Modified Files

#### `src/router/index.ts`

Add route:

```typescript
{
  path: '/monitoring',
  name: 'monitoring',
  component: () => import('@/views/MonitoringDashboardView.vue'),
  meta: { requiresAuth: true, title: 'BACEN Monitoring' }
}
```

## Error Handling & Resilience

| Scenario | Behavior |
|----------|----------|
| WebSocket disconnect | Auto-reconnect (1s→30s backoff). Yellow "Reconnecting" indicator. Fresh snapshot on reconnect. |
| Backend crash | Supervisor restarts Broadcaster/AlertManager. Re-subscribes to NATS. Stats reset and rebuild. |
| NATS disconnect | Gnat.ConnectionSupervisor reconnects automatically. Feed freezes during outage. JetStream replays missed messages. Broadcaster deduplicates. |
| High volume (>100 msgs/sec) | Broadcaster batches into 100ms windows. Server-side filtering reduces per-client load. Client buffer capped at 500 items. |
| JWT expiry | Channel sends `"token_expired"` event. Vue redirects to login. |
| Browser memory | Message buffer capped at 500. Old messages dropped. Stats are numeric counters only. |

## Build Sequence

1. Backend: `monitoring_socket.ex` + `monitoring_channel.ex` + `broadcaster.ex` + `alert_manager.ex`
2. Backend: Modify `endpoint.ex` + `application.ex`
3. Backend: Test with `wscat` or Postman WebSocket
4. Frontend: `monitoringSocket.ts` + `monitoring.ts` (store)
5. Frontend: Components (StatsPanel, LiveMessageFeed, AlertPanel, FilterBar, MessageDetailDrawer)
6. Frontend: `MonitoringDashboardView.vue` + router
7. E2E: Connect frontend to backend, verify real-time flow
8. Deploy: Cloud Build both services, verify in GKE

## Dependencies

**Zero new dependencies.** Everything uses existing stack:
- Phoenix Channels (already in Phoenix)
- Phoenix.PubSub (already running)
- Gnat (already connected to NATS)
- Vuetify v-virtual-scroll, v-sparkline, v-navigation-drawer
- Pinia (already in use)

---

*Design approved 2026-02-07. Implementation target: TBD.*