diff --git a/lith/musical-socket.mjs b/lith/musical-socket.mjs index 5433398e0f..de2e9bec2e 100644 --- a/lith/musical-socket.mjs +++ b/lith/musical-socket.mjs @@ -1,15 +1,15 @@ import {WebSocketServer, WebSocket} from 'ws'; +import {attachSocketRoute} from './socket-routing.mjs'; // Same decision service and durable allowance as HTTP. No audio or tokens in URLs. export function attachMusicalSocket(server, {authenticate, decide, authMs=5000, lifetimeMs=600000}={}) { const wss=new WebSocketServer({noServer:true,maxPayload:8192,perMessageDeflate:false}); const accounts=new Map(); const upgrade=(req,socket,head)=>{ - if(req.url!=='/api/easel-musical-stream')return; if(wss.clients.size>=64){socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n');return;} wss.handleUpgrade(req,socket,head,ws=>wss.emit('connection',ws)); }; - server.on('upgrade',upgrade); + const detach=attachSocketRoute(server,['/api/easel-musical-stream'],upgrade); wss.on('connection',ws=>{ let subject,authenticating=false,active=null,queued=null,closed=false,alive=true; let messages=0,windowAt=Date.now(); @@ -64,5 +64,5 @@ export function attachMusicalSocket(server, {authenticate, decide, authMs=5000, ws.on('error',()=>{}); ws.on('close',()=>{closed=true;clearTimeout(authTimer);clearTimeout(lifeTimer);clearInterval(pingTimer);active?.controller.abort();queued=null;if(subject){const n=(accounts.get(subject)||1)-1;if(n)accounts.set(subject,n);else accounts.delete(subject);}}); }); - return {close(){server.off('upgrade',upgrade);for(const ws of wss.clients)ws.terminate();wss.close();},wss}; + return {close(){detach();for(const ws of wss.clients)ws.terminate();wss.close();},wss}; } diff --git a/lith/socket-routing.mjs b/lith/socket-routing.mjs new file mode 100644 index 0000000000..438eab1bb1 --- /dev/null +++ b/lith/socket-routing.mjs @@ -0,0 +1,31 @@ +const routers = new WeakMap(); + +// One upgrade listener must own both dispatch and rejection. Independent +// listeners that ignore unknown paths leave those sockets open until timeout. +export function attachSocketRoute(server, paths, handler) { + let router = routers.get(server); + if (!router) { + const routes = new Map(); + const upgrade = (request, socket, head) => { + const handle = routes.get(request.url); + if (handle) return handle(request, socket, head); + socket.on("error", () => {}); + socket.end("HTTP/1.1 404 Not Found\r\nConnection: close\r\nContent-Length: 0\r\n\r\n", () => socket.destroy()); + }; + router = { routes, upgrade }; + routers.set(server, router); + server.on("upgrade", upgrade); + } + if (paths.some(path => router.routes.has(path))) throw Error("Socket route already registered"); + for (const path of paths) router.routes.set(path, handler); + let detached = false; + return () => { + if (detached) return; + detached = true; + for (const path of paths) router.routes.delete(path); + if (!router.routes.size) { + server.off("upgrade", router.upgrade); + routers.delete(server); + } + }; +} diff --git a/lith/socket-routing.test.mjs b/lith/socket-routing.test.mjs new file mode 100644 index 0000000000..53261a239a --- /dev/null +++ b/lith/socket-routing.test.mjs @@ -0,0 +1,34 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { once } from "node:events"; +import { WebSocket } from "ws"; +import { attachMusicalSocket } from "./musical-socket.mjs"; +import { attachWhistlegraphSocket } from "./whistlegraph-socket.mjs"; + +test("unmatched upgrades close promptly while all three registered routes still upgrade", { timeout: 5000 }, async t => { + const server = createServer((_req, res) => res.end("ok")); + const sockets = new Set(); + server.on("connection", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); + const musical = attachMusicalSocket(server, {}), whistlegraph = attachWhistlegraphSocket(server, {}); + t.after(() => { musical.close(); whistlegraph.close(); for (const socket of sockets) socket.destroy(); server.close(); }); + server.listen(0, "127.0.0.1"); await once(server, "listening"); + const base = `ws://127.0.0.1:${server.address().port}`; + for (const path of ["/", "/unknown", "/api/easel-musical-stream?unexpected=1"]) { + const status = await new Promise(resolve => { + const ws = new WebSocket(base + path); + const timer = setTimeout(() => { ws.terminate(); resolve("timeout"); }, 500); + ws.on("error", () => {}); + ws.on("unexpected-response", (_req, res) => { clearTimeout(timer); res.resume(); ws.terminate(); resolve(res.statusCode); }); + }); + assert.equal(status, 404, path); + } + for (const path of ["/api/easel-musical-stream", "/api/whistlegraph-stream", "/api/walkieware-stream"]) { + const ws = new WebSocket(base + path); + await once(ws, "open"); ws.close(); await once(ws, "close"); + } + musical.close(); + const ws = new WebSocket(base + "/api/whistlegraph-stream"); + await once(ws, "open"); ws.close(); await once(ws, "close"); + assert.equal(await (await fetch(base.replace("ws:", "http:") + "/")).text(), "ok"); +}); diff --git a/lith/whistlegraph-socket.mjs b/lith/whistlegraph-socket.mjs index 1368168c86..b34f604bd3 100644 --- a/lith/whistlegraph-socket.mjs +++ b/lith/whistlegraph-socket.mjs @@ -1,4 +1,5 @@ import {WebSocketServer,WebSocket} from 'ws'; +import {attachSocketRoute} from './socket-routing.mjs'; import {randomUUID} from 'node:crypto'; import {publicThread,sourceHash} from '../system/backend/whistlegraph.mjs'; @@ -7,12 +8,11 @@ export function attachWhistlegraphSocket(server,{authenticate,store,authMs=5000, const rooms=new Map(); const send=(ws,value)=>{if(ws?.readyState===WebSocket.OPEN){if(ws.bufferedAmount>8_100_000)ws.close(1008,'Slow reader');else ws.send(JSON.stringify(value));}}; const upgrade=(req,socket,head)=>{ - // Older installed clients retain their authenticated connection route. - if(!['/api/whistlegraph-stream','/api/walkieware-stream'].includes(req.url))return; if(wss.clients.size>=64){socket.destroy();return;} wss.handleUpgrade(req,socket,head,ws=>wss.emit('connection',ws)); }; - server.on('upgrade',upgrade); + // Older installed clients retain their authenticated connection route. + const detach=attachSocketRoute(server,['/api/whistlegraph-stream','/api/walkieware-stream'],upgrade); wss.on('connection',ws=>{ let owner,row,room,role,closed=false,chain=Promise.resolve(),pending=0,alive=true,messages=0,windowAt=Date.now(); const authTimer=setTimeout(()=>ws.close(1008,'Authenticate first'),authMs); @@ -89,5 +89,5 @@ export function attachWhistlegraphSocket(server,{authenticate,store,authMs=5000, if(room&&!room.clients.size)rooms.delete(row._id); }); }); - return {wss,close(){server.off('upgrade',upgrade);for(const ws of wss.clients)ws.terminate();wss.close();}}; + return {wss,close(){detach();for(const ws of wss.clients)ws.terminate();wss.close();}}; } diff --git a/system/netlify/functions/painting-metadata.mjs b/system/netlify/functions/painting-metadata.mjs index 27205c6767..e3feb71347 100644 --- a/system/netlify/functions/painting-metadata.mjs +++ b/system/netlify/functions/painting-metadata.mjs @@ -5,12 +5,9 @@ // Single painting: ?slug=xxx&handle=yyy (optional) // Gallery list: ?limit=20&sort=recent&offset=0 (no slug parameter) -import { MongoClient } from "mongodb"; +import { connect } from "../../backend/database.mjs"; import { userPaintingSlugQuery } from "../../backend/painting-slug-query.mjs"; -const client = new MongoClient(process.env.MONGODB_CONNECTION_STRING); -const dbName = process.env.MONGODB_NAME || "aesthetic"; - function respond(statusCode, body) { return { statusCode, @@ -35,8 +32,7 @@ export async function handler(event) { // If no slug, return list of paintings (for gallery) if (!slug) { try { - await client.connect(); - const db = client.db(dbName); + const { db } = await connect(); const paintings = db.collection("paintings"); const users = db.collection("users"); @@ -103,15 +99,12 @@ export async function handler(event) { } catch (error) { console.error("Error fetching paintings list:", error); return respond(500, { error: "Internal server error" }); - } finally { - await client.close(); } } // Original single painting lookup by slug try { - await client.connect(); - const db = client.db(dbName); + const { db } = await connect(); const paintings = db.collection("paintings"); // Build query @@ -157,7 +150,5 @@ export async function handler(event) { } catch (error) { console.error("Error fetching painting metadata:", error); return respond(500, { error: "Internal server error" }); - } finally { - await client.close(); } } diff --git a/system/tests/production-endpoints.test.mjs b/system/tests/production-endpoints.test.mjs index 2479bc3a7d..d9dbad3890 100644 --- a/system/tests/production-endpoints.test.mjs +++ b/system/tests/production-endpoints.test.mjs @@ -30,6 +30,36 @@ async function load(name, { env = {}, mocks = {}, ...options } = {}) { return module.namespace.handler; } +for (const faster of ["lookup", "gallery"]) test(`painting metadata ${faster} completion preserves another in-flight lookup`, async () => { + let closed = false, entered, release; + const started = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const db = { collection: () => ({ + async findOne(query) { + if (query.$or?.[0]?.slug === "slow") { entered(); await gate; } + if (closed) throw Error("Cannot use a session that has ended"); + return { slug: "fixture", code: "fixture" }; + }, + find() { return { sort() { return this; }, skip() { return this; }, limit() { return this; }, async toArray() { return []; } }; }, + }) }; + class MongoClient { + async connect() { closed = false; } + db() { return db; } + async close() { closed = true; } + } + const handler = await load("painting-metadata.mjs", { mocks: { + mongodb: { MongoClient }, + "../../backend/database.mjs": { connect: async () => ({ db }) }, + "../../backend/painting-slug-query.mjs": { userPaintingSlugQuery: slug => ({ slug }) }, + } }); + const pending = handler({ httpMethod: "GET", queryStringParameters: { slug: "slow" } }); + await started; + const first = await handler({ httpMethod: "GET", queryStringParameters: faster === "lookup" ? { slug: "fast" } : {} }); + release(); + assert.equal(first.statusCode, 200); + assert.equal((await pending).statusCode, 200); +}); + test("gives constructs Stripe and returns its feed, then uses the cache", async () => { let reads = 0; class Stripe {