Nexus-Guard-Suite/.sisyphus/plans/realtime-traffic-monitoring.md

# Real-Time Traffic Monitoring + gRPC + TimescaleDB

## TL;DR
> Real-time device/node status via SSE + gRPC, traffic monitoring with TimescaleDB, historical charts with daily aggregation, toggle controls for chart display.

**Deliverables**:
- gRPC streaming for device-agent ↔ server (Rx/Tx data)
- SSE endpoint for dashboard real-time updates
- TimescaleDB schema for traffic logging
- Traffic recorder (Redis → DB batch)
- Dashboard traffic chart with historical data
- Toggle to disable real-time display (per device/global)

**Estimated Effort**: Large
**Parallel Execution**: YES - 4 waves
**Critical Path**: T1 → T2 → T3 → T4 → T5 → T6

---

## Context

### Original Request
User wants real-time device/node online status without page refresh, Rx/Tx traffic with charts, daily/historical logging, and toggle controls for chart display. System scales to 1000+ devices.

### Architecture Decision
- **Device-Agent → Server**: gRPC streaming (bidirectional, efficient for 1000+ connections)
- **Dashboard ← Server**: SSE (browser native, HTTP friendly, auto-reconnect)
- **Real-time state**: Redis (fast in-memory, pub/sub)
- **Traffic recording**: TimescaleDB (PostgreSQL extension, time-series optimized)
- **Historical query**: TimescaleDB continuous aggregates

### Two Device Types
1. **Device-Agent**: Custom Go agent, sends Rx/Tx via gRPC stream
2. **WireGuard Client**: Official WG clients (MikroTik, phone), data from kernel (`wg show`)

---

## Work Objectives

### Core Objective
Real-time device status + traffic monitoring for 1000+ devices with historical charts.

### Must Have
- gRPC streaming for agent traffic data
- SSE for dashboard real-time updates
- TimescaleDB for traffic logging
- Traffic chart per device/node
- Toggle to disable chart display
- Historical data query (daily/hourly)

### Must NOT Have
- Do NOT remove existing heartbeat system
- Do NOT change existing API endpoints
- Do NOT add external dependencies (Kafka, RabbitMQ)
- Do NOT use WebSocket (SSE is sufficient for dashboard)

---

## Verification Strategy

### Test Decision
- **Infrastructure exists**: YES (Go, Vue 3, PostgreSQL)
- **Automated tests**: Tests-after
- **Framework**: Go test + npm test

---

## Execution Strategy

### Parallel Execution Waves

```
Wave 1 (Foundation):
├── T1: TimescaleDB schema + migration
├── T2: gRPC proto definition
└── T3: Traffic recorder (Redis → DB)

Wave 2 (Backend):
├── T4: gRPC server implementation
├── T5: SSE endpoint
└── T6: Traffic query API

Wave 3 (Agent):
├── T7: Device-agent gRPC client
└── T8: Kernel sync enhancement

Wave 4 (Frontend):
├── T9: Dashboard traffic chart
├── T10: Toggle controls
└── T11: Historical data view
```

---

## TODOs

