From d9833c17d5df1224c8a9634ff354323bdf7d767b Mon Sep 17 00:00:00 2001 From: whyour Date: Fri, 18 Sep 2026 20:22:18 +0800 Subject: [PATCH] perf: reduce resident dependencies and duplicate WebSocket authentication (#3071) * perf: load HTTP dependencies only in the HTTP worker * docs: record 30-minute idle resource comparison * perf: isolate database initialization from the cluster primary * docs: record isolated database startup resource comparison * perf: avoid an extra shell child in Docker health probes * perf: load notification dependencies only when needed * perf: retain queue cleanup after notification loading comparison * fix: stop page polling after unmount and avoid duplicate fetches * perf: share WebSocket session polling across connections * docs: measure real WebSocket container CPU and memory * perf: reuse JWT validation within each WebSocket session check * perf: bound login IP lookup allocations with an address cache * refactor: limit resource optimization PR to three core changes * docs: benchmark the three core optimizations against develop * chore: remove documentation and benchmark artifacts from PR --- back/app.ts | 62 +++--- back/bootstrap/database.ts | 26 +++ back/config/util.ts | 6 +- back/loaders/sock.ts | 80 +++++-- back/shared/errors.ts | 4 + back/shared/startupProcess.ts | 47 ++++ test/back/database-bootstrap.test.cjs | 193 +++++++++++++++++ test/back/role-loading.test.cjs | 118 ++++++++++ test/back/sock-polling.test.cjs | 297 ++++++++++++++++++++++++++ 9 files changed, 779 insertions(+), 54 deletions(-) create mode 100644 back/bootstrap/database.ts create mode 100644 back/shared/errors.ts create mode 100644 back/shared/startupProcess.ts create mode 100644 test/back/database-bootstrap.test.cjs create mode 100644 test/back/role-loading.test.cjs create mode 100644 test/back/sock-polling.test.cjs diff --git a/back/app.ts b/back/app.ts index 693e34e1..96276014 100644 --- a/back/app.ts +++ b/back/app.ts @@ -1,17 +1,12 @@ import 'reflect-metadata'; import cluster, { type Worker } from 'cluster'; -import compression from 'compression'; -import cors from 'cors'; -import express from 'express'; -import helmet from 'helmet'; +import type express from 'express'; import { Container } from 'typedi'; import config from './config'; import Logger from './loaders/logger'; -import { monitoringMiddleware } from './middlewares/monitoring'; -import { errStack } from './config/util'; +import { errStack } from './shared/errors'; import { type GrpcServerService } from './services/grpc'; import { type HttpServerService } from './services/http'; -import cronClient from './schedule/client'; interface WorkerMetadata { id: number; @@ -21,24 +16,12 @@ interface WorkerMetadata { } class Application { - private app: express.Application; private httpServerService?: HttpServerService; private grpcServerService?: GrpcServerService; private isShuttingDown = false; private workerMetadataMap = new Map(); private httpWorker?: Worker; - constructor() { - this.app = express(); - // 创建一个全局中间件,删除查询参数中的t - this.app.use((req: express.Request, res: express.Response, next: express.NextFunction) => { - if (req.query.t) { - delete req.query.t; - } - next(); - }); - } - async start() { try { if (cluster.isPrimary) { @@ -158,17 +141,37 @@ class Application { } private async initializeDatabase() { - const dbLoader = await import('./loaders/db'); - await dbLoader.default(); + const { runStartupProcess } = await import('./shared/startupProcess'); + await runStartupProcess(require.resolve('./bootstrap/database')); } - private setupMiddlewares() { - this.app.use(helmet({ + private async setupMiddlewares(): Promise { + // Only the HTTP worker needs an Express application and its middleware. + const [ + { default: createExpress }, + { default: helmet }, + { default: cors }, + { default: compression }, + { monitoringMiddleware }, + ] = await Promise.all([ + import('express'), + import('helmet'), + import('cors'), + import('compression'), + import('./middlewares/monitoring'), + ]); + const app = createExpress(); + app.use((req, res, next) => { + if (req.query.t) delete req.query.t; + next(); + }); + app.use(helmet({ contentSecurityPolicy: false, })); - this.app.use(cors(config.cors)); - this.app.use(compression()); - this.app.use(monitoringMiddleware); + app.use(cors(config.cors)); + app.use(compression()); + app.use(monitoringMiddleware); + return app; } private setupMasterShutdown() { @@ -247,16 +250,16 @@ class Application { const { initGrpcCerts } = await import('./config/grpcCerts'); await initGrpcCerts(); - this.setupMiddlewares(); + const app = await this.setupMiddlewares(); const { HttpServerService } = await import('./services/http'); this.httpServerService = Container.get(HttpServerService); const appLoader = await import('./loaders/app'); - await appLoader.default({ app: this.app }); + await appLoader.default({ app }); const server = await this.httpServerService.initialize( - this.app, + app, config.port, ); @@ -279,6 +282,7 @@ class Application { this.gracefulShutdown(serviceType); } else if (serviceType === 'http' && (msg === 'reregister-crons' || msg === 'scheduler-unavailable')) { + const { default: cronClient } = await import('./schedule/client'); cronClient.readiness.invalidate(); } }); diff --git a/back/bootstrap/database.ts b/back/bootstrap/database.ts new file mode 100644 index 00000000..4fdeb4d0 --- /dev/null +++ b/back/bootstrap/database.ts @@ -0,0 +1,26 @@ +import { errStack } from '../shared/errors'; + +// Also terminate if the primary disappears without forwarding a signal. +const onParentDisconnect = () => process.exit(1); +process.once('disconnect', onParentDisconnect); + +async function initializeDatabase() { + const { sequelize } = await import('../data'); + try { + const { default: loadDatabase } = await import('../loaders/db'); + await loadDatabase(); + } finally { + await sequelize.close(); + } +} + +initializeDatabase() + .catch((error) => { + console.error(`[boot] Database initialization failed:\n${errStack(error)}`); + process.exitCode = 1; + }) + .finally(() => { + process.removeListener('disconnect', onParentDisconnect); + // Let pending log writes drain and exit naturally after closing SQLite. + if (process.connected) process.disconnect(); + }); diff --git a/back/config/util.ts b/back/config/util.ts index dbd2f2a1..13cad395 100644 --- a/back/config/util.ts +++ b/back/config/util.ts @@ -652,11 +652,7 @@ export function safeJSONParse(value?: string) { } } -export function errStack(error: unknown): string { - return error instanceof Error && error.stack - ? error.stack - : String(error); -} +export { errStack } from '../shared/errors'; export async function rmPath(path: string) { try { diff --git a/back/loaders/sock.ts b/back/loaders/sock.ts index b99b721f..b09d6aeb 100644 --- a/back/loaders/sock.ts +++ b/back/loaders/sock.ts @@ -1,4 +1,4 @@ -import sockJs from 'sockjs'; +import sockJs, { Connection } from 'sockjs'; import { Server } from 'http'; import { Container } from 'typedi'; import SockService from '../services/sock'; @@ -13,6 +13,41 @@ export default async ({ server }: { server: Server }) => { log: () => {}, }); const sockService = Container.get(SockService); + const sessions = new Map(); + let sessionTimer: ReturnType | undefined; + let checking = false; + + const checkSessions = async () => { + if (checking || sessions.size === 0) return; + checking = true; + // Connections accepted during this read must not use an older snapshot. + const batch = [...sessions]; + try { + const current = await shareStore.getAuthInfo(); + // Reuse validation only within this synchronous check of one auth snapshot. + const validated = new Map>(); + for (const [conn, { token, platform }] of batch) { + if (!sessions.has(conn)) continue; + let platforms = validated.get(token); + if (!platforms) { + platforms = new Map(); + validated.set(token, platforms); + } + let valid = platforms.get(platform); + if (valid === undefined) { + valid = isValidToken(current, token, platform, config.jwt.secret); + platforms.set(platform, valid); + } + if (!valid) conn.close('401'); + } + } catch { + for (const [conn] of batch) { + if (sessions.has(conn)) conn.close('401'); + } + } finally { + checking = false; + } + }; echo.on('connection', async (conn) => { if (!conn.headers || !conn.url || !conn.pathname) { @@ -20,35 +55,40 @@ export default async ({ server }: { server: Server }) => { return; } - const authInfo = await shareStore.getAuthInfo(); + let closed = false; + conn.on('close', () => { + closed = true; + sessions.delete(conn); + sockService.removeClient(conn); + if (sessions.size === 0 && sessionTimer) { + clearInterval(sessionTimer); + sessionTimer = undefined; + } + }); + + let authInfo; + try { + authInfo = await shareStore.getAuthInfo(); + } catch { + if (!closed) conn.close('401'); + return; + } + if (closed) return; const platform = getPlatform(conn.headers['user-agent'] || '') || 'desktop'; const headerToken = conn.url.replace(`${conn.pathname}?token=`, ''); if (isValidToken(authInfo, headerToken, platform, config.jwt.secret)) { sockService.addClient(conn); - const checkSession = setInterval(async () => { - try { - const current = await shareStore.getAuthInfo(); - if ( - !isValidToken(current, headerToken, platform, config.jwt.secret) - ) { - conn.close('401'); - } - } catch { - conn.close('401'); - } - }, 1000); - checkSession.unref(); + sessions.set(conn, { token: headerToken, platform }); + if (!sessionTimer) { + sessionTimer = setInterval(checkSessions, 1000); + sessionTimer.unref(); + } conn.on('data', (message) => { conn.write(message); }); - conn.on('close', function () { - clearInterval(checkSession); - sockService.removeClient(conn); - }); - return; } diff --git a/back/shared/errors.ts b/back/shared/errors.ts new file mode 100644 index 00000000..72945c9b --- /dev/null +++ b/back/shared/errors.ts @@ -0,0 +1,4 @@ +// Keep error formatting independent of database, HTTP and gRPC dependencies. +export function errStack(error: unknown): string { + return error instanceof Error && error.stack ? error.stack : String(error); +} diff --git a/back/shared/startupProcess.ts b/back/shared/startupProcess.ts new file mode 100644 index 00000000..64ee9dcb --- /dev/null +++ b/back/shared/startupProcess.ts @@ -0,0 +1,47 @@ +import { fork } from 'child_process'; + +// Keep one-off startup dependencies out of the long-lived cluster primary. +export function runStartupProcess(entrypoint: string): Promise { + return new Promise((resolve, reject) => { + const child = fork(entrypoint, [], { stdio: 'inherit' }); + let interrupted: NodeJS.Signals | undefined; + let killTimer: NodeJS.Timeout | undefined; + + const terminate = (signal: NodeJS.Signals) => { + if (interrupted) return; + interrupted = signal; + child.kill(signal); + killTimer = setTimeout(() => child.kill('SIGKILL'), 8000); + killTimer.unref(); + }; + const onSigterm = () => terminate('SIGTERM'); + const onSigint = () => terminate('SIGINT'); + const onExit = () => child.kill('SIGKILL'); + const cleanup = () => { + if (killTimer) clearTimeout(killTimer); + process.removeListener('SIGTERM', onSigterm); + process.removeListener('SIGINT', onSigint); + process.removeListener('exit', onExit); + }; + + process.once('SIGTERM', onSigterm); + process.once('SIGINT', onSigint); + process.once('exit', onExit); + child.once('error', (error) => { + cleanup(); + reject(error); + }); + child.once('close', (code, signal) => { + cleanup(); + if (code === 0 && !interrupted) { + resolve(); + } else { + reject( + new Error( + `Startup process failed (${interrupted || signal || code})`, + ), + ); + } + }); + }); +} diff --git a/test/back/database-bootstrap.test.cjs b/test/back/database-bootstrap.test.cjs new file mode 100644 index 00000000..d3507968 --- /dev/null +++ b/test/back/database-bootstrap.test.cjs @@ -0,0 +1,193 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { fork } = require('node:child_process'); +const { once } = require('node:events'); +const { Sequelize, QueryTypes } = require('sequelize'); +const ts = require('typescript'); +const { createRequire } = require('node:module'); + +const initializer = require.resolve('../../back/bootstrap/database'); +const startup = require.resolve('../../back/shared/startupProcess'); + +function fixture(t) { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ql-bootstrap-')); + t.after(() => fs.rmSync(dir, { recursive: true, force: true })); + fs.mkdirSync(path.join(dir, 'data/db'), { recursive: true }); + fs.mkdirSync(path.join(dir, 'data/syslog'), { recursive: true }); + fs.writeFileSync(path.join(dir, '.env'), ''); + const wrapper = path.join(dir, 'parent.cjs'); + fs.writeFileSync(wrapper, ` + const cp = require('node:child_process'); + const original = cp.fork; + cp.fork = (...args) => { + const child = original(...args); + process.send?.({ childPid: child.pid }); + return child; + }; + const { runStartupProcess } = require(${JSON.stringify(startup)}); + runStartupProcess(process.argv[2]).then(() => { + const modules = Object.keys(require.cache); + process.send?.({ done: true, modules }); + }).catch(error => { + console.error(error.message); + process.exitCode = 1; + }).finally(() => { if (process.connected) process.disconnect(); }); + `); + return { + dir, + storage: path.join(dir, 'data/db/database.sqlite'), + run(entrypoint = initializer) { + const child = fork(wrapper, [entrypoint], { + env: { ...process.env, QL_DIR: dir, QL_DATA_DIR: path.join(dir, 'data') }, + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + }); + const messages = []; + let output = ''; + child.on('message', (message) => messages.push(message)); + child.stdout.on('data', (data) => { output += data; }); + child.stderr.on('data', (data) => { output += data; }); + t.after(() => { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); + }); + const done = once(child, 'close').then(([code, signal]) => ({ code, signal, output, messages })); + return { child, done, messages }; + }, + }; +} + +async function waitFor(predicate) { + const deadline = Date.now() + 10000; + while (!predicate()) { + assert.ok(Date.now() < deadline, 'timed out waiting for child'); + await new Promise(resolve => setTimeout(resolve, 20)); + } +} + +test('isolated initialization creates a fresh database and does not retain ORM modules in the parent', { timeout: 20000 }, async t => { + const f = fixture(t); + const result = await f.run().done; + assert.equal(result.code, 0, result.output); + const done = result.messages.find(message => message.done); + assert.ok(done); + assert.equal(done.modules.some(file => /[\\/]sequelize[\\/]|[\\/]data[\\/]index\.ts$/.test(file)), false); + const db = new Sequelize({ dialect: 'sqlite', storage: f.storage, logging: false }); + try { + const tables = await db.getQueryInterface().showAllTables(); + for (const table of ['Crontabs', 'Envs', 'RunningInstances', 'SchemaMigrations']) assert.ok(tables.includes(table), table); + assert.equal((await db.query('SELECT id FROM SchemaMigrations', { type: QueryTypes.SELECT })).length, 15); + } finally { await db.close(); } +}); + +test('isolated initialization upgrades an existing database and remains idempotent', { timeout: 30000 }, async t => { + const f = fixture(t); + let db = new Sequelize({ dialect: 'sqlite', storage: f.storage, logging: false }); + await db.query('CREATE TABLE Crontabs (id INTEGER PRIMARY KEY, name TEXT)'); + await db.query("INSERT INTO Crontabs (id, name) VALUES (1, 'keep-me')"); + await db.close(); + for (let i = 0; i < 2; i++) { + const result = await f.run().done; + assert.equal(result.code, 0, result.output); + } + db = new Sequelize({ dialect: 'sqlite', storage: f.storage, logging: false }); + try { + const rows = await db.query('SELECT * FROM Crontabs', { type: QueryTypes.SELECT }); + assert.equal(rows[0].name, 'keep-me'); + assert.ok(Object.hasOwn(rows[0], 'queued_token')); + assert.equal((await db.query('SELECT id FROM SchemaMigrations', { type: QueryTypes.SELECT })).length, 15); + } finally { await db.close(); } +}); + +test('database errors and missing initializer files reject startup', { timeout: 20000 }, async t => { + const f = fixture(t); + fs.writeFileSync(f.storage, 'invalid SQLite database'); + const invalid = await f.run().done; + assert.equal(invalid.code, 1); + assert.match(invalid.output, /SQLITE_NOTADB/); + assert.equal(invalid.messages.some(message => message.done), false); + const missing = await f.run(path.join(f.dir, 'missing.cjs')).done; + assert.equal(missing.code, 1); + assert.match(missing.output, /MODULE_NOT_FOUND/); +}); + +for (const signal of ['SIGTERM', 'SIGINT']) { + test(`startup ${signal} terminates the initializer and never reports success`, { timeout: 15000 }, async t => { + const f = fixture(t); + const target = path.join(f.dir, 'waiting.cjs'); + const ready = path.join(f.dir, 'ready'); + fs.writeFileSync(target, ` + require('node:fs').writeFileSync(${JSON.stringify(ready)}, String(process.pid)); + setInterval(() => {}, 1000); + `); + const running = f.run(target); + await waitFor(() => fs.existsSync(ready)); + const pid = Number(fs.readFileSync(ready, 'utf8')); + running.child.kill(signal); + const result = await running.done; + assert.equal(result.code, 1, result.output); + assert.match(result.output, new RegExp(signal)); + assert.equal(result.messages.some(message => message.done), false); + assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' }); + }); +} + +test('a database initializer blocked on SQLite exits when its parent disappears', { timeout: 20000 }, async t => { + const f = fixture(t); + const db = new Sequelize({ dialect: 'sqlite', storage: f.storage, logging: false }); + await db.query('BEGIN EXCLUSIVE'); + try { + const running = f.run(); + await waitFor(() => running.messages.some(message => message.childPid)); + const pid = running.messages.find(message => message.childPid).childPid; + // Allow the real initializer to enter the database operation before losing IPC. + await new Promise(resolve => setTimeout(resolve, 1000)); + running.child.kill('SIGKILL'); + await running.done; + await waitFor(() => { + try { process.kill(pid, 0); return false; } + catch (error) { if (error.code === 'ESRCH') return true; throw error; } + }); + } finally { + await db.query('ROLLBACK'); + await db.close(); + } +}); + +test('application waits for isolated initialization and does not fork workers after a failure', async () => { + const file = path.resolve('back/app.ts'); + const source = fs.readFileSync(file, 'utf8'); + const compiled = ts.transpileModule(source.slice(0, source.indexOf('\nconst app = new Application();')) + '\nmodule.exports = Application;', { + compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020, esModuleInterop: true }, + }).outputText; + for (const fail of [false, true]) { + let complete; + const pending = new Promise((resolve, reject) => { complete = () => fail ? reject(new Error('migration failed')) : resolve(); }); + const exits = [], calls = []; + const localRequire = createRequire(file); + const mocks = { + cluster: { isPrimary: true }, + './config': {}, + './loaders/logger': { error() {} }, + './shared/startupProcess': { runStartupProcess: entry => { assert.equal(entry, initializer); return pending; } }, + }; + const req = Object.assign(name => { + calls.push(name); + return Object.hasOwn(mocks, name) ? mocks[name] : localRequire(name); + }, { resolve: localRequire.resolve }); + const module = { exports: {} }; + new Function('require', 'module', 'exports', 'process', compiled)(req, module, module.exports, { exit: code => exits.push(code) }); + const app = new module.exports(); + let started = false; + app.startMasterProcess = () => { started = true; }; + const start = app.start(); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(started, false); + complete(); + await start; + assert.equal(started, !fail); + assert.deepEqual(exits, fail ? [1] : []); + assert.equal(calls.includes('./loaders/db'), false); + } +}); diff --git a/test/back/role-loading.test.cjs b/test/back/role-loading.test.cjs new file mode 100644 index 00000000..ac59c8bb --- /dev/null +++ b/test/back/role-loading.test.cjs @@ -0,0 +1,118 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const { createRequire } = require('node:module'); +const path = require('node:path'); +const ts = require('typescript'); +const source = fs.readFileSync('back/app.ts', 'utf8'); +const compiled = ts.transpileModule( + source.slice(0, source.indexOf('\nconst app = new Application();')) + + '\nmodule.exports = Application;', + { + compilerOptions: { + module: ts.ModuleKind.CommonJS, + target: ts.ScriptTarget.ES2020, + esModuleInterop: true, + }, + }, +).outputText; + +function fixture() { + const required = [], + listeners = new Map(); + let invalidations = 0; + const module = { exports: {} }; + const localRequire = createRequire(path.resolve('back/app.ts')); + const mocks = { + './config': { cors: { origin: ['http://localhost'] } }, + './loaders/logger': { error() {}, warn() {}, info() {} }, + './middlewares/monitoring': { + monitoringMiddleware: (_req, _res, next) => next(), + }, + './schedule/client': { + default: { readiness: { invalidate: () => invalidations++ } }, + __esModule: true, + }, + }; + new Function('require', 'module', 'exports', 'process', compiled)( + (name) => { + required.push(name); + return Object.hasOwn(mocks, name) ? mocks[name] : localRequire(name); + }, + module, + module.exports, + { env: {}, on: (name, fn) => listeners.set(name, fn) }, + ); + return { + app: new module.exports(), + required, + listeners, + invalidations: () => invalidations, + }; +} + +test('primary construction does not load HTTP middleware, metrics or the gRPC client', () => { + const { required } = fixture(); + for (const name of [ + 'express', + 'helmet', + 'cors', + 'compression', + './middlewares/monitoring', + './schedule/client', + './config/util', + ]) { + assert.equal(required.includes(name), false, name); + } +}); + +test('lazy HTTP initialization preserves query filtering, security headers, CORS and compression', async (t) => { + const { app, required } = fixture(); + const http = await app.setupMiddlewares(); + http.get('/test', (req, res) => + res.json({ query: req.query, data: 'x'.repeat(2048) }), + ); + const server = await new Promise((resolve) => { + const server = http.listen(0, '127.0.0.1', () => resolve(server)); + }); + t.after( + () => + new Promise((resolve) => { + server.close(resolve); + server.closeAllConnections(); + }), + ); + const response = await fetch( + `http://127.0.0.1:${server.address().port}/test?t=123&keep=yes`, + { + headers: { Origin: 'http://localhost', 'Accept-Encoding': 'gzip' }, + }, + ); + assert.equal(response.status, 200); + assert.equal(response.headers.get('x-content-type-options'), 'nosniff'); + assert.equal( + response.headers.get('access-control-allow-origin'), + 'http://localhost', + ); + assert.equal(response.headers.get('content-encoding'), 'gzip'); + assert.deepEqual((await response.json()).query, { keep: 'yes' }); + assert.ok(required.includes('express')); + assert.equal(required.includes('./schedule/client'), false); +}); + +test('worker recovery messages still invalidate HTTP readiness, but gRPC shutdown setup does not load its client', async () => { + const grpc = fixture(); + grpc.app.setupWorkerShutdown('grpc'); + await grpc.listeners.get('message')('scheduler-unavailable'); + assert.equal(grpc.required.includes('./schedule/client'), false); + const http = fixture(); + http.app.setupWorkerShutdown('http'); + await http.listeners.get('message')('scheduler-unavailable'); + await http.listeners.get('message')('reregister-crons'); + assert.equal(http.invalidations(), 2); + const shutdowns = []; + http.app.gracefulShutdown = (role) => shutdowns.push(role); + await http.listeners.get('message')('shutdown'); + http.listeners.get('SIGTERM')(); + assert.deepEqual(shutdowns, ['http', 'http']); +}); diff --git a/test/back/sock-polling.test.cjs b/test/back/sock-polling.test.cjs new file mode 100644 index 00000000..1acb084d --- /dev/null +++ b/test/back/sock-polling.test.cjs @@ -0,0 +1,297 @@ +const assert = require('node:assert/strict'); +const test = require('node:test'); +const fs = require('node:fs'); +const path = require('node:path'); +const { EventEmitter } = require('node:events'); +const ts = require('typescript'); +const jwt = require('jsonwebtoken'); +const { isValidToken } = require('../../back/shared/auth'); + +async function setup() { + const secret = 'polling-test'; + const token = jwt.sign({}, secret, { algorithm: 'HS384', expiresIn: '1h' }); + let auth = { token }; + let read = async () => auth; + let reads = 0; + const validations = []; + let onConnection; + const clients = new Set(); + const timers = new Set(); + const mocks = { + sockjs: { + createServer: () => ({ + on: (_, fn) => { + onConnection = fn; + }, + installHandlers() {}, + }), + }, + typedi: { + Container: { + get: () => ({ + addClient: (c) => clients.add(c), + removeClient: (c) => clients.delete(c), + }), + }, + }, + '../services/sock': class {}, + '../config/util': { getPlatform: (agent) => agent || 'desktop' }, + '../shared/store': { + shareStore: { + getAuthInfo: () => { + reads++; + return read(); + }, + }, + }, + '../shared/auth': { + isValidToken: (...args) => { + validations.push({ token: args[1], platform: args[2] }); + return isValidToken(...args); + }, + }, + '../config': { baseUrl: '', jwt: { secret } }, + }; + const source = fs.readFileSync( + path.join(__dirname, '../../back/loaders/sock.ts'), + 'utf8', + ); + const { outputText } = ts.transpileModule(source, { + compilerOptions: { + module: ts.ModuleKind.CommonJS, + target: ts.ScriptTarget.ES2020, + esModuleInterop: true, + }, + }); + const module = { exports: {} }; + new Function( + 'require', + 'module', + 'exports', + 'setInterval', + 'clearInterval', + outputText, + )( + (name) => { + assert.ok(Object.hasOwn(mocks, name), name); + return mocks[name]; + }, + module, + module.exports, + (callback, delay) => { + assert.equal(delay, 1000); + const timer = { callback, unref() {} }; + timers.add(timer); + return timer; + }, + (timer) => timers.delete(timer), + ); + await module.exports.default({ server: {} }); + function connection(value = token) { + const conn = new EventEmitter(); + Object.assign(conn, { + headers: {}, + pathname: '/api/ws/a/b/websocket', + url: `/api/ws/a/b/websocket?token=${value}`, + write() {}, + close(code) { + this.closeCode = code; + this.emit('close'); + }, + }); + return conn; + } + return { + token, + secret, + clients, + timers, + validations, + connection, + connect: (c) => onConnection(c), + get reads() { + return reads; + }, + setAuth(value) { + auth = value; + }, + setRead(fn) { + read = fn; + }, + tick: () => Promise.all([...timers].map((t) => t.callback())), + }; +} + +function deferred() { + let resolve, reject; + const promise = new Promise((a, b) => { + resolve = a; + reject = b; + }); + return { promise, resolve, reject }; +} + +test('multiple sockets share one periodic read, revoke together, and stop polling when empty', async () => { + const h = await setup(); + assert.equal(h.timers.size, 0); + const sockets = Array.from({ length: 20 }, () => h.connection()); + await Promise.all(sockets.map(h.connect)); + assert.equal(h.reads, 20, 'initial authentication remains per connection'); + assert.equal(h.timers.size, 1); + await h.tick(); + assert.equal(h.reads, 21); + assert.equal(h.clients.size, 20); + h.setAuth({ token: '' }); + await h.tick(); + assert.ok(sockets.every((c) => c.closeCode === '401')); + assert.equal(h.clients.size, 0); + assert.equal(h.timers.size, 0); + const c = h.connection(); + h.setAuth({ token: h.token }); + await h.connect(c); + assert.equal(h.timers.size, 1); + c.close(); + assert.equal(h.timers.size, 0); +}); + +test('slow reads do not overlap and old snapshots do not reject newly accepted sockets', async () => { + const h = await setup(); + const old = h.connection(); + await h.connect(old); + const pending = deferred(); + h.setRead(() => pending.promise); + const tick = h.tick(); + await h.tick(); + assert.equal(h.reads, 2); + old.close(); + h.setRead(async () => ({ token: h.token })); + const fresh = h.connection(); + await h.connect(fresh); + pending.resolve({ token: '' }); + await tick; + assert.equal(h.clients.has(fresh), true); + assert.equal(fresh.closeCode, undefined); + await h.tick(); + assert.equal(h.reads, 4); + fresh.close(); +}); + +test('periodic store failures disconnect the checked sessions', async () => { + const h = await setup(); + const c = h.connection(); + await h.connect(c); + h.setRead(async () => { + throw new Error('database unavailable'); + }); + await h.tick(); + assert.equal(c.closeCode, '401'); + assert.equal(h.clients.size, 0); + assert.equal(h.timers.size, 0); +}); + +test('closed connections cannot be added after initial authentication finishes', async () => { + const h = await setup(); + const pending = deferred(); + h.setRead(() => pending.promise); + const c = h.connection(); + const connecting = h.connect(c); + c.close(); + pending.resolve({ token: h.token }); + await connecting; + assert.equal(h.clients.size, 0); + assert.equal(h.timers.size, 0); +}); + +test('initial store errors reject the connection without creating a timer', async () => { + const h = await setup(); + h.setRead(async () => { + throw new Error('database unavailable'); + }); + const c = h.connection(); + await h.connect(c); + assert.equal(c.closeCode, '401'); + assert.equal(h.clients.size, 0); + assert.equal(h.timers.size, 0); +}); + +test('each periodic check still validates JWT expiry and platform session membership', async () => { + const h = await setup(); + const expired = h.connection( + jwt.sign({}, h.secret, { algorithm: 'HS384', expiresIn: -1 }), + ); + await h.connect(expired); + assert.equal(expired.closeCode, '404'); + const c = h.connection(); + await h.connect(c); + h.setAuth({ tokens: { desktop: [{ value: h.token, expiration: 1 }] } }); + await h.tick(); + assert.equal(c.closeCode, '401'); + assert.equal(h.timers.size, 0); +}); + +test('an accepted JWT is disconnected when it expires without a store change', async () => { + const h = await setup(); + const c = h.connection(); + await h.connect(c); + const now = Date.now; + try { + Date.now = () => now() + 2 * 60 * 60 * 1000; + await h.tick(); + } finally { + Date.now = now; + } + assert.equal(c.closeCode, '401'); + assert.equal(h.timers.size, 0); +}); + +test('same token and platform share validation only within the current round', async () => { + const h = await setup(); + const sockets = Array.from({ length: 20 }, () => h.connection()); + await Promise.all(sockets.map(h.connect)); + assert.equal( + h.validations.length, + 20, + 'initial authentication stays independent', + ); + h.validations.length = 0; + await h.tick(); + assert.equal(h.validations.length, 1); + await h.tick(); + assert.equal(h.validations.length, 2, 'new rounds revalidate'); + h.setAuth({ token: '' }); + await h.tick(); + assert.equal( + h.validations.length, + 3, + 'invalid results are also shared for this round', + ); + assert.ok(sockets.every((c) => c.closeCode === '401')); + assert.equal(h.timers.size, 0); +}); + +test('validation results never cross tokens or platforms', async () => { + const h = await setup(); + const other = jwt.sign({ account: 'other' }, h.secret, { + algorithm: 'HS384', + expiresIn: '1h', + }); + h.setAuth({ + tokens: { + desktop: [{ value: h.token }, { value: other }], + mobile: [{ value: h.token }], + }, + }); + const desktop = h.connection(); + const mobile = h.connection(); + mobile.headers['user-agent'] = 'mobile'; + const separate = h.connection(other); + await Promise.all([desktop, mobile, separate].map(h.connect)); + h.validations.length = 0; + h.setAuth({ tokens: { desktop: [{ value: h.token }], mobile: [] } }); + await h.tick(); + assert.equal(h.validations.length, 3); + assert.equal(desktop.closeCode, undefined); + assert.equal(mobile.closeCode, '401'); + assert.equal(separate.closeCode, '401'); + desktop.close(); +});