# Realtime Dashboard Specification (Draft)

This document is a working draft that defines the realtime metrics, AMI events, data model, and implementation approach for the realtime Asterisk dashboard. We will use this as our single source of truth while implementing the listener, Redis state updates, processor, and frontend consumer.

> Note: This is a draft. Do not start implementation yet — review and confirm fields, keys, and priorities.

---

## 1. Goals / Overview

Build a low-latency realtime dashboard showing agents, queues and calls using AMI events as the primary source of truth. The listener (daemon) will transform incoming AMI events into compact JSON messages, store them in Redis and maintain small Redis aggregates for fast reads by the dashboard UI or socket server.

The dashboard should show per-queue and global summaries with the following metrics (priority ordering is shown):

- Total Agents Online (with breakdown by queue)
- Total Busy Agents (in-call / out-call) (with breakdown by queue)
- Total Paused Agents (paused in any queue, with reason when available)
- Currently running inbound calls (queue-related calls only) (with breakdown by queue)
- Currently running outbound calls (queue-related calls only) (with breakdown by queue)
- Waiting calls (calls currently waiting in queue) (with breakdown by queue)
- Missed calls (calls entered queue but never answered) (with breakdown by queue)
- Total calls answered per queue (cumulative)
- Average call time (per queue and global, rolling windows)
- Total calls out (calls initiated by extensions to external numbers)
- Total calls out ANSWERED (external calls with final answered status)

---

## 2. High-value AMI Events to consume

Prioritized list of AMI events to subscribe to and which metric(s) they help update:

- QueueParams
  - Use: initial queue snapshot & periodic reconciliation for waiting/completed/abandoned counts.
  - Fields: `Queue`, `Calls`, `Holdtime`, `TalkTime`, `Completed`, `Abandoned`.

- QueueMember
  - Use: agent membership and per-agent stats (CallsTaken, LastCall, LoginTime, InCall, Status, Paused, Wrapuptime)
  - Fields: `Queue`, `Location`/`Interface`, `Name`, `CallsTaken`, `LastCall`, `LoginTime`, `InCall`, `Status`, `Paused`, `Penalty`, `Wrapuptime`.

- QueueCallerJoin / QueueCallerLeave / QueueCallerAbandon / QueueEntry
  - Use: waiting counters, positions and abandonment tracking.
  - Fields: `Queue`, `Position`, `CallerID`, `Uniqueid`, `Timestamp`.

- QueueMemberPause / QueueMemberStatus / QueueMemberAdded / QueueMemberRemoved
  - Use: track pause/unpause events and membership changes.

- AgentLogin / AgentLogoff / AgentCalled / AgentComplete
  - Use: agent login/logout and ACD call completion counts.

- Newchannel / DialBegin / DialEnd / Hangup / Newexten / Newstate
  - Use: active call lifecycle (create, bridge, hangup), compute durations, differentiate inbound vs outbound.
  - Key fields: `Channel`, `Uniqueid`, `Linkedid`, `CallerID`, `ConnectedLineNum`, `Context`, `Exten`, `ChannelState`, `ChannelStateDesc`.

- BridgeCreate / BridgeEnter / BridgeLeave / BridgeDestroy
  - Use: detect bridged calls (in-call state) and conference events.

- PeerStatus / ContactStatus / DeviceStateChange / ExtensionStatus
  - Use: endpoint reachability and device presence for agent/extension health.

- FullyBooted, Reload, Shutdown
  - Use: monitor system lifecycle, initiate reconcilation after reload/boot.

- CoreShowChannels / CoreShowChannelsComplete (periodic action)
  - Use: snapshot reconciliation for active channels.

- UserEvent
  - Use: optionally emit custom metadata from dialplan for the dashboard.

---

## 3. Event → Redis state mapping (suggested keys)

We will write raw-ish events to a list (`realtime:events`) for archival. For the dashboard we maintain compact aggregates.

Primary Redis keys and schemas:

- `realtime:events` (LIST)
  - Append full JSON event for auditing/processing by batch worker.

- `realtime:ext:{ext}` (HASH)
  - Fields: `status` (Idle/InUse/Unavailable), `last_event` (timestamp), `peer_status`, `roundtripusec`.
  - Updated by: `PeerStatus`, `ContactStatus`, `DeviceStateChange`, `ExtensionStatus`.

