Implement real-time call updates using Redis and Socket.io

README.md

@@ -143,3 +143,169 @@ Placeholder test files are in `src/services/CallService.test.ts` and `src/routes
 - Dockerfiles per service
 - stronger contract sharing between packages
 - better observability for local debugging
+
+## Implementation Notes (Luis Noriega)
+
+# VoyceLink Call Center Dashboard
+
+This project implements a real-time call monitoring system using a microservices-based architecture. It processes call events, persists them in a relational database, and propagates updates through a real-time streaming layer to a web-based dashboard.
+
+The system is composed of three main services:
+
+- A backend service responsible for business logic and persistence
+- A real-time service responsible for event distribution via WebSockets
+- A frontend application responsible for data visualization and user interaction
+
+## Architecture
+
+The system follows an event-driven architecture in which the backend service publishes domain events to a message broker, and a separate real-time service consumes those events and distributes them to connected clients.
+
+Data flow:
+
+External clients send call events to the backend API
+- The backend processes and persists the event
+- The event is published to Redis using a Pub/Sub channel
+- The real-time service subscribes to that channel
+- Incoming events are broadcast to frontend clients via WebSockets
+- The frontend reacts to these updates and refreshes its state
+- Backend Service (call-service)
+
+## Responsibilities
+
+The backend service handles:
+
+Validation and processing of incoming call events
+Enforcement of business rules and state transitions
+Persistence of calls and events in PostgreSQL
+Exposure of REST endpoints for querying data
+Publication of events to Redis for real-time propagation
+Event Processing
+
+The system supports the following event types:
+
+- call_initiated
+- call_answered
+- call_hold
+- call_ended
+
+Each event triggers state transitions and may generate derived metadata based on defined business rules.
+
+## Business Rules
+
+The following conditions are enforced during event processing:
+
+A call must exist before transitioning to subsequent states
+Valid state transitions are required (e.g., waiting → active)
+Additional flags are generated when thresholds are exceeded:
+WAIT_TIME_EXCEEDED when waitTime is greater than 30 seconds
+HOLD_TIME_EXCEEDED when holdDuration is greater than 60 seconds
+SHORT_CALL when call duration is less than 10 seconds
+Persistence Layer
+
+Two primary tables are used:
+
+- calls: stores the current state and metadata of each call
+- call_events: stores the historical sequence of events per call
+All events are stored with associated metadata, allowing reconstruction of call history.
+
+## API Endpoints
+
+The backend exposes the following endpoints:
+
+- POST /api/events
+Accepts and processes incoming call events
+- GET /api/calls
+Retrieves calls with optional filtering by status and queue
+- GET /api/calls//events
+Retrieves the event history for a specific call
+
+## Redis Integration
+
+A Redis Pub/Sub mechanism is used to propagate events to the real-time service.
+After persisting an event, the backend publishes it to a Redis channel:
+
+await redis.publish('call_events', JSON.stringify(event));
+This decouples the backend from the real-time layer and allows for horizontal scalability.
+
+## Real-time Service
+
+The real-time service is responsible for:
+
+- Subscribing to Redis channels
+- Receiving and parsing incoming events
+- Broadcasting updates to connected clients via WebSockets
+
+## Redis Subscription
+
+The service subscribes to the call_events channel and invokes a handler for each message received. The message payload is parsed and transformed into a format suitable for client consumption.
+
+## WebSocket Layer
+
+Socket.io is used to maintain persistent connections with frontend clients.
+Clients can subscribe to specific call identifiers, allowing the system to emit updates only to relevant consumers.
+
+socket.join(callId);
+Event Broadcasting
+
+# Frontend Application
+
+The frontend application is responsible for:
+
+- Fetching and displaying call data
+- Allowing filtering and selection of calls
+- Displaying event history for individual calls
+- Reacting to real-time updates via WebSockets
+
+## Data Integration
+
+All mock data has been replaced with real API calls.
+The application interacts with the backend using HTTP requests and maintains synchronization through WebSocket events.
+Call List Management
+
+## Event History Management
+
+The event history for a selected call is handled by a dedicated hook that:
+
+    Fetches the full event history on selection
+    Subscribes to real-time updates for that specific call
+    Appends new events as they are received
+    WebSocket Client
+
+A singleton Socket.io client is implemented to manage the connection lifecycle and subscriptions. Clients subscribe and unsubscribe from call-specific channels dynamically.
+
+## Testing
+
+Unit and integration tests were implemented for the backend using Vitest.
+
+Test coverage includes:
+    Event processing and persistence
+    State transitions and validation rules
+    Flag generation logic
+    API response validation
+    Error handling scenarios
+    Design Considerations
+
+The system was designed with the following principles:
+
+- Separation of concerns between services
+- Event-driven communication using Redis
+- Backend as the single source of truth
+- Scalable real-time communication via WebSockets
+- Minimal coupling between components
+- Trade-offs
+
+Some trade-offs were made to balance simplicity and correctness:
+
+The frontend performs a full refetch on real-time updates rather than partial state mutation
+Authentication is limited to API key validation
+State management in the frontend is kept simple without a global store
+Running the System
+    Start infrastructure services (PostgreSQL and Redis) using Docker
+    Start the backend service
+    Start the real-time service
+    Start the frontend application
+
+Events can be sent to the backend via HTTP requests, and the system will process and propagate updates in real time.
+
+## Path video
+https://drive.google.com/file/d/1usn9aCGOl-cZBBaeA1mr-SUPlYW9G_Ne/view?usp=drive_link

packages/call-service/package.json

@@ -16,7 +16,7 @@
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
     "express": "^4.19.2",
-    "ioredis": "^5.3.2",
+    "ioredis": "^5.10.1",
     "pg": "^8.11.5",
     "uuid": "^9.0.1",
     "zod": "^3.23.8"

packages/call-service/src/app.ts

@@ -0,0 +1,13 @@
+import express from 'express';
+import cors from 'cors';
+import eventsRouter from './routes/events';
+import callsRouter from './routes/calls';
+
+const app = express();
+
+app.use(cors());
+app.use(express.json());
+app.use('/api/events', eventsRouter);
+app.use('/api/calls', callsRouter);
+app.get('/health', (_req, res) => res.json({ status: 'ok' }));
+export default app;

packages/call-service/src/bus/publisher.ts

@@ -1,13 +1,11 @@
-import Redis from 'ioredis';
-import { config } from '../config';
 import type { CallStatusUpdate } from '../domain/call';
 
-const redis = new Redis(config.redisUrl);
-
 export const CHANNEL = 'call-status-updates';
 
 export async function publishStatusUpdate(
   update: CallStatusUpdate,
 ): Promise<void> {
-  await redis.publish(CHANNEL, JSON.stringify(update));
+  // Disabled to avoid duplicate Redis publishing.
+  // CallService now handles event publishing directly.
+  return;
 }

packages/call-service/src/lib/redis.ts

@@ -0,0 +1,12 @@
+import Redis from 'ioredis';
+import { config } from '../config';
+
+export const redis = new Redis(config.redisUrl);
+
+redis.on('connect', () => {
+  console.log('Redis connected');
+});
+
+redis.on('error', (err) => {
+  console.error('Redis error', err);
+});

packages/call-service/src/routes/events.test.ts

@@ -1,7 +1,61 @@
-import { describe, it } from 'vitest';
+import { describe, it, expect, beforeEach } from 'vitest';
+import request from 'supertest';
+import  app  from '../app';
+import { db } from '../db/client';
 
 describe('POST /api/events', () => {
-  it.todo('returns 201 and persists the event for a valid call_initiated payload');
-  it.todo('returns 400 for an invalid payload');
-  it.todo('returns 401 when the API key is missing');
+  beforeEach(async () => {
+    await db.query('DELETE FROM call_events');
+    await db.query('DELETE FROM calls');
+  });
+  const API_KEY = 'change-me';
+  it('returns 201 and persists the event for a valid call_initiated payload', async () => {
+    const res = await request(app)
+      .post('/api/events')
+      .set('x-api-key', API_KEY)
+      .send({
+        event: 'call_initiated',
+        callId: 'int-1',
+        type: 'voice',
+        queueId: 'medical_spanish',
+      });
+
+    expect(res.status).toBe(201);
+    expect(res.body.callId).toBe('int-1');
+
+    const result = await db.query(
+      'SELECT * FROM calls WHERE id = $1',
+      ['int-1']
+    );
+
+    expect(result.rowCount).toBe(1);
+  });
+
+  it('returns 400 for an invalid payload', async () => {
+    const res = await request(app)
+      .post('/api/events')
+      .set('x-api-key', 'change-me')
+      .send({
+        event: 'call_initiated',
+        callId: 'int-2',
+        type: 'voice',
+      });
+
+    expect(res.status).toBe(400);
+    expect(res.body.message).toBe('Invalid event payload');
+    expect(res.body.issues).toBeDefined();
+  });
+
+  it('returns 401 when the API key is missing', async () => {
+    const res = await request(app)
+      .post('/api/events')
+      .send({
+        event: 'call_initiated',
+        callId: 'int-3',
+        type: 'voice',
+        queueId: 'medical_spanish',
+      });
+    expect(res.status).toBe(401);
+    expect(res.body.message).toBeDefined();
+  });
 });

packages/call-service/src/routes/events.ts

@@ -12,15 +12,14 @@ router.post('/', apiKeyAuth, async (req: Request, res: Response) => {
     const payload: EventPayload = eventPayloadSchema.parse(req.body);
     const event = await callService.processEvent(payload);
     res.status(201).json(event);
-  } catch (error) {
-    if (error instanceof ZodError) {
-      res.status(400).json({
+  }  catch (error: any) {
+    if (error?.name === 'ZodError') {
+      return res.status(400).json({
         message: 'Invalid event payload',
         issues: error.issues,
       });
-      return;
     }
-
+    console.error(error);
     res.status(500).json({ message: 'Internal server error' });
   }
 });