- [ ] 1. **TimescaleDB schema + migration**

  **What to do**:
  - Create migration file `apps/server-core/migrations/003_timescaledb_traffic.sql`
  - Create `device_traffic` table:
    ```sql
    CREATE TABLE device_traffic (
        time        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        device_id   UUID NOT NULL,
        node_id     UUID,
        rx_bytes    BIGINT DEFAULT 0,
        tx_bytes    BIGINT DEFAULT 0,
        rx_rate     BIGINT DEFAULT 0,
        tx_rate     BIGINT DEFAULT 0
    );
    SELECT create_hypertable('device_traffic', 'time');
    ```
  - Create daily aggregate view:
    ```sql
    CREATE MATERIALIZED VIEW device_traffic_daily
    WITH (timescaledb.continuous) AS
    SELECT
        time_bucket('1 day', time) AS day,
        device_id,
        MAX(rx_bytes) - MIN(rx_bytes) AS rx_total,
        MAX(tx_bytes) - MIN(tx_bytes) AS tx_total
    FROM device_traffic
    GROUP BY day, device_id;
    ```
  - Create hourly aggregate view
  - Add retention policy (90 days raw, 1 year aggregated)
  - Add indexes on device_id + time

  **Must NOT do**:
  - Do NOT remove existing tables
  - Do NOT change existing schema

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T2, T3)
  - **Parallel Group**: Wave 1
  - **Blocks**: T6
  - **Blocked By**: None

  **References**:
  - `apps/server-core/migrations/` - existing migration pattern
  - TimescaleDB docs: https://docs.timescale.com/

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] Migration runs without errors
  - [ ] Tables and views created

  **Commit**: YES
  - Message: `feat(db): add TimescaleDB schema for traffic monitoring`
  - Files: `apps/server-core/migrations/003_timescaledb_traffic.sql`

- [ ] 2. **gRPC proto definition**

  **What to do**:
  - Create `apps/server-core/proto/traffic.proto`:
    ```protobuf
    syntax = "proto3";
    package nexusguard.traffic;
    
    service TrafficService {
      rpc StreamTraffic(stream TrafficReport) returns (stream TrafficCommand);
      rpc ReportTraffic(TrafficReport) returns (TrafficAck);
    }
    
    message TrafficReport {
      string device_id = 1;
      int64 rx_bytes = 2;
      int64 tx_bytes = 3;
      int64 timestamp = 4;
    }
    
    message TrafficCommand {
      string command = 1;
      string target = 2;
    }
    
    message TrafficAck {
      bool success = 1;
      string message = 2;
    }
    ```
  - Generate Go code: `protoc --go_out=. --go-grpc_out=. traffic.proto`
  - Generate TypeScript types for frontend (optional)

  **Must NOT do**:
  - Do NOT include sensitive data in proto
  - Do NOT add authentication in proto (handle at interceptors)

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T1, T3)
  - **Parallel Group**: Wave 1
  - **Blocks**: T4, T7
  - **Blocked By**: None

  **References**:
  - gRPC Go docs: https://grpc.io/docs/languages/go/
  - Protobuf docs: https://protobuf.dev/

  **Acceptance Criteria**:
  - [ ] Proto file compiles without errors
  - [ ] Generated Go code exists
  - [ ] Generated TypeScript types exist

  **Commit**: YES
  - Message: `feat(grpc): add traffic proto definition`
  - Files: `apps/server-core/proto/traffic.proto`, generated files

- [ ] 3. **Traffic recorder (Redis → DB batch)**

  **What to do**:
  - Create `apps/server-core/internal/traffic/recorder.go`:
    - `TrafficRecorder` struct with Redis client + TimescaleDB connection
    - `Record(deviceID, rxBytes, txBytes int64)` - stores to Redis (fast)
    - `StartBatchSync(ctx, interval)` - goroutine that batch inserts to DB every 60s
    - `GetDeviceTraffic(deviceID, from, to time.Time)` - query historical data
    - `GetNodeTraffic(nodeID, from, to time.Time)` - aggregate per node
  - Redis key format: `traffic:{device_id}:{timestamp}`
  - Batch insert: collect from Redis, insert to TimescaleDB, delete from Redis
  - Handle zero-value timestamps
  - Thread-safe with sync.Mutex

  **Must NOT do**:
  - Do NOT block on Redis write
  - Do NOT lose data on server restart (Redis persistence)
  - Do NOT query DB on every traffic report

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T1, T2)
  - **Parallel Group**: Wave 1
  - **Blocks**: T4, T5, T6
  - **Blocked By**: None

  **References**:
  - `apps/server-core/internal/heartbeat/redis.go` - Redis pattern
  - TimescaleDB insert pattern

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] Traffic recorded to Redis on Report()
  - [ ] Batch sync inserts to TimescaleDB

  **Commit**: YES
  - Message: `feat(traffic): add Redis → TimescaleDB recorder`
  - Files: `apps/server-core/internal/traffic/recorder.go`