- `realtime:agent:{agent}` (HASH)
  - Fields: `state` (available/in_call/paused/offline), `queue_memberships` (comma list), `calls_taken`, `last_call_at`, `login_time`, `paused_reason`, `wrapuptime`.
  - Updated by: `QueueMember`, `AgentLogin`, `AgentLogoff`, `AgentComplete`, `QueueMemberPause`.

- `realtime:active:channels` (SET)
  - Members: `uniqueid` for active channels. Add on `Newchannel`/`DialBegin`/`BridgeEnter`. Remove on `Hangup`/`BridgeLeave`.

- `realtime:channel:{uniqueid}` (HASH)
  - Fields: `channel`, `callerid`, `connected_to`, `start_ts`, `bridge_id`, `direction` (inbound/outbound/unknown), `queue` (if related).

- `realtime:queue:{queue}:waiting` (INT)
  - Counter of waiting callers. Best maintained by `QueueParams` snapshots and `QueueCallerJoin/Leave/Abandon` deltas.

- `realtime:queue:{queue}:answered` (COUNTER)
  - Cumulative answered calls for the queue (increment on `QueueParams.Completed` deltas or `DialEnd`/`AgentComplete` when matched to queue).

- `realtime:queue:{queue}:abandoned` (COUNTER)
  - Increment on `QueueCallerAbandon` or `QueueParams.Abandoned` deltas.

- `realtime:stats:summary` (HASH)
  - Precomputed dashboard summary for fast API reads. Fields: `agents_online`, `agents_busy`, `agents_paused`, `calls_waiting_total`, `inbound_active`, `outbound_active`, `missed_total`, `last_updated_ts`, etc.

- Pub/Sub channel: `realtime:pub` (PUBSUB)
  - Publish concise delta messages (e.g. `agent:state`, `queue:join`, `channel:up`) for socket server consumption.

---

## 4. Metric definitions and computation rules

Each dashboard metric has a clear source and aggregation rule.

- Total Agents Online
  - Source: `QueueMember` / `AgentLogin` events and `realtime:agent:*` hashes.
  - Definition: count of agents with `state != offline` (or `login_time` exists). Breakdown by queue: count agents with membership in that queue and state != offline.

- Total Busy Agents (In call / Out call)
  - Source: `QueueMember.InCall`, `BridgeEnter`/`BridgeLeave`, `Newchannel` state.
  - Definition: agents with `state == in_call` (InCall=1 or currently bridged). If `outbound` (direction) set from DialBegin by matching dialed extension patterns.

- Total Paused Agents
  - Source: `QueueMember.Paused`, `QueueMemberPause` events.
  - Definition: agent hashes where `paused_reason` present or `state == paused`.

- Currently running inbound calls (queue related only)
  - Source: `QueueEntry`/`QueueCallerJoin` then `BridgeEnter` to agent; or channel metadata `queue` set when joined.
  - Definition: active channels that have `direction == inbound` and `queue` set.

- Currently running outbound calls (queue related only)
  - Source: `DialBegin` with origin extension in local/internal contexts plus queue membership for agent-originated outbound.
  - Definition: active channels with `direction == outbound` and associated queue (if initiated as part of a queue outbound flow).

- Waiting calls
  - Source: `QueueParams.Calls`, `QueueCallerJoin/Leave` deltas.
  - Definition: use `QueueParams.Calls` as authoritative snapshot; update local counter on delta.

- Missed calls
  - Source: `QueueCallerAbandon`, `QueueParams.Abandoned`.
  - Definition: increment abandoned when caller abandons before being answered.

- Total calls answered in each queue
  - Source: `QueueParams.Completed` (snapshot) or `AgentComplete`/`DialEnd` matched to queue context.

- Average call time (per queue)
  - Source: use `Hangup` or `DialEnd` durations when `queue` metadata present; calculate rolling averages (e.g., 1m/5m/1h windows) using Redis TS or a simple moving average.

- Total calls out / Total calls out ANSWERED
  - Source: `DialBegin` with origin extension and DialEnd/Hangup with `DialStatus`/`Cause`.
  - Definition: increment `calls_out_total` on DialBegin when origin matches internal extension list; increment `calls_out_answered` when call reaches answered state and then completes with non-zero answered status.

---

## 5. Implementation notes and trade-offs

