// ws-server/ws-core.js require("dotenv").config(); const WebSocket = require("ws"); const jwt = require("jsonwebtoken"); const { randomUUID } = require("crypto"); const dbWs = require("../db/dbWs"); const PORT = process.env.WS_PORT || 4002; const HOST = process.env.WS_HOST || "0.0.0.0"; const JWT_SECRET = process.env.JWT_SECRET; if (!JWT_SECRET) { console.error("โŒ ERRORE: JWT_SECRET non definito nel .env"); process.exit(1); } const wss = new WebSocket.Server({ port: PORT, host: HOST }); console.log(`๐Ÿš€ WebSocket server attivo su ws://${HOST}:${PORT}`); (async () => { try { console.log( "๐Ÿงน [WS CLEANUP] Pulizia sessioni e pending al riavvio server...", ); await dbWs.deleteAllSessions(); await dbWs.deleteAllPendingEvents(); console.log("๐Ÿงน [WS CLEANUP] Completata"); } catch (err) { console.error("โŒ [WS CLEANUP ERROR]:", err); } })(); const wsBySession = new Map(); const WS_DORMANT_MS = 2 * 60 * 1000; function sendToClient(ws, msg) { const json = JSON.stringify(msg); console.log(`๐Ÿ“ค [WS SEND] โ†’ user=${ws.user}: ${json}`); ws.send(json); } async function decideAuthRecovery({ session_id, user, device_id }) { const sessionRow = await dbWs.getSession(session_id, device_id); let needFull = false; let needRecovery = false; let reason = "none"; if (!sessionRow) { needRecovery = true; reason = "session not found after cleanup"; } else { const lastAck = sessionRow.last_ack || 0; const dormancy = lastAck > 0 ? Date.now() - lastAck : 0; if (sessionRow.need_full_sync) { needRecovery = true; reason = "db recovery flag"; } else if (lastAck > 0 && dormancy >= WS_DORMANT_MS) { needRecovery = true; reason = "dormant"; } else { needRecovery = false; reason = "recent or clean auth"; } } await dbWs.setSessionNeedFullSync(session_id, device_id, needRecovery); console.log("๐Ÿงช [WS AUTH DECISION]", { user, session_id, device_id, needFull, needRecovery, reason, last_ack: sessionRow?.last_ack, last_sync: sessionRow?.last_sync, db_need_full_sync: sessionRow?.need_full_sync, }); return { needFull, needRecovery, reason }; } wss.on("connection", (ws) => { console.log("๐Ÿ”Œ [WS CONNECT] Nuovo client"); ws.authenticated = false; ws.user = null; ws.session_id = null; ws.device_id = null; ws.request_id = null; ws.on("message", async (msg) => { console.log(`๐Ÿ“ฉ [WS RAW] msg=${msg.toString()}`); let data; try { data = JSON.parse(msg.toString()); } catch (err) { console.error("โŒ Errore parsing JSON:", err); return; } if (data.type === "auth") { let payload; try { payload = jwt.verify(data.token, JWT_SECRET); const user = payload.name; const session_id = data.session_id || randomUUID(); const device_id = data.device_id || "default-device"; const request_id = data.request_id || null; ws.user = user; ws.session_id = session_id; ws.device_id = device_id; ws.request_id = request_id; ws.authenticated = true; if (!wsBySession.has(session_id)) { wsBySession.set(session_id, new Map()); } wsBySession.get(session_id).set(device_id, ws); console.log( `๐Ÿ” [WS AUTH OK] user=${user} session=${session_id} device=${device_id}`, ); await dbWs.upsertSession(session_id, user, device_id); const decision = await decideAuthRecovery({ session_id, user, device_id, }); sendToClient(ws, { type: "auth_ok", user, session_id, need_full_sync: decision.needFull, need_recovery: decision.needRecovery, reason: decision.reason, }); } catch (err) { if (err.name === "TokenExpiredError") { sendToClient(ws, { type: "token_expired", }); return; } console.error("โŒ [WS AUTH ERROR]:", err); sendToClient(ws, { type: "auth_error", error: err.message, }); ws.close(); } return; } if (!ws.authenticated) { sendToClient(ws, { type: "error", message: "Not authenticated" }); return; } if (data.type === "ack" && data.event_id) { console.log(`๐ŸŸข [WS ACK] user=${ws.user} event=${data.event_id}`); await dbWs.updateSessionAck(ws.session_id, ws.device_id); await dbWs.deletePendingEvent({ user_id: ws.user, session_id: ws.session_id, device_id: ws.device_id, event_id: data.event_id, }); return; } if (data.type === "client_recovered") { console.log(`๐Ÿ”„ [WS CLIENT RECOVERED] user=${ws.user}`); const session = await dbWs.getSession(ws.session_id, ws.device_id); if (!session) { sendToClient(ws, { type: "need_recovery", reason: "session missing" }); return; } const lastAck = session.last_ack || 0; const dormancy = lastAck > 0 ? Date.now() - lastAck : 0; if (dormancy >= WS_DORMANT_MS || session.need_full_sync) { sendToClient(ws, { type: "need_recovery", reason: dormancy >= WS_DORMANT_MS ? "dormant" : "db recovery flag", }); return; } const pending = await dbWs.getPendingEvents( ws.user, ws.session_id, ws.device_id, ); for (const ev of pending) { ws.send(ev.payload); } return; } if (data.type === "pong") { await dbWs.updateSessionAck(ws.session_id, ws.device_id); return; } if (data.type === "recovery_done") { console.log( `โœ… [WS RECOVERY DONE] user=${ws.user} session=${ws.session_id} device=${ws.device_id}`, ); await dbWs.clearNeedFullSync(ws.session_id, ws.device_id); await dbWs.updateSessionAck(ws.session_id, ws.device_id); return; } console.log(`๐Ÿ“จ [WS MESSAGE] user=${ws.user} data=${JSON.stringify(data)}`); }); ws.on("close", () => { console.log(`โŒ [WS DISCONNECT] user=${ws.user} session=${ws.session_id}`); if (ws.session_id && ws.device_id) { const devices = wsBySession.get(ws.session_id); if (devices) { devices.delete(ws.device_id); if (devices.size === 0) { wsBySession.delete(ws.session_id); } } } }); }); module.exports = { wss, wsBySession };