test: 新增实时链路回归（本地写入自动推送 + SSE 自动拉取）

2026-10-16 15:28:45 +08:00
parent 9bd04441e1
commit 8fad42edc7
1 changed files with 112 additions and 0 deletions
@@ -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<string, string>()
+;(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<Record<string, any>> {
+    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) })