320 lines
20 KiB
JavaScript
320 lines
20 KiB
JavaScript
"use strict";
|
|
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
|
|
function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
|
|
return new (P || (P = Promise))(function (resolve, reject) {
|
|
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
|
|
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
|
|
function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
|
|
step((generator = generator.apply(thisArg, _arguments || [])).next());
|
|
});
|
|
};
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
exports.handleWebSocketMessage = exports.createWebSocketServer = exports.broadcastToSession = exports.sessions = exports.SessionState = void 0;
|
|
const ws_1 = require("ws");
|
|
const LLMService_1 = require("../services/LLMService");
|
|
const EncryptionService_1 = require("../services/EncryptionService");
|
|
// Initialize Encryption Service
|
|
const encryptionService = new EncryptionService_1.EncryptionService(process.env.ENCRYPTION_KEY || '');
|
|
// Define the SessionState enum
|
|
var SessionState;
|
|
(function (SessionState) {
|
|
SessionState["SETUP"] = "SETUP";
|
|
SessionState["GATHERING"] = "GATHERING";
|
|
SessionState["HARMONIZING"] = "HARMONIZING";
|
|
SessionState["FINAL"] = "FINAL";
|
|
SessionState["ERROR"] = "ERROR";
|
|
})(SessionState = exports.SessionState || (exports.SessionState = {}));
|
|
exports.sessions = new Map();
|
|
// Initialize LLM Service (API key from environment)
|
|
const llmService = new LLMService_1.LLMService(process.env.GEMINI_API_KEY || '');
|
|
// Structured logging function
|
|
const logEvent = (eventName, sessionId, details = {}) => {
|
|
console.log(JSON.stringify(Object.assign({ timestamp: new Date().toISOString(), eventName, sessionId }, details)));
|
|
};
|
|
// Metrics recording function
|
|
const recordMetric = (metricName, value, sessionId, details = {}) => {
|
|
console.log(JSON.stringify(Object.assign({ timestamp: new Date().toISOString(), metricName, value, sessionId }, details)));
|
|
};
|
|
// Helper to create a serializable version of the session state
|
|
const getSerializableSession = (sessionData, currentClientId = null) => {
|
|
const filteredResponses = new Map();
|
|
sessionData.responses.forEach((response, clientId) => {
|
|
if (clientId === currentClientId) {
|
|
// For the current client, decrypt and send their own full response
|
|
const decryptedWants = response.wants.map((d) => encryptionService.decrypt(d));
|
|
const decryptedAccepts = response.accepts.map((d) => encryptionService.decrypt(d));
|
|
const decryptedNoGoes = response.noGoes.map((d) => encryptionService.decrypt(d));
|
|
const decryptedAfraidToAsk = encryptionService.decrypt(response.afraidToAsk);
|
|
filteredResponses.set(clientId, { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: decryptedAfraidToAsk });
|
|
}
|
|
else {
|
|
// For other clients, only send non-AfraidToAsk parts (wants, accepts, noGoes without AfraidToAsk)
|
|
const decryptedWants = response.wants.map((d) => encryptionService.decrypt(d));
|
|
const decryptedAccepts = response.accepts.map((d) => encryptionService.decrypt(d));
|
|
const decryptedNoGoes = response.noGoes.map((d) => encryptionService.decrypt(d));
|
|
filteredResponses.set(clientId, { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: "" }); // Hide afraidToAsk for other clients
|
|
}
|
|
});
|
|
return Object.assign(Object.assign({}, sessionData), { responses: Object.fromEntries(filteredResponses), clients: Array.from(sessionData.clients.keys()) });
|
|
};
|
|
const broadcastToSession = (sessionId, message, excludeClientId = null) => {
|
|
const sessionData = exports.sessions.get(sessionId);
|
|
if (sessionData) {
|
|
sessionData.clients.forEach((client, clientId) => {
|
|
if (clientId !== excludeClientId && client.readyState === ws_1.WebSocket.OPEN) {
|
|
const serializableMessage = Object.assign(Object.assign({}, message), { payload: Object.assign(Object.assign({}, message.payload), { session: getSerializableSession(sessionData, clientId) }) });
|
|
client.send(JSON.stringify(serializableMessage));
|
|
}
|
|
});
|
|
}
|
|
};
|
|
exports.broadcastToSession = broadcastToSession;
|
|
const SESSION_TIMEOUT_MINUTES = parseInt(process.env.SESSION_TIMEOUT_MINUTES || '1440', 10);
|
|
const SESSION_TIMEOUT_MS = SESSION_TIMEOUT_MINUTES * 60 * 1000; // Convert minutes to milliseconds
|
|
// Function to clean up inactive sessions
|
|
const cleanupInactiveSessions = () => {
|
|
const now = Date.now();
|
|
for (const [sessionId, sessionData] of exports.sessions.entries()) {
|
|
if (sessionData.clients.size === 0 && (now - sessionData.lastActivity > SESSION_TIMEOUT_MS)) {
|
|
exports.sessions.delete(sessionId);
|
|
logEvent('session_purged_inactive', sessionId);
|
|
console.log(`Inactive session ${sessionId} purged.`);
|
|
}
|
|
}
|
|
};
|
|
const createWebSocketServer = (server) => {
|
|
const wss = new ws_1.WebSocketServer({ server });
|
|
// Schedule periodic cleanup of inactive sessions
|
|
setInterval(cleanupInactiveSessions, 60 * 60 * 1000); // Run every hour
|
|
wss.on('connection', (ws, req) => {
|
|
const url = new URL(req.url || '', `http://${req.headers.host}`);
|
|
const sessionId = url.pathname.split('/').pop();
|
|
if (!sessionId) {
|
|
ws.close(1008, 'Invalid session ID');
|
|
return;
|
|
}
|
|
let sessionData = null;
|
|
// Set up a ping interval to keep the connection alive
|
|
const pingInterval = setInterval(() => {
|
|
if (ws.readyState === ws_1.WebSocket.OPEN) {
|
|
ws.ping();
|
|
}
|
|
}, 30000); // Send ping every 30 seconds
|
|
ws.on('message', (message) => __awaiter(void 0, void 0, void 0, function* () {
|
|
const parsedMessage = JSON.parse(message.toString());
|
|
const updatedSessionData = yield (0, exports.handleWebSocketMessage)(ws, sessionId, parsedMessage);
|
|
if (updatedSessionData) {
|
|
sessionData = updatedSessionData;
|
|
}
|
|
}));
|
|
ws.on('close', () => {
|
|
clearInterval(pingInterval); // Clear the interval when the connection closes
|
|
let disconnectedClientId = null;
|
|
const currentSessionData = exports.sessions.get(sessionId); // Retrieve the latest sessionData
|
|
if (currentSessionData) { // Check if sessionData is not null
|
|
for (const [clientId, clientWs] of currentSessionData.clients.entries()) {
|
|
if (clientWs === ws) {
|
|
disconnectedClientId = clientId;
|
|
break;
|
|
}
|
|
}
|
|
if (disconnectedClientId) {
|
|
currentSessionData.clients.delete(disconnectedClientId);
|
|
console.log(`Client ${disconnectedClientId} disconnected from session: ${sessionId}. Remaining clients: ${currentSessionData.clients.size}`);
|
|
}
|
|
else {
|
|
console.log(`An unregistered client disconnected from session: ${sessionId}.`);
|
|
}
|
|
if (currentSessionData.clients.size === 0) {
|
|
// Only purge session if it's in SETUP, FINAL, or ERROR state
|
|
if (currentSessionData.state === SessionState.SETUP ||
|
|
currentSessionData.state === SessionState.FINAL ||
|
|
currentSessionData.state === SessionState.ERROR) {
|
|
exports.sessions.delete(sessionId);
|
|
logEvent('session_purged', sessionId);
|
|
console.log(`Session ${sessionId} closed and state cleared.`);
|
|
}
|
|
else {
|
|
console.log(`Session ${sessionId} is in ${currentSessionData.state} state. Not purging despite no active clients.`);
|
|
}
|
|
}
|
|
}
|
|
else {
|
|
console.log(`Client disconnected from session: ${sessionId}. Session data was null.`);
|
|
}
|
|
});
|
|
ws.on('error', (error) => {
|
|
console.error(`WebSocket error in session ${sessionId}:`, error);
|
|
});
|
|
});
|
|
return wss;
|
|
};
|
|
exports.createWebSocketServer = createWebSocketServer;
|
|
const handleWebSocketMessage = (ws, sessionId, parsedMessage) => __awaiter(void 0, void 0, void 0, function* () {
|
|
const { type, clientId, payload } = parsedMessage;
|
|
if (!clientId) {
|
|
console.error(`Received message without clientId in session ${sessionId}. Type: ${type}`);
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'clientId is required' } }));
|
|
return exports.sessions.get(sessionId) || null; // Return current session state if available
|
|
}
|
|
let sessionData = exports.sessions.get(sessionId);
|
|
// Update lastActivity timestamp on any message
|
|
if (sessionData) {
|
|
sessionData.lastActivity = Date.now();
|
|
console.log(`Session ${sessionId}: lastActivity updated to ${sessionData.lastActivity}`);
|
|
}
|
|
// If sessionData is null here, it means a JOIN_SESSION message hasn't been processed yet for a new session.
|
|
// The JOIN_SESSION case will handle its creation.
|
|
if (!sessionData && type !== 'JOIN_SESSION') {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session not found and message is not JOIN_SESSION' } }));
|
|
return null; // Session not found, return null
|
|
}
|
|
switch (type) {
|
|
case 'JOIN_SESSION':
|
|
if (!sessionData) {
|
|
// Create a new session if it doesn't exist
|
|
const newSessionData = {
|
|
state: SessionState.SETUP,
|
|
topic: null,
|
|
description: null,
|
|
expectedResponses: 0,
|
|
submittedCount: 0,
|
|
responses: new Map(),
|
|
clients: new Map(),
|
|
finalResult: null,
|
|
lastActivity: Date.now(), // Initialize lastActivity
|
|
};
|
|
exports.sessions.set(sessionId, newSessionData); // Explicitly set in global map
|
|
sessionData = newSessionData; // Update local reference to the newly created session
|
|
console.log(`New session ${sessionId} created upon client ${clientId} joining. lastActivity: ${sessionData.lastActivity}`);
|
|
}
|
|
// Register the client to the session's clients map
|
|
if (!sessionData.clients.has(clientId)) {
|
|
sessionData.clients.set(clientId, ws);
|
|
console.log(`Client ${clientId} joined session: ${sessionId}. Total clients: ${sessionData.clients.size}`);
|
|
}
|
|
ws.send(JSON.stringify({ type: 'STATE_UPDATE', payload: { session: getSerializableSession(sessionData, clientId) } }));
|
|
console.log(`Client ${clientId} received STATE_UPDATE for session ${sessionId} upon joining.`);
|
|
return sessionData; // Return the updated sessionData
|
|
case 'PING':
|
|
if (!sessionData) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for PING' } }));
|
|
return null;
|
|
}
|
|
// Respond to client pings with a pong
|
|
if (ws.readyState === ws_1.WebSocket.OPEN) {
|
|
ws.pong();
|
|
}
|
|
return sessionData || null; // Return current sessionData or null if undefined
|
|
case 'SETUP_SESSION':
|
|
if (!sessionData) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for SETUP_SESSION' } }));
|
|
return null;
|
|
}
|
|
if (sessionData.state === SessionState.SETUP) {
|
|
const { expectedResponses, topic, description } = payload;
|
|
if (typeof expectedResponses !== 'number' || expectedResponses <= 0) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Invalid expectedResponses' } }));
|
|
return sessionData || null; // Return current sessionData on error
|
|
}
|
|
sessionData.expectedResponses = expectedResponses;
|
|
sessionData.topic = topic || 'Untitled Session';
|
|
sessionData.description = description || null;
|
|
sessionData.state = SessionState.GATHERING;
|
|
(0, exports.broadcastToSession)(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
|
console.log(`Session ${sessionId} moved to GATHERING with topic "${sessionData.topic}" and ${expectedResponses} expected responses.`);
|
|
}
|
|
else {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in SETUP state. Current state: ${sessionData.state}` } }));
|
|
}
|
|
return sessionData || null; // Return current sessionData
|
|
case 'SUBMIT_RESPONSE':
|
|
if (!sessionData) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'Session data not available for SUBMIT_RESPONSE' } }));
|
|
return null;
|
|
}
|
|
if (sessionData.state === SessionState.GATHERING) {
|
|
if (sessionData.responses.has(clientId)) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'You have already submitted a response for this session.' } }));
|
|
return sessionData || null; // Return current sessionData on error
|
|
}
|
|
const { wants, accepts, noGoes, afraidToAsk } = payload.response;
|
|
if ([...wants, ...accepts, ...noGoes].some((desire) => desire.length > 500) || afraidToAsk.length > 500) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: 'One of your desires or afraidToAsk exceeds the 500 character limit.' } }));
|
|
return sessionData || null; // Return current sessionData on error
|
|
}
|
|
const hasContradictionsGist = yield llmService.checkForInnerContradictions(payload.response);
|
|
if (hasContradictionsGist) {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Your submission contains inner contradictions: ${hasContradictionsGist} Please resolve them and submit again.` } }));
|
|
return sessionData || null; // Return current sessionData on error
|
|
}
|
|
const encryptedWants = wants.map((d) => encryptionService.encrypt(d));
|
|
const encryptedAccepts = accepts.map((d) => encryptionService.encrypt(d));
|
|
const encryptedNoGoes = noGoes.map((d) => encryptionService.encrypt(d));
|
|
const encryptedAfraidToAsk = encryptionService.encrypt(afraidToAsk);
|
|
sessionData.responses.set(clientId, { wants: encryptedWants, accepts: encryptedAccepts, noGoes: encryptedNoGoes, afraidToAsk: encryptedAfraidToAsk });
|
|
sessionData.submittedCount++;
|
|
console.log(`Client ${clientId} submitted response. Submitted count: ${sessionData.submittedCount}/${sessionData.expectedResponses}`);
|
|
if (sessionData.submittedCount === sessionData.expectedResponses) {
|
|
sessionData.state = SessionState.HARMONIZING;
|
|
(0, exports.broadcastToSession)(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
|
console.log(`Session ${sessionId} moved to HARMONIZING. Triggering LLM analysis.`);
|
|
// Perform LLM analysis asynchronously
|
|
(() => __awaiter(void 0, void 0, void 0, function* () {
|
|
try {
|
|
console.log('llm_analysis_started', sessionId);
|
|
const startTime = process.hrtime.bigint();
|
|
const allDecryptedDesires = Array.from(sessionData.responses.values()).map(encryptedResponse => {
|
|
const decryptedWants = encryptedResponse.wants.map((d) => encryptionService.decrypt(d));
|
|
const decryptedAccepts = encryptedResponse.accepts.map((d) => encryptionService.decrypt(d));
|
|
const decryptedNoGoes = encryptedResponse.noGoes.map((d) => encryptionService.decrypt(d));
|
|
const decryptedAfraidToAsk = encryptionService.decrypt(encryptedResponse.afraidToAsk);
|
|
return { wants: decryptedWants, accepts: decryptedAccepts, noGoes: decryptedNoGoes, afraidToAsk: decryptedAfraidToAsk };
|
|
});
|
|
const decision = yield llmService.analyzeDesires(allDecryptedDesires);
|
|
sessionData.finalResult = decision;
|
|
sessionData.state = SessionState.FINAL;
|
|
(0, exports.broadcastToSession)(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
|
console.log('llm_analysis_completed', sessionId, { result: decision });
|
|
console.log('llm_analysis_duration', 0, sessionId, { status: 'success' });
|
|
console.log('llm_analysis_availability', 'available', sessionId);
|
|
console.log(`Analysis complete for session ${sessionId}. Result:`, decision);
|
|
}
|
|
catch (error) {
|
|
console.error(`Error during analysis for session ${sessionId}:`, error.message);
|
|
sessionData.state = SessionState.ERROR;
|
|
(0, exports.broadcastToSession)(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
|
console.log('llm_analysis_error', sessionId, { error: error.message });
|
|
console.log('llm_analysis_availability', 'unavailable', sessionId, { error: error.message });
|
|
}
|
|
}))();
|
|
}
|
|
else {
|
|
// Only broadcast the latest count if the session is not yet harmonizing
|
|
(0, exports.broadcastToSession)(sessionId, { type: 'STATE_UPDATE', payload: {} });
|
|
}
|
|
}
|
|
else {
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Session is not in GATHERING state. Current state: ${sessionData.state}` } }));
|
|
}
|
|
return sessionData || null; // Return current sessionData
|
|
default:
|
|
console.warn(`Unknown message type: ${type} from client ${clientId} in session ${sessionId}`);
|
|
ws.send(JSON.stringify({ type: 'ERROR', payload: { message: `Unknown message type: ${type}` } }));
|
|
return sessionData || null; // Return current sessionData
|
|
}
|
|
});
|
|
exports.handleWebSocketMessage = handleWebSocketMessage;
|
|
const cleanupInactiveSessions = () => {
|
|
console.log('Running cleanupInactiveSessions...');
|
|
const now = Date.now();
|
|
for (const [sessionId, sessionData] of exports.sessions.entries()) {
|
|
console.log(`Session ${sessionId}: clients.size=${sessionData.clients.size}, lastActivity=${sessionData.lastActivity}, timeSinceLastActivity=${now - sessionData.lastActivity}, SESSION_TIMEOUT_MS=${SESSION_TIMEOUT_MS}`);
|
|
if (sessionData.clients.size === 0 && (now - sessionData.lastActivity > SESSION_TIMEOUT_MS)) {
|
|
exports.sessions.delete(sessionId);
|
|
logEvent('session_purged_inactive', sessionId);
|
|
console.log(`Inactive session ${sessionId} purged.`);
|
|
}
|
|
}
|
|
};
|