3 Commits

Author SHA1 Message Date
b11f4ea336 feat: log backfill endpoint for desktop log panel
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
- LogStreamService: ring buffer (500 entries) + getRecentLogs() method
- SupervisorController: GET /api/supervisor/logs/recent returns buffered
  log entries so the desktop log panel shows history on tab open, not
  just live stream

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-17 08:51:55 +05:30
96ae867288 feat: server log streaming via SSE for desktop log panel
- LogStreamService: singleton that extends ConsoleLogger, captures all
  NestJS log output into an RxJS Subject while preserving stdout
- main.ts: uses LogStreamService.instance as app logger
- supervisor.controller.ts: new @Sse('logs/stream') endpoint pipes
  log entries (timestamp, level, context, message) to connected clients

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-17 08:22:11 +05:30
9a016a2ed0 feat: real-time active call SSE — hold/unhold status for supervisor live monitor
- SupervisorService: added activeCallSubject (RxJS Subject), emits on all
  activeCalls Map mutations (Answered, Calling, Disconnect, Hold, Unhold)
- SupervisorController: new @Sse('active-calls/stream') endpoint
- OzonetelAgentController: callControl HOLD/UNHOLD updates activeCalls Map
  status via supervisor.updateCallStatus()

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-17 05:45:14 +05:30
5 changed files with 113 additions and 5 deletions

View File

@@ -0,0 +1,61 @@
import { ConsoleLogger } from '@nestjs/common';
import { Subject } from 'rxjs';
export type LogEntry = {
timestamp: string;
level: 'log' | 'error' | 'warn' | 'debug' | 'verbose';
context: string;
message: string;
};
// Singleton — created once in main.ts, accessed by the SSE controller
// via LogStreamService.instance. NestJS DI isn't available at bootstrap
// time (the logger is created before the container), so we use a static
// instance instead of @Injectable().
export class LogStreamService extends ConsoleLogger {
static readonly instance = new LogStreamService();
readonly logSubject = new Subject<LogEntry>();
private readonly buffer: LogEntry[] = [];
private static readonly MAX_BUFFER = 500;
getRecentLogs(limit = 200): LogEntry[] {
return this.buffer.slice(-limit);
}
private emit(level: LogEntry['level'], message: unknown, context?: string) {
const entry: LogEntry = {
timestamp: new Date().toISOString(),
level,
context: context ?? this.context ?? '',
message: typeof message === 'string' ? message : JSON.stringify(message),
};
this.buffer.push(entry);
if (this.buffer.length > LogStreamService.MAX_BUFFER) this.buffer.shift();
this.logSubject.next(entry);
}
log(message: unknown, context?: string) {
super.log(message, context);
this.emit('log', message, context);
}
error(message: unknown, stack?: string, context?: string) {
super.error(message, stack, context);
this.emit('error', message, context);
}
warn(message: unknown, context?: string) {
super.warn(message, context);
this.emit('warn', message, context);
}
debug(message: unknown, context?: string) {
super.debug(message, context);
this.emit('debug', message, context);
}
verbose(message: unknown, context?: string) {
super.verbose(message, context);
this.emit('verbose', message, context);
}
}

View File