- [ ] 4. **gRPC server implementation**

  **What to do**:
  - Create `apps/server-core/api/grpc_server.go`:
    - Implement `TrafficServiceServer` interface
    - `StreamTraffic`: bidirectional streaming
      - Receive `TrafficReport` from agent
      - Call `recorder.Record()`
      - Send `TrafficCommand` if needed (e.g., rate limit)
    - `ReportTraffic`: unary call for simple reports
    - Connection management: track active agents
    - Graceful shutdown
  - Add gRPC server to main.go (separate port, e.g., 8081)
  - Add TLS support (optional, for production)

  **Must NOT do**:
  - Do NOT expose gRPC to public internet (internal only)
  - Do NOT change existing HTTP API

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: NO (depends on T2, T3)
  - **Parallel Group**: Wave 2
  - **Blocks**: T7
  - **Blocked By**: T2, T3

  **References**:
  - `apps/server-core/main.go` - server startup pattern
  - gRPC Go server examples

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] gRPC server starts on port 8081
  - [ ] StreamTraffic accepts connections

  **Commit**: YES
  - Message: `feat(grpc): implement traffic streaming server`
  - Files: `apps/server-core/api/grpc_server.go`, `apps/server-core/main.go`

- [ ] 5. **SSE endpoint**

  **What to do**:
  - Create `apps/server-core/api/sse.go`:
    - `SSEHandler` struct with Redis + recorder
    - `StreamStatus(c *gin.Context)` - SSE endpoint
    - Register client in Redis pub/sub channel
    - Push device status updates (online/offline, Rx/Tx rates)
    - Handle client disconnect (cleanup)
    - Heartbeat ping every 30s (keep connection alive)
  - Register route: `GET /api/v1/devices/stream`
  - Use Redis pub/sub for multi-instance support

  **Must NOT do**:
  - Do NOT block on SSE write
  - Do NOT store SSE clients in memory (use Redis pub/sub)

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: NO (depends on T3)
  - **Parallel Group**: Wave 2
  - **Blocks**: T9
  - **Blocked By**: T3

  **References**:
  - SSE spec: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
  - Redis pub/sub pattern

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] `curl -N http://localhost:8080/api/v1/devices/stream` returns SSE stream
  - [ ] Status updates pushed on device changes

  **Commit**: YES
  - Message: `feat(sse): add device status streaming endpoint`
  - Files: `apps/server-core/api/sse.go`, `apps/server-core/main.go`

- [ ] 6. **Traffic query API**

  **What to do**:
  - Add endpoints to `apps/server-core/api/traffic.go`:
    - `GET /api/v1/devices/:id/traffic?from=&to=` - device traffic history
    - `GET /api/v1/nodes/:id/traffic?from=&to=` - node aggregate traffic
    - `GET /api/v1/traffic/summary` - today's summary (all devices)
  - Query TimescaleDB with time bucket aggregation
  - Return JSON with timestamps + values for chart
  - Support different granularities: minute, hour, day

  **Must NOT do**:
  - Do NOT expose raw traffic data (use aggregates)
  - Do NOT allow querying beyond retention period

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: NO (depends on T1, T3)
  - **Parallel Group**: Wave 2
  - **Blocks**: T9, T11
  - **Blocked By**: T1, T3

  **References**:
  - `apps/server-core/api/devices.go` - API pattern
  - TimescaleDB time_bucket queries

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] GET /api/v1/devices/:id/traffic returns data
  - [ ] Response format suitable for charts

  **Commit**: YES
  - Message: `feat(api): add traffic query endpoints`
  - Files: `apps/server-core/api/traffic.go`, `apps/server-core/main.go`

