diff --git a/intern-log/repro/engine-realtime.ts b/intern-log/repro/engine-realtime.ts new file mode 100644 index 0000000..35b9ff7 --- /dev/null +++ b/intern-log/repro/engine-realtime.ts @@ -0,0 +1,112 @@ +// 实时链路引擎级验证:本地写入 -> 3s 防抖自动推送;外部端写入 -> SSE -> 自动拉取 +import 'fake-indexeddb/auto' + +const BASE = 'http://127.0.0.1:3780' +let pass = 0, fail = 0 +function check(name: string, ok: boolean, detail?: unknown): void { + if (ok) { pass++; console.log(' PASS ' + name) } + else { fail++; console.log(' FAIL ' + name + (detail !== undefined ? ' -> ' + String(JSON.stringify(detail)).slice(0, 200) : '')) } +} + +// Node 没有 localStorage:先垫片再导入 auth/engine +const store = new Map() +;(globalThis as any).localStorage = { + getItem: (k: string) => (store.has(k) ? (store.get(k) as string) : null), + setItem: (k: string, v: string) => { store.set(k, String(v)) }, + removeItem: (k: string) => { store.delete(k) } +} +store.set('dsh_cloud_creds', JSON.stringify({ username: 'audit_a', password: 'aapass123' })) +store.set('dsh_session_user', JSON.stringify({ userId: 'local_rt', username: 'audit_a' })) + +// Node 24 自带 EventSource;老环境则垫一个基于 fetch 流的最小实现 +if (typeof (globalThis as any).EventSource === 'undefined') { + class MiniES { + onmessage: ((e: { data: string }) => void) | null = null + onerror: (() => void) | null = null + private ab = new AbortController() + constructor(url: string) { + fetch(url, { signal: this.ab.signal }).then(async r => { + const dec = new TextDecoder() + let buf = '' + const reader = (r.body as any).getReader() + while (true) { + const { done, value } = await reader.read() + if (done) break + buf += dec.decode(value) + while (buf.includes('event: change')) { + buf = buf.slice(buf.indexOf('event: change') + 'event: change'.length) + const cb = this.onmessage + if (cb) cb({ data: '{}' }) + } + } + }).catch(() => { const cb = this.onerror; if (cb) cb() }) + } + close(): void { this.ab.abort() } + } + ;(globalThis as any).EventSource = MiniES +} + +async function main() { + const schema = await import('../src/db/schema') + const bindDatabase = schema.bindDatabase, WorklogDB = schema.WorklogDB + const engine = await import('../src/sync/engine') + const repo = await import('../src/db/repo') + check('Node 有 EventSource(原生或垫片)', typeof (globalThis as any).EventSource !== 'undefined') + + bindDatabase(new WorklogDB('rt_a')) + await repo.setSetting('sync_base_url', BASE) // 测试环境指向本机服务 + engine.enableRealtimeSync() + + // ---- 场景1:本地写入 -> 防抖 3s -> 自动推送 ---- + await schema.db.project.add({ name: '实时推送项目', archived: 0, created_at: '2026-01-01T00:00:00' }) + const t0 = performance.now() + + const login = await fetch(BASE + '/api/auth/login', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ username: 'audit_a', password: 'aapass123' }) + }) + const tok = (await login.json()).token as string + async function serverSnapshot(): Promise> { + const r = await fetch(BASE + '/api/sync', { + method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + tok }, + body: JSON.stringify({ tables: {} }) + }) + return (await r.json()).tables + } + let pushedAt = -1 + for (let i = 0; i < 30; i++) { + await new Promise(r => globalThis.setTimeout(r, 300)) + const snap = await serverSnapshot() + if ((snap.project || []).some((p: any) => p.name === '实时推送项目')) { pushedAt = performance.now() - t0; break } + } + check('本地写入 4.5 秒内自动推送上云', pushedAt > 0, 'syncLastError=' + engine.syncLastError.value) + if (pushedAt > 0) console.log(' INFO 到云端可见耗时: ' + Math.round(pushedAt) + ' ms') + check('自动推送未留下错误', engine.syncLastError.value === '', engine.syncLastError.value) + + // ---- 场景2:外部端写入 -> SSE 通知 -> 本端自动拉取 ---- + await fetch(BASE + '/api/sync', { + method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + tok }, + body: JSON.stringify({ tables: { memo: [{ uuid: 'ext-memo-1', title: '来自另一端', content: 'sse', updated_at: '2026-02-03T00:00:00' }] } }) + }) + const t1 = performance.now() + let pulledAt = -1 + for (let i = 0; i < 25; i++) { + await new Promise(r => globalThis.setTimeout(r, 250)) + const got = await schema.db.memo.toArray() + if (got.some(m => (m as any).uuid === 'ext-memo-1')) { pulledAt = performance.now() - t1; break } + } + check('外部端写入后 6 秒内自动拉取到本地', pulledAt > 0, 'syncLastError=' + engine.syncLastError.value) + if (pulledAt > 0) console.log(' INFO SSE 通知到本地落库耗时: ' + Math.round(pulledAt) + ' ms') + check('自动拉取未留下错误', engine.syncLastError.value === '', engine.syncLastError.value) + + // ---- 场景3:autoSyncNow 手动失败可见性(断開服务)---- + engine.stopRealtime() + await repo.setSetting('sync_base_url', 'http://127.0.0.1:9') // 不可达端口 + await engine.autoSyncNow() + check('服务不可达时 syncLastError 被记录', engine.syncLastError.value !== '', engine.syncLastError.value) + await repo.setSetting('sync_base_url', BASE) + + console.log('==== 实时链路审计: PASS ' + pass + ' / FAIL ' + fail + ' ====') + process.exit(fail > 0 ? 1 : 0) +} +main().catch(e => { console.error('FAILED:', (e && (e.stack || e.message)) || e); process.exit(1) })