- Keep the event handler minimal and non-blocking. Buffer events and flush to Redis in short batches (you already have this). Maintain critical aggregates synchronously in Redis pipelines for correctness.
- For noisy events (ContactStatus/PeerStatus) coalesce updates: update `realtime:ext:{ext}` but don't publish every change to avoid flooding the UI.
- Use periodic reconciliation: run `QueueStatus` and `CoreShowChannels` every N seconds (configurable, e.g., 5s) to fix drift.
- For event sequences that belong to the same snapshot, wait for `*Complete` events before processing the full snapshot (e.g., CoreShowChannelsComplete).
- Ensure AMI manager user has sufficient event permissions.
- Use Redis pipelines and atomic ops (HSET, SADD, SREM, INCRBY) to avoid race conditions.
- Provide a `--debug` mode to print brief event summaries to console.

---

## 6. Data retention & archiving

- `realtime:events` can be a high-volume list. Consider a TTL-based archival strategy or periodic trimming after processing into MySQL (`asterisk_events`) which is already planned.
- Keep `realtime:ext:*`, `realtime:agent:*` and `realtime:queue:*` as short-lived current-state hashes; optionally update a timestamp field and expire keys not updated for a long time (e.g., 24h).

---

## 7. Dashboard API / Frontend integration

- Option A: WebSocket gateway consumes `realtime:pub` and pushes deltas to connected clients.
- Option B: REST endpoint reads `realtime:stats:summary` and returns snapshot; front-end polls every 1–3 seconds.

Prefer WebSocket (low-latency) for live updates, backed by `realtime:stats:summary` for snapshot on reconnection.

---

## 8. Implementation plan (Full Node Listener + Node WebSocket)

Below is a step-by-step plan to implement the Full Node.js approach (AMI listener + WebSocket gateway). This plan assumes we will migrate the real-time ingestion and aggregation from PHP to Node.js, and expose the dashboard via a Node WebSocket service. The plan includes libraries, primitives to use, migration strategy (parallel run), and deployment notes.

Phase A — Decide stack & prerequisites

- Node version: `>=18` (LTS recommended).
- Core libraries:
  - AMI client: `asterisk-manager` (npm) or implement a tiny AMI TCP parser if custom features required.
  - Redis: `ioredis` (recommended for performance and streams support).
  - WebSocket: `ws` (lightweight) or `socket.io` (features like rooms/reconnect if needed).
  - Process manager: `pm2` or `systemd` for production.

Phase B — Scaffold repository and packaging

1. Create a new service folder `services/realtime-node/` with `package.json` and a minimal README.
2. Provide environment configuration via `.env` (AMI host/port/user/secret, Redis URL, queues to monitor, log level).

Phase C — Implement Node AMI listener (producer)

Goals: persistent AMI connection, robust reconnect/heartbeat, parse events to structured JSON, write to Redis.

1. Implement AMI connection manager:
  - Connect to AMI TCP (5038), send `Action: Login`, subscribe to all events.
  - Implement heartbeat: send `Action: Ping` periodically and detect stale connection.
  - Reconnect with exponential backoff on errors.

2. Event parsing & normalization:
  - Parse incoming AMI key/value events to a canonical JSON shape: { type, ts, uniqueid?, channel?, meta: {...} }.
  - Ensure field names are normalized (e.g., `Uniqueid` -> `uniqueid`).

3. Persist raw events and seed aggregates in Redis:
  - Use Redis Streams (`XADD realtime:events * payload`) for reliability and consumer groups, or fall back to LIST `realtime:events`.
  - Update immediate aggregates in Redis using pipelines for atomic ops (e.g., `HSET realtime:agent:1000 ...`, `SADD realtime:active:channels <uid>`, `INCR realtime:queue:700:answered`).
  - Publish concise delta messages on `realtime:pub` (JSON) for WebSocket gateway to broadcast.

Phase D — Aggregation rules & reconcilers

1. Implement in-memory mapping logic to determine `direction` (inbound/outbound) using configurable rules (contexts or dial patterns).
2. Implement periodic reconciliation actions:
  - Every N seconds run AMI actions: `QueueStatus` and `CoreShowChannels`.
  - Consume the resulting multi-event snapshot and reconcile Redis aggregates (use `*Complete` event to mark snapshot end).
3. For noisy events (ContactStatus/PeerStatus) throttled updates: update `realtime:ext:{ext}` every X seconds and only publish if changed.

Phase E — WebSocket gateway (consumer & broadcaster)

Option 1 (co-located): run a WebSocket server inside the same Node process as the AMI listener. Easier to implement, less IPC overhead.
Option 2 (separate): split listener and gateway into two Node processes. Gateway subscribes to Redis pub/sub or consumes Redis Streams.