- [ ] 7. **Device-agent gRPC client**

  **What to do**:
  - Modify `apps/device-agent/main.go`:
    - Add gRPC client connection to server (port 8081)
    - Periodic traffic report (every 5 seconds):
      - Read Rx/Tx from WireGuard interface
      - Send `TrafficReport` via gRPC stream
    - Handle server commands (rate limit, disconnect)
    - Reconnect on connection loss
  - Add `--grpc-port` flag (default 8081)
  - Add traffic collection from WireGuard kernel

  **Must NOT do**:
  - Do NOT break existing heartbeat system
  - Do NOT add new dependencies (use existing wgctrl)

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: NO (depends on T2, T4)
  - **Parallel Group**: Wave 3
  - **Blocks**: None
  - **Blocked By**: T2, T4

  **References**:
  - `apps/device-agent/main.go` - agent entry point
  - `apps/device-agent/internal/tunnel/wireguard.go` - WG interface access

  **Acceptance Criteria**:
  - [ ] `go build` passes
  - [ ] Agent connects to gRPC server
  - [ ] Traffic reports sent every 5 seconds

  **Commit**: YES
  - Message: `feat(agent): add gRPC traffic streaming client`
  - Files: `apps/device-agent/main.go`, new gRPC client code

- [ ] 8. **Kernel sync enhancement**

  **What to do**:
  - Enhance `apps/server-core/internal/wgmanager/handshakesync.go`:
    - Add Rx/Tx bytes collection per peer
    - Store traffic data to Redis/TimescaleDB
    - Handle WireGuard client devices (no agent)
  - Update `GetPeerHandshakes()` to include traffic data
  - Ensure kernel sync records traffic for all WG clients

  **Must NOT do**:
  - Do NOT change existing handshake logic
  - Do NOT break agent devices

  **Recommended Agent Profile**:
  - **Category**: `quick`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: NO (depends on T3)
  - **Parallel Group**: Wave 3
  - **Blocks**: None
  - **Blocked By**: T3

  **References**:
  - `apps/server-core/internal/wgmanager/handshakesync.go` - existing sync

  **Acceptance Criteria**:
  - [ ] `go build -tags dev ./...` passes
  - [ ] Kernel sync records Rx/Tx for WG clients

  **Commit**: YES
  - Message: `feat(wgmanager): enhance kernel sync with traffic data`
  - Files: `apps/server-core/internal/wgmanager/handshakesync.go`

- [ ] 9. **Dashboard traffic chart**

  **What to do**:
  - Create `apps/dashboard-ui/src/components/TrafficChart.vue`:
    - Line chart for Rx/Tx over time
    - Use Chart.js or ApexCharts
    - Real-time updates via SSE
    - Responsive design
  - Add chart to `DeviceDetail.vue` (per-device)
  - Add chart to `Dashboard.vue` (per-node aggregate)
  - Time range selector (1h, 6h, 24h, 7d, 30d)
  - Auto-refresh every 5 seconds

  **Must NOT do**:
  - Do NOT add heavy chart libraries (use lightweight)
  - Do NOT block UI on chart render

  **Recommended Agent Profile**:
  - **Category**: `visual-engineering`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T10, T11)
  - **Parallel Group**: Wave 4
  - **Blocks**: None
  - **Blocked By**: T5, T6

  **References**:
  - `apps/dashboard-ui/src/views/DeviceDetail.vue` - existing page
  - Chart.js docs: https://www.chartjs.org/

  **Acceptance Criteria**:
  - [ ] `npm run build` passes
  - [ ] Chart displays real-time traffic
  - [ ] Time range selector works

  **Commit**: YES
  - Message: `feat(ui): add traffic chart component`
  - Files: `apps/dashboard-ui/src/components/TrafficChart.vue`, `apps/dashboard-ui/src/views/DeviceDetail.vue`, `apps/dashboard-ui/src/views/Dashboard.vue`