@@ -3,9 +3,11 @@ import type { NestExpressApplication } from '@nestjs/platform-express';
import { join } from 'path'; import { join } from 'path';
import { AppModule } from './app.module'; import { AppModule } from './app.module';
import { ConfigService } from '@nestjs/config'; import { ConfigService } from '@nestjs/config';
import { LogStreamService } from './logging/log-stream.service';
async function bootstrap() { async function bootstrap() {
const app = await NestFactory.create<NestExpressApplication>(AppModule); const logger = LogStreamService.instance;
const app = await NestFactory.create<NestExpressApplication>(AppModule, { logger });
const config = app.get(ConfigService); const config = app.get(ConfigService);
app.enableCors({ app.enableCors({

View File

@@ -382,6 +382,13 @@ export class OzonetelAgentController {
try { try {
const result = await this.ozonetelAgent.callControl(body); const result = await this.ozonetelAgent.callControl(body);
if (body.action === 'HOLD') {
this.supervisor.updateCallStatus(body.ucid, 'on-hold');
} else if (body.action === 'UNHOLD') {
this.supervisor.updateCallStatus(body.ucid, 'active');
}
return result; return result;
} catch (error: any) { } catch (error: any) {
const message = error.response?.data?.message ?? error.message ?? 'Call control failed'; const message = error.response?.data?.message ?? error.message ?? 'Call control failed';

View File

@@ -1,6 +1,7 @@
import { Controller, Get, Post, Body, Query, Sse, Logger } from '@nestjs/common'; import { Controller, Get, Post, Body, Query, Sse, Logger } from '@nestjs/common';
import { Observable, filter, map } from 'rxjs'; import { Observable, filter, map } from 'rxjs';
import { SupervisorService } from './supervisor.service'; import { SupervisorService } from './supervisor.service';
import { LogStreamService } from '../logging/log-stream.service';
@Controller('api/supervisor') @Controller('api/supervisor')
export class SupervisorController { export class SupervisorController {
@@ -13,6 +14,16 @@ export class SupervisorController {
return this.supervisor.getActiveCalls(); return this.supervisor.getActiveCalls();
} }
@Sse('active-calls/stream')
streamActiveCalls(): Observable<MessageEvent> {
this.logger.log('[SSE] Active calls stream opened');
return this.supervisor.activeCallSubject.pipe(
map(event => ({
data: JSON.stringify(event),
} as MessageEvent)),
);
}
@Get('team-performance') @Get('team-performance')
async getTeamPerformance(@Query('date') date?: string) { async getTeamPerformance(@Query('date') date?: string) {
const targetDate = date ?? new Date().toISOString().split('T')[0]; const targetDate = date ?? new Date().toISOString().split('T')[0];
@@ -66,4 +77,19 @@ export class SupervisorController {
} as MessageEvent)), } as MessageEvent)),
); );
} }
@Get('logs/recent')
getRecentLogs(@Query('limit') limit?: string) {
return LogStreamService.instance.getRecentLogs(limit ? parseInt(limit, 10) : 200);
}
@Sse('logs/stream')
streamLogs(): Observable<MessageEvent> {
this.logger.log('[SSE] Log stream opened');
return LogStreamService.instance.logSubject.pipe(
map(entry => ({
data: JSON.stringify(entry),
} as MessageEvent)),
);
}
} }

View File

@@ -36,6 +36,7 @@ export class SupervisorService implements OnModuleInit {
private readonly agentStates = new Map<string, AgentStateEntry>(); private readonly agentStates = new Map<string, AgentStateEntry>();
private readonly acwTimers = new Map<string, NodeJS.Timeout>(); private readonly acwTimers = new Map<string, NodeJS.Timeout>();
readonly agentStateSubject = new Subject<{ agentId: string; state: AgentOzonetelState | string; timestamp: string }>(); readonly agentStateSubject = new Subject<{ agentId: string; state: AgentOzonetelState | string; timestamp: string }>();
readonly activeCallSubject = new Subject<{ type: 'update' | 'remove'; call?: ActiveCall; ucid: string }>();
// Worklist update stream — emitted when a missed call is created or // Worklist update stream — emitted when a missed call is created or
// assigned. Frontend SSE listener triggers an immediate worklist // assigned. Frontend SSE listener triggers an immediate worklist
// refresh so agents see new missed calls without waiting for the 30s poll. // refresh so agents see new missed calls without waiting for the 30s poll.
@@ -95,10 +96,9 @@ export class SupervisorService implements OnModuleInit {
this.logger.warn(`Ignoring call event for offline agent ${agentId} (${ucid})`); this.logger.warn(`Ignoring call event for offline agent ${agentId} (${ucid})`);
return; return;
} }
this.activeCalls.set(ucid, { const call: ActiveCall = { ucid, agentId, callerNumber, callType, startTime: eventTime, status: 'active' };
ucid, agentId, callerNumber, this.activeCalls.set(ucid, call);
callType, startTime: eventTime, status: 'active', this.activeCallSubject.next({ type: 'update', call, ucid });
});
this.logger.log(`Active call: ${agentId}${callerNumber} (${ucid})`); this.logger.log(`Active call: ${agentId}${callerNumber} (${ucid})`);
// Persist CALL_START as AgentEvent on the "Answered" moment // Persist CALL_START as AgentEvent on the "Answered" moment
@@ -130,6 +130,7 @@ export class SupervisorService implements OnModuleInit {
} else if (action === 'Disconnect') { } else if (action === 'Disconnect') {
const wasActive = this.activeCalls.get(ucid); const wasActive = this.activeCalls.get(ucid);
this.activeCalls.delete(ucid); this.activeCalls.delete(ucid);
this.activeCallSubject.next({ type: 'remove', ucid });
this.logger.log(`Call ended: ${ucid}`); this.logger.log(`Call ended: ${ucid}`);
// Persist CALL_END — pair against the start for duration. // Persist CALL_END — pair against the start for duration.
@@ -294,6 +295,17 @@ export class SupervisorService implements OnModuleInit {
// definitely stale (e.g. Disconnect webhook was dropped). // definitely stale (e.g. Disconnect webhook was dropped).
private static readonly NON_CALL_AGENT_STATES = new Set(['ready', 'offline', 'paused']); private static readonly NON_CALL_AGENT_STATES = new Set(['ready', 'offline', 'paused']);
updateCallStatus(ucid: string, status: 'active' | 'on-hold') {
const call = this.activeCalls.get(ucid);
if (!call) {
this.logger.warn(`[CALL-STATUS] No active call found for UCID ${ucid}`);
return;
}
call.status = status;
this.activeCallSubject.next({ type: 'update', call, ucid });
this.logger.log(`[CALL-STATUS] ${ucid}${status} (agent=${call.agentId})`);
}
getActiveCalls(): ActiveCall[] { getActiveCalls(): ActiveCall[] {
// Sweep stale entries before returning. The activeCalls Map is a // Sweep stale entries before returning. The activeCalls Map is a
// best-effort in-memory projection of Ozonetel call events — if // best-effort in-memory projection of Ozonetel call events — if