diff options
| author | Adam Soltys <[email protected]> | 2022-09-02 14:47:08 -0700 |
|---|---|---|
| committer | Adam Soltys <[email protected]> | 2022-09-02 14:47:14 -0700 |
| commit | 3161dd7ee95b6c7d70f8341298dd4fab84cda8b0 (patch) | |
| tree | 16cd6c55d5dc481331b1e8b640d179f193e7ea13 /lib/sockets.js | |
| parent | 7a8d64833de8873afb92982b3f5bfff59b30c542 (diff) | |
| download | coinos-server-3161dd7ee95b6c7d70f8341298dd4fab84cda8b0.tar.gz coinos-server-3161dd7ee95b6c7d70f8341298dd4fab84cda8b0.zip | |
es6 migration
Diffstat (limited to 'lib/sockets.js')
| -rw-r--r-- | lib/sockets.js | 289 |
1 files changed, 151 insertions, 138 deletions
diff --git a/lib/sockets.js b/lib/sockets.js index 91d9636c..dcb8a840 100644 --- a/lib/sockets.js +++ b/lib/sockets.js @@ -1,11 +1,14 @@ -import jwt from 'jsonwebtoken'; -import { v4 as uuidv4 } from 'uuid'; -import WebSocket from 'ws'; -import { differenceInSeconds, subSeconds } from 'date-fns'; -const code = 1000; +import app from "../app.js"; +import jwt from "jsonwebtoken"; +import { v4 as uuidv4 } from "uuid"; +import WebSocket from "ws"; +import { differenceInSeconds, subSeconds } from "date-fns"; +import fastifyWs from "@fastify/websocket"; +import whitelist from "./whitelist.js"; +const code = 1000; -wss = new WebSocket.Server({ clientTracking: false, noServer: true }); +const wss = new WebSocket.Server({ clientTracking: false, noServer: true }); /* eslint-disable-next-line */ const pick = (O, ...K) => K.reduce((o, k) => ((o[k] = O[k]), o), {}); @@ -13,14 +16,13 @@ const all = {}; const subscriptions = []; const lastSeen = {}; -emit = (username, type, data) => { +export const emit = (username, type, data) => { if (data.username !== undefined) data = pick(data, ...whitelist); - if (type === "payment" && data.amount > 0) { for (let i = subscriptions.length - 1; i >= 0; i--) { let s = subscriptions[i]; - + if ( s.invoice.amount === data.amount || (data.address && s.invoice.address === data.address) || @@ -39,7 +41,7 @@ emit = (username, type, data) => { } }; -broadcast = (type, data) => { +export const broadcast = (type, data) => { const sent = []; for (const username in sockets) { for (const id in sockets[username]) { @@ -54,152 +56,163 @@ broadcast = (type, data) => { }); }; -server.on("upgrade", function(req, socket, head) { - wss.handleUpgrade(req, socket, head, function(ws) { - wss.emit("connection", ws, req); - }); -}); +app.register(fastifyWs); +app.register(async function(app) { + app.get("/ws", { websocket: true }, ({ socket: ws }, req) => { + const id = uuidv4(); + ws.id = id; + ws.beats = 0; + ws.send(JSON.stringify({ type: "connected", data: id })); + all[id] = ws; -const names = {}; + ws.on("message", async message => { + let type, data; + try { + ({ type, data } = JSON.parse(message)); + } catch (e) { + l.error("coudn't parse socket message"); + } -setInterval(() => { - for (const username in sockets) { - for (const id in sockets[username]) { - let ws = sockets[username][id]; - ws.beats++; - if (ws.beats > 10) ws.close(code, `${username} ${id} lost heartbeat`); - } - } -}, 2000); + switch (type) { + case "heartbeat": + ws.beats = 0; + if (ws.user) emit(ws.user.username, "id", ws.id); + break; -wss.on("connection", function(ws, req) { - const id = uuidv4(); - ws.id = id; - ws.beats = 0; - ws.send(JSON.stringify({ type: "connected", data: id })); - all[id] = ws; + case "login": + let token = data; - ws.on("message", async message => { - let type, data; - try { - ({ type, data } = JSON.parse(message)); - } catch (e) { - l.error("coudn't parse socket message"); - } + // if (differenceInSeconds(new Date(), lastSeen[token]) < 5) return; + lastSeen[token] = new Date(); - switch (type) { - case "heartbeat": - ws.beats = 0; - if (ws.user) emit(ws.user.username, "id", ws.id); - break; + try { + if (!token || token === "null" || !jwt.decode(token)) + return ws.close(code, `bad token ${token}`); + const username = jwt.decode(token).username; + if (!username) throw new Error("Invalid JWT token"); - case "login": - let token = data; + if (!sockets[username]) sockets[username] = {}; + const existing = Object.keys(sockets[username]); + if (existing.length > 4) { + const p = existing.find(sid => sid !== id); + sockets[username][p].send(JSON.stringify({ type: "logout" })); + setTimeout(() => { + if (sockets[username][p]) + sockets[username][p].close(code, "too many sockets"); + }, 5000); + } + sockets[username][id] = ws; + names[id] = username; - // if (differenceInSeconds(new Date(), lastSeen[token]) < 5) return; - lastSeen[token] = new Date(); + let user = await getUser(username); + if (user.locked) throw new Error("Account is locked"); + let payments = await user.getPayments({ + where: { + account_id: user.account_id + }, + order: [["id", "DESC"]], + limit: 12, + include: [ + { + model: db.Account, + as: "account" + }, + { + model: db.Payment, + as: "fee_payment" + } + ] + }); - try { - if (!token || token === "null" || !jwt.decode(token)) return ws.close(code, `bad token ${token}`); - const username = jwt.decode(token).username; - if (!username) throw new Error("Invalid JWT token"); + let accounts = await user.getAccounts(); + let keys = await user.getKeys(); + user = user.get({ plain: true }); + user.accounts = accounts; + user.keys = keys; + user.payments = payments; - if (!sockets[username]) sockets[username] = {}; - const existing = Object.keys(sockets[username]); - if (existing.length > 4) { - const p = existing.find(sid => sid !== id); - sockets[username][p].send(JSON.stringify({ type: "logout" })); - setTimeout(() => { - if (sockets[username][p]) sockets[username][p].close(code, "too many sockets"); - }, 5000); + ws.user = user; + ws.send(JSON.stringify({ type: "login", data: user })); + ws.send( + JSON.stringify({ type: "version", data: config.clientVersion }) + ); + } catch (e) { + l.error("caught socket error", e.message); + if (e.message.includes("locked")) { + ws.send(JSON.stringify({ type: "locked" })); + ws.send(JSON.stringify({ type: "logout" })); + } + if (e.message.includes("not found")) + ws.send(JSON.stringify({ type: "logout" })); + setTimeout( + () => ws && ws.close(code, `closing due to error ${e.message}`), + 1000 + ); } - sockets[username][id] = ws; - names[id] = username; - - let user = await getUser(username); - if (user.locked) throw new Error("Account is locked"); - let payments = await user.getPayments({ + break; + case "token": + if (!ws.user) return; + ws.token = data; + break; + case "lnurl": + sessions[data] = ws; + break; + case "updateMemo": + const payment = await db.Payment.findOne({ where: { - account_id: user.account_id + user_id: ws.user.id, + id: data.id }, - order: [["id", "DESC"]], - limit: 12, - include: [{ + include: { model: db.Account, as: "account" - }, - { - model: db.Payment, - as: "fee_payment" - }] + } }); - - let accounts = await user.getAccounts(); - let keys = await user.getKeys(); - user = user.get({ plain: true }); - user.accounts = accounts; - user.keys = keys; - user.payments = payments; - - ws.user = user; - ws.send(JSON.stringify({ type: "login", data: user })); - ws.send( - JSON.stringify({ type: "version", data: config.clientVersion }) - ); - } catch (e) { - l.error("caught socket error", e.message); - if (e.message.includes("locked")) { - ws.send(JSON.stringify({ type: "locked" })); - ws.send(JSON.stringify({ type: "logout" })); + if (payment) { + payment.memo = data.memo; + await payment.save(); + emit(ws.user.username, "payment", payment); } - if (e.message.includes("not found")) - ws.send(JSON.stringify({ type: "logout" })); - setTimeout(() => ws && ws.close(code, `closing due to error ${e.message}`), 1000); - } - break; - case "token": - if (!ws.user) return; - ws.token = data; - break; - case "lnurl": - sessions[data] = ws; - break; - case "updateMemo": - const payment = await db.Payment.findOne({ - where: { - user_id: ws.user.id, - id: data.id - }, - include: { - model: db.Account, - as: "account" - } - }); - if (payment) { - payment.memo = data.memo; - await payment.save(); - emit(ws.user.username, "payment", payment); + break; + case "subscribe": + subscriptions.push({ invoice: data, ws }); + break; + default: + l.warn("received socket message of unknown type", type, data); + } + }); + + ws.on("close", function(code, reason) { + // if (reason) console.log("closing socket", reason); + try { + const username = names[id]; + if (sockets[username] && sockets[username][id]) { + delete sockets[username][id]; } - break; - case "subscribe": - subscriptions.push({ invoice: data, ws }); - break; - default: - l.warn("received socket message of unknown type", type, data); - } - }); - ws.on("close", function(code, reason) { - // if (reason) console.log("closing socket", reason); - try { - const username = names[id]; - if (sockets[username] && sockets[username][id]) { - delete sockets[username][id]; + if (all[id]) delete all[id]; + } catch (e) { + l.error("problem closing socket", e.message); } - - if (all[id]) delete all[id]; - } catch (e) { - l.error("problem closing socket", e.message); - } + }); }); }); + +app.listen({ port: 3000 }, err => { + if (err) { + fastify.log.error(err); + process.exit(1); + } +}); + +const names = {}; + +setInterval(() => { + for (const username in sockets) { + for (const id in sockets[username]) { + let ws = sockets[username][id]; + ws.beats++; + if (ws.beats > 10) ws.close(code, `${username} ${id} lost heartbeat`); + } + } +}, 2000); |