- [ ] 10. **Toggle controls**

  **What to do**:
  - Add toggle to `DeviceDetail.vue`:
    - "Real-time Traffic" toggle (per device)
    - When OFF: chart shows historical only, no live updates
    - When ON: chart updates in real-time via SSE
  - Add global toggle to `Dashboard.vue`:
    - "Disable All Real-time Charts" toggle
    - Saves preference to localStorage
  - Backend: SSE still sends data, frontend just ignores if toggle OFF
  - Traffic recording always active (toggle only affects display)

  **Must NOT do**:
  - Do NOT stop recording when toggle is OFF
  - Do NOT add new backend endpoints

  **Recommended Agent Profile**:
  - **Category**: `visual-engineering`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T9, T11)
  - **Parallel Group**: Wave 4
  - **Blocks**: None
  - **Blocked By**: T9

  **References**:
  - `apps/dashboard-ui/src/views/DeviceDetail.vue`
  - localStorage pattern

  **Acceptance Criteria**:
  - [ ] `npm run build` passes
  - [ ] Per-device toggle works
  - [ ] Global toggle works
  - [ ] Recording continues when display is OFF

  **Commit**: YES
  - Message: `feat(ui): add real-time chart toggle controls`
  - Files: `apps/dashboard-ui/src/components/TrafficChart.vue`, `apps/dashboard-ui/src/views/DeviceDetail.vue`, `apps/dashboard-ui/src/views/Dashboard.vue`

- [ ] 11. **Historical data view**

  **What to do**:
  - Create `apps/dashboard-ui/src/views/TrafficHistory.vue`:
    - Full-page traffic history view
    - Date range picker
    - Device/node selector
    - Export to CSV
    - Daily/hourly aggregation
  - Add route: `/traffic-history`
  - Query backend traffic API
  - Display in table + chart

  **Must NOT do**:
  - Do NOT allow querying beyond retention period
  - Do NOT expose raw data (use aggregates only)

  **Recommended Agent Profile**:
  - **Category**: `visual-engineering`
  - **Skills**: `[]`

  **Parallelization**:
  - **Can Run In Parallel**: YES (with T9, T10)
  - **Parallel Group**: Wave 4
  - **Blocks**: None
  - **Blocked By**: T6

  **References**:
  - `apps/dashboard-ui/src/router/index.ts` - routing
  - Date picker component

  **Acceptance Criteria**:
  - [ ] `npm run build` passes
  - [ ] History page accessible at /traffic-history
  - [ ] Date range filter works
  - [ ] Export to CSV works

  **Commit**: YES
  - Message: `feat(ui): add traffic history view`
  - Files: `apps/dashboard-ui/src/views/TrafficHistory.vue`, `apps/dashboard-ui/src/router/index.ts`

---

## Final Verification Wave

- [ ] F1. **Plan Compliance Audit** — `oracle`
- [ ] F2. **Code Quality Review** — `unspecified-high`
- [ ] F3. **Real Manual QA** — `unspecified-high`
- [ ] F4. **Scope Fidelity Check** — `deep`

---

## Commit Strategy

- Commit #1: Backend — TimescaleDB + gRPC proto
- Commit #2: Backend — gRPC server + SSE
- Commit #3: Agent — gRPC client
- Commit #4: Frontend — Dashboard charts

---

## Success Criteria

### Verification Commands
```bash
go build -tags dev ./...  # Expected: no errors
cd apps/dashboard-ui && npm run build  # Expected: no errors
psql -d nexusguard -c "SELECT * FROM device_traffic LIMIT 1"  # Expected: empty or data
```

### Final Checklist
- [ ] gRPC streaming works for 1000+ connections
- [ ] SSE pushes real-time status to dashboard
- [ ] TimescaleDB stores traffic data
- [ ] Charts show real-time + historical data
- [ ] Toggle controls work (per device + global)
- [ ] Performance: <100ms latency for status updates