import assert from 'node:assert/strict'; import { readFileSync } from 'node:fs'; import { test } from 'node:test'; const source = readFileSync(new URL('../../public/static/js/realtime-probe.js', import.meta.url)); const { RealtimeProbe, ProbeError, apiAddress, apiEndpoint, redact } = await import(`data:text/javascript;base64,${source.toString('base64')}`); const A = 'admin.orders.withdraw'; const B = 'admin.users.online'; const flush = () => new Promise((resolve) => setImmediate(resolve)); const event = (seq, channel = A) => ({ seq, id: seq, channel, event: channel === A ? 'order.updated' : 'user.online', action: channel === A ? 'updated' : 'online', key: { id: 88 }, data: { status: 1 } }); const config = { code: 0, data: { key: 'fixture-key', ws_host: 'localhost', ws_port: 8080, force_tls: false, channels: [A, B, 'admin.stats.platform_count', 'admin.stats.unprocessed'], last_seq: 0, session_expires_at: 1900000000, auth_endpoint: '/admin/broadcasting/auth' } }; class Subscription { constructor() { this.listeners = new Map(); } listen(name, callback) { this.listeners.set(name, callback); return this; } subscribed(callback) { this.ready = callback; return this; } error(callback) { this.failed = callback; return this; } emit(message) { this.listeners.get(`.${message.event}`)?.(message); } } class Echo { constructor(options) { this.options = options; this.channels = new Map(); this.id = '1.1'; this.connector = { pusher: { connection: { bind: (name, callback) => { this[name] = callback; } } } }; } private(name) { if (!this.channels.has(name)) this.channels.set(name, new Subscription()); return this.channels.get(name); } leave(name) { this.channels.delete(name); } socketId() { return this.id; } disconnect() { this.state_change?.({ current: 'disconnected' }); } } const deferred = () => { let resolve; const promise = new Promise((done) => { resolve = done; }); return { resolve, promise }; }; async function setup(request) { const logs = [], events = [], requests = []; const probe = new RealtimeProbe({ Echo, Pusher: function Pusher() {}, onLog: (entry) => logs.push(entry), onEvent: (entry) => events.push(entry), request: async (url, options) => { requests.push({ url, options }); if (request) return request(url, options); if (url.includes('/config')) return structuredClone(config); if (url.includes('/auth')) return { auth: 'fixture-signature' }; return { code: 0, data: { last_seq: 0, resync: false, events: [] } }; } }); await probe.configure('https://api.example.test/', 'Bearer fixture-token'); return { probe, logs, events, requests }; } test('configuration and historical replay never count as real-time delivery', async () => { const { probe } = await setup(); probe.connect([A]); probe.echo.state_change({ current: 'connected' }); probe.echo.private(A).ready(); await flush(); probe.receive(event(1), 'replay'); assert.equal(probe.state.liveCount, 0); assert.equal(probe.state.replayCount, 1); probe.echo.private(A).emit(event(2)); assert.equal(probe.state.liveCount, 1); }); test('manual reconnect retains each channel resume point and ignores a stale HTTP completion', async () => { const old = deferred(); let replayCalls = 0; const { probe, events, requests } = await setup((url) => { if (url.includes('/config')) return structuredClone(config); if (++replayCalls === 1) return old.promise; return { code: 0, data: { resync: false, last_seq: 2, events: [event(2)] } }; }); probe.connect([A]); const first = probe.echo; first.private(A).ready(); first.private(A).emit(event(1)); probe.disconnect(); old.resolve({ code: 0, data: { resync: false, last_seq: 10, events: [event(10)] } }); await flush(); probe.connect([A]); probe.echo.private(A).ready(); await flush(); assert.equal(new URL(requests.at(-1).url).searchParams.get('after_seq'), '0'); assert.deepEqual(events.map((e) => e.event.seq), [1, 2]); }); test('changing credentials invalidates in-flight responses and clears old session data', async () => { const response = deferred(); const { probe } = await setup(); probe.request = () => response.promise; const loading = probe.configure('https://api.example.test', 'new-token'); probe.invalidate(); response.resolve(structuredClone(config)); assert.equal(await loading, null); assert.equal(probe.state.config, null); assert.equal(probe.token, ''); assert.equal(probe.channels.size, 0); }); test('duplicate detection is scoped by seq and channel, not just seq', async () => { const { probe, events } = await setup(); probe.receive(event(1), 'live'); probe.receive(event(1), 'replay'); probe.receive(event(1, B), 'live'); assert.deepEqual(events.map((e) => e.duplicate), [false, true, false]); assert.equal(probe.state.duplicateCount, 1); }); test('rejected authorization stays visible and 401 stops the expired session', async () => { const { probe, logs } = await setup(); probe.connect([A]); probe.request = async () => { throw new ProbeError('expired', 401); }; let callbackError; await probe.authorize(`private-${A}`, '1.1', (error) => { callbackError = error; }, probe.epoch); assert.equal(callbackError.status, 401); assert.equal(probe.state.expired, true); assert.equal(probe.echo, null); assert.throws(() => probe.connect([A]), /登录已过期/); assert.ok(logs.some((entry) => entry.message.includes('登录已过期'))); }); test('applying subscriptions twice does not add duplicate handlers; leaving stops that channel', async () => { const { probe, events } = await setup(); probe.connect([A]); const sub = probe.echo.private(A); probe.subscribe([A]); sub.emit(event(1)); assert.equal(events.length, 1); probe.subscribe([]); sub.emit(event(2)); assert.equal(events.length, 1); }); test('redaction removes credentials from log messages and nested event payloads', () => { const token = 'fixture-token'; const safe = redact({ message: `Echo failed for Bearer ${token}`, data: { token, password: 'hidden', text: token } }, token); assert.equal(JSON.stringify(safe).includes(token), false); assert.equal(safe.data.password, '[已隐藏]'); }); test('base and auth URLs cannot carry credentials or change the request origin', () => { assert.throws(() => apiAddress('https://name:password@example.test'), /账号/); assert.throws(() => apiAddress('javascript:alert(1)'), /http/); assert.throws(() => apiEndpoint('https://api.example.test', '//other.example.test/auth'), /路径/); assert.equal(apiEndpoint('https://api.example.test/base', '/admin/broadcasting/auth'), 'https://api.example.test/base/admin/broadcasting/auth'); }); test('invalid reconfiguration does not leave a previous usable configuration', async () => { const { probe } = await setup(); await assert.rejects(probe.configure('not-an-address', 'new-token'), /完整 API 地址/); assert.equal(probe.state.config, null); assert.throws(() => probe.connect([A]), /先获取/); }); test('statistics channels receive stats.updated snapshots without treating them as list rows', async () => { const { probe, events } = await setup(); const name = 'admin.stats.platform_count'; probe.connect([name]); probe.echo.private(name).emit({ seq: 1, channel: name, event: 'stats.updated', action: 'updated', key: { scope: 'all' }, data: { total: 2, apple: 1, android: 0, pc: 1, mobile: 0, wechat: 0, unknown: 0 } }); assert.equal(probe.state.liveCount, 1); assert.equal(events[0].event.data.total, 2); assert.deepEqual(events[0].event.key, { scope: 'all' }); });