const ws = require('ws'); const URL = require('url'); const { listen } = require('../lib/db/notify'); const channels = require('../lib/db/channels'); const jwt = require('../lib/jwt'); function start (server, pingInterval=30000) { const wsServer = new ws.Server({ noServer: true }); wsServer.on('connection', socket => { function scheduleJwtRefresh (token) { if (!token) { console.warn("No token to refresh!"); return; } const decoded = jwt.decode(token); console.log("scheduleJwtRefresh for token", token); console.log("decoded as", decoded); const exp = decoded?.exp; if (exp) { const timeout = (exp*1000 - Date.now()) / 2; if (!socket._jwtRefresh) { socket._jwtRefresh = setTimeout(() => refreshJwt(token), timeout); console.log(`Scheduled JWT refresh in ${timeout/1000} seconds at time ${(new Date(Date.now() + timeout)).toISOString()}`); } } else { console.log("Token has no exp claim. Refresh not scheduled"); } } function refreshJwt (token) { console.log("refreshJwt called"); jwt.checkValidCredentials({jwt: token}).then( decoded => { console.log("refreshJwt decoded JWT = ", decoded); if (decoded) { // The connection is now authenticated. // Let us remember this user's details socket._jwt = decoded; console.log("Renewing JWT via websocket"); delete decoded.exp; const token = jwt.issue(decoded); socket.send(JSON.stringify({ channel: ".jwt", payload: { token } })); scheduleJwtRefresh(token); } else { console.warn("FAILED to decode JWT"); delete socket._jwt; } }) .catch( err => { console.log("refreshJwt: Invalid credentials found"); console.error(err); delete socket._jwt; socket.close(); }); } socket.alive = true; socket.on('pong', function () { this.alive = true; }) socket.on('message', message => { // console.log("Websocket message:"); // console.log(message); try { const payload = JSON.parse(message); if (payload?.jwt) { refreshJwt(payload.jwt); } } catch (err) { console.warn("Websocket message decoding failed", err); } }); socket.on('close', () => { if (socket._jwtRefresh) { clearTimeout(socket._jwtRefresh); } }); }); server.on('upgrade', (request, socket, head) => { // console.log("Received upgrade request", request.url); const url = URL.parse(request.url); if (/^\/ws\/?$/.test(url.pathname)) { wsServer.handleUpgrade(request, socket, head, socket => { wsServer.emit('connection', socket, request); }); } }); listen(channels, (data) => { wsServer.clients.forEach( (socket) => { if (socket._jwt) { // Only send notifications to authenticated users // FIXME should implement authorisation control as in the API socket.send(JSON.stringify(data)); } }) }); const interval = setInterval( () => { wsServer.clients.forEach( (socket) => { if (!socket.alive) { return socket.terminate(); } socket.alive = false; socket.ping(); }) }, pingInterval); wsServer.on('close', () => clearInterval(interval)); return wsServer; } module.exports = { start }