Key responsibilities for the WebSocket gateway:
  - Authenticate clients (JWT or session token), implement rooms per queue or per-agent for targeted updates.
  - Subscribe to `realtime:pub` (or use consumer groups on `realtime:events`) and forward deltas to connected clients.
  - Maintain a periodic snapshot fallback by reading `realtime:stats:summary` for new clients.
  - Handle client reconnects gracefully and replay recent deltas (optionally store recent N deltas in Redis list for replay).

Phase F — Migration & parallel run

1. Run Node listener in parallel with existing PHP listener in read-only mode (Node subscribes and writes to separate Redis keys/prefix `realtime_node:*`) to validate parity.
2. Compare aggregates between PHP and Node for N minutes/hours. Implement a diff check tool (small script) to compare counts for queues and agents.
3. Flip producers: once parity is proven, switch consumers (WebSocket gateway reads Node's pubstream), and decommission the PHP producer.

Phase G — Persistence & archiving

1. Use Redis Streams with consumer groups for robust processing by the MySQL writer consumer (or continue using existing `redis:process:realtime`).
2. After successful persistence in MySQL, trim/ack events in the stream or archive to long-term storage.

Phase H — Testing, monitoring & observability

1. Unit tests for event parsing and mapping logic.
2. Integration tests against a staging Asterisk instance (or a recorded AMI trace) to validate message shapes and reconcilers.
3. Load test: simulate AMI event bursts and ensure Node service keeps up; measure event->websocket latency.
4. Add metrics (Prometheus) for:
  - AMI connection state and reconnects
  - Redis errors/retries
  - Events processed/sec
  - WebSocket connections and message lag

Phase I — Deployment

1. Deploy with `pm2` or `systemd` with restart on crash; use env-based config and a `deploy` script.
2. Provide log rotation and centralized logging (e.g., filebeat → ELK or Loki).
3. Use `pm2` health checks or external monitoring to restart on failure.

Phase J — Rollout checklist

1. Dry-run: run Node listener + gateway in a staging environment alongside the PHP components.
2. Smoke test: generate synthetic queue events and validate dashboard updates.
3. Cutover: switch clients to Node gateway, monitor for errors, rollback plan available.

Sample `package.json` dependencies (example):

```json
{
  "name": "realtime-node",
  "version": "0.1.0",
  "dependencies": {
   "asterisk-manager": "^1.0.0",
   "ioredis": "^5.0.0",
   "ws": "^8.0.0",
   "dotenv": "^16.0.0"
  }
}
```

Quick start commands (after implementation):

```bash
cd services/realtime-node
npm install
# development
node src/index.js
# production (pm2)
pm2 start src/index.js --name realtime-node
```

Acceptance criteria

- Node listener connects and maintains AMI session with healthy reconnects.
- Events are written to Redis Streams (`realtime:events`) and aggregates are updated in Redis.
- WebSocket gateway publishes deltas and new clients can fetch `realtime:stats:summary` for snapshot.
- Parity test against existing PHP producer shows matching aggregates within acceptable drift (configurable, e.g., 1%).

---

When you confirm this plan I will: (a) scaffold the Node service, (b) implement a minimal AMI connection + event parser that writes to Redis streams, and (c) add a tiny WebSocket broadcaster that subscribes to the produced stream and sends deltas to connected clients. 


---

## 9. Open questions / decisions required

- How do we classify outbound vs inbound calls definitively in your dialplan (patterns)? We need a reliable rule (e.g., external dialed numbers start with 0 or use `context`).
- Confirm which agent identifier to use: extension number, `Agent/Name`, or `Interface`. Use the canonical extension/AOR where possible.
- Confirm retention policy for `realtime:events` (keep X hours/days).
- Confirm sampling/aggregation windows for average call time and rolling stats.

---

## 10. Appendix: sample JSON payload (for `realtime:events`)

```json
{
  "type": "Newchannel",
  "timestamp": "2025-11-14T12:34:56Z",
  "uniqueid": "1600000123.123",
  "channel": "PJSIP/1000-0000001a",
  "callerid": "1000",
  "connected_line": "",
  "meta": {
    "Context": "from-internal",
    "Exten": "0123456789",
    "Priority": "1"
  }
}
```

---

Please review this draft and tell me any additions or changes (naming, metrics, or data retention). Once you confirm, I will implement the next phase (start wiring event→Redis aggregates in `deamons/default-listener.php`).