fix(0.2.1): upload 契约对齐客户端 —— 外键列即父行 uuid (projectId/columnId/parentId/taskId); 附 Node 闭环验证脚本

2026-10-26 01:14:34 +08:00
parent a714013e3e
commit 9caaa303e5
4 changed files with 292 additions and 15 deletions
@@ -0,0 +1,210 @@
+// 0.2.1 闭环验证脚本 (Node 客户端): 同步下行 -> 本地离线写入 -> 上行 -> 服务端编号 -> 同步回填
+// 用法: node scripts/verify-powersync-client.mjs [BASE_URL] [EMAIL] [PASSWORD]
+//   BASE_URL 默认 http://bh.vps.honor3.com:13001
+// 流程:
+//   1. REST 登录 -> /api/powersync/token
+//   2. Node SDK 连接 PowerSync, 等待首次同步, 校验 ETL 数据 (项目/任务/编号)
+//   3. 本地新建一条验证任务 (number 为空, 模拟离线写入)
+//   4. 等待 upload 完成并同步回填 -> 本地行应获得服务端分配的 number
+//   5. 清理: 本地删除该任务 (上行级联删除), 恢复数据原状
+import 'dotenv/config'
+import crypto from 'node:crypto'
+import os from 'node:os'
+import path from 'node:path'
+import fs from 'node:fs'
+import { column, Schema, Table, PowerSyncDatabase, createConsoleLogger, LogLevels } from '@powersync/node'
+
+const BASE = (process.argv[2] || 'http://bh.vps.honor3.com:13001').replace(/\/+$/, '')
+const EMAIL = process.argv[3] || 'admin@iboard.local'
+const PASSWORD = process.argv[4] || 'admin123'
+
+const uuid = () => crypto.randomUUID()
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
+
+// 与 src/lib/powersync/schema.js 同构的最小 schema (验证只用到 Project/Task/TaskTag)
+const AppSchema = new Schema({
+  Project: new Table({
+    uuid: column.text, code: column.text, name: column.text, ownerId: column.text,
+    isPersonal: column.integer, archived: column.integer, notes: column.text,
+    taskKinds: column.text, icon: column.text, createdAt: column.text, updatedAt: column.text,
+  }),
+  Task: new Table({
+    uuid: column.text, projectId: column.text, columnId: column.text, parentId: column.text,
+    title: column.text, description: column.text, kind: column.text, markDate: column.text,
+    status: column.text, doneAt: column.text, assigneeId: column.text, creatorId: column.text,
+    extras: column.text, sortOrder: column.integer, number: column.integer,
+    createdAt: column.text, updatedAt: column.text,
+  }),
+  TaskTag: new Table({ uuid: column.text, taskId: column.text, name: column.text }),
+  TaskMember: new Table({ uuid: column.text, taskId: column.text, userId: column.text }),
+  ProjectMember: new Table({ uuid: column.text, projectId: column.text, userId: column.text, permission: column.text, createdAt: column.text }),
+  BoardColumn: new Table({
+    uuid: column.text, projectId: column.text, name: column.text, sortOrder: column.integer,
+    boundStatus: column.text, retention: column.text, createdAt: column.text, updatedAt: column.text,
+  }),
+  User: new Table({ name: column.text, email: column.text }),
+})
+
+async function api(path, { method = 'GET', token, body } = {}) {
+  const res = await fetch(BASE + '/api' + path, {
+    method,
+    headers: { 'Content-Type': 'application/json', ...(token ? { Authorization: 'Bearer ' + token } : {}) },
+    body: body ? JSON.stringify(body) : undefined,
+  })
+  const data = await res.json().catch(() => ({}))
+  if (!res.ok) throw new Error(`${path} -> ${res.status}: ${data.error || '请求失败'}`)
+  return data
+}
+
+class Connector {
+  constructor(appToken, endpoint) {
+    // appToken: 应用 JWT (调业务 API 上行用); endpoint + sync JWT 由 fetchCredentials 提供
+    this.appToken = appToken
+    this.endpoint = endpoint
+  }
+  async fetchCredentials() {
+    const cred = await api('/powersync/token', { method: 'POST', token: this.appToken })
+    return { endpoint: cred.endpoint, token: cred.token }
+  }
+  async uploadData(database) {
+    const batch = await database.getCrudBatch()
+    if (!batch) return
+    const ops = []
+    for (let i = 0; i < batch.crud.length; i++) {
+      const op = batch.crud[i]
+      let data = op.opData || {}
+      if (op.op !== 'DELETE') {
+        const row = await database.getOptional(`SELECT * FROM "${op.table}" WHERE id = ?`, [op.id])
+        if (row) data = row
+      }
+      ops.push({ id: String(i), table: op.table, op: op.op === 'DELETE' ? 'DELETE' : 'PUT', uuid: op.id, data })
+    }
+    const res = await fetch(BASE + '/api/powersync/upload', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + this.appToken },
+      body: JSON.stringify({ ops }),
+    })
+    const out = await res.json().catch(() => ({}))
+    if (!res.ok) throw new Error('upload failed: ' + JSON.stringify(out))
+    const failed = (out.results || []).filter((r) => !r.ok)
+    if (failed.length) console.log('[verify] upload 部分失败:', JSON.stringify(failed))
+    await batch.complete()
+  }
+}
+
+async function main() {
+  console.log(`[verify] BASE = ${BASE}`)
+  const login = await api('/auth/login', { method: 'POST', body: { email: EMAIL, password: PASSWORD } })
+  console.log('[verify] 1. 登录 OK, user:', login.user.email)
+  const cred = await api('/powersync/token', { method: 'POST', token: login.token })
+  if (!cred.endpoint) throw new Error('POWERSYNC_URL 未配置 (服务端 /powersync/token 返回空 endpoint)')
+  console.log('[verify] 2. 同步凭证 OK, endpoint:', cred.endpoint)
+
+  const dbFile = path.join(os.tmpdir(), `iboard-verify-${Date.now()}.db`)
+  const logger = createConsoleLogger({ minLevel: LogLevels.debug })
+  const db = new PowerSyncDatabase({
+    schema: AppSchema,
+    database: { dbFilename: dbFile, databaseWorkerLogLevel: LogLevels.debug },
+    sync: { logLevel: LogLevels.debug },
+    logger,
+  })
+  db.registerListener({
+    statusChanged: (s) => {
+      const c = s?.dataFlowStatus || {}
+      console.log(`[verify]    sync status: connected=${s?.connected} connecting=${s?.connecting} err=${c.downloadError?.message || c.uploadError?.message || '-'}`)
+    },
+  })
+  try {
+    await db.connect(new Connector(login.token, cred.endpoint))
+    console.log('[verify]    connect() 已建立')
+  } catch (e) {
+    console.error('[verify] connect 失败:', e?.message)
+    throw e
+  }
+  await Promise.race([
+    db.waitForFirstSync(),
+    sleep(45000).then(() => { throw new Error('waitForFirstSync 超时 (45s)') }),
+  ]).catch(async (e) => {
+    console.log('[verify] waitForFirstSync 未完成, 轮询内部状态...')
+    for (let i = 0; i < 10; i++) {
+      await sleep(3000)
+      const st = db.currentStatus
+      const counts = {
+        projects: (await db.getOptional('SELECT COUNT(*) AS n FROM Project'))?.n,
+        tasks: (await db.getOptional('SELECT COUNT(*) AS n FROM Task'))?.n,
+      }
+      console.log(`[verify]    poll#${i}:`, JSON.stringify({
+        connected: st?.connected, connecting: st?.connecting, hasSynced: st?.hasSynced,
+        lastSyncedAt: st?.lastSyncedAt, dl: st?.dataFlowStatus?.downloadError?.message,
+        ul: st?.dataFlowStatus?.uploadError?.message, ...counts,
+      }))
+      if (st?.hasSynced && counts.projects > 0) {
+        console.log('[verify]    轮询确认同步已完成')
+        return
+      }
+    }
+    throw e
+  })
+  console.log('[verify] 3. 首次同步完成')
+
+  const projects = await db.getAll('SELECT id, code, name FROM Project ORDER BY id')
+  console.log(`[verify]    项目 ${projects.length} 个:`, projects.map((p) => `${p.code || '?'}(${p.name})`).join(', '))
+  const tasks = await db.getAll('SELECT COUNT(*) AS n FROM Task')
+  const numbered = await db.getAll('SELECT COUNT(*) AS n FROM Task WHERE number IS NOT NULL')
+  console.log(`[verify]    任务 ${tasks[0].n} 条 (已有编号 ${numbered[0].n} 条)`)
+  if (!projects.length || tasks[0].n === 0) throw new Error('同步数据为空, ETL 未生效?')
+
+  // 本地新建验证任务 (模拟离线写入: number 为空, 等服务端分配)
+  const target = projects.find((p) => p.code) || projects[0]
+  const firstCol = await db.get(`SELECT id, boundStatus FROM BoardColumn WHERE projectId = ? ORDER BY sortOrder, id LIMIT 1`, [target.id])
+  const marker = `离线闭环验证-${Date.now()}`
+  const newId = uuid()
+  const ts = new Date().toISOString()
+  await db.writeTransaction(async (tx) => {
+    await tx.execute(
+      `INSERT INTO Task (id, uuid, projectId, columnId, parentId, title, description, kind, markDate, status, doneAt, assigneeId, creatorId, extras, sortOrder, number, createdAt, updatedAt)
+       VALUES (?, ?, ?, ?, NULL, ?, '', NULL, NULL, ?, NULL, NULL, ?, '[]', 0, NULL, ?, ?)`,
+      [newId, newId, target.id, firstCol.id, marker, firstCol.boundStatus || 'todo', String(login.user.id), ts, ts],
+    )
+  })
+  console.log(`[verify] 4. 本地写入任务 "${marker}" (number 为空), 等待上行 + 服务端定号 + 同步回填...`)
+
+  let syncedBack = null
+  for (let i = 0; i < 30; i++) {
+    await sleep(2000)
+    const row = await db.getOptional('SELECT number, updatedAt FROM Task WHERE id = ?', [newId])
+    const localCheck = await db.getOptional(`SELECT 1 AS x FROM ps_crud LIMIT 1`)
+    if (!localCheck) {
+      // 队列已清空 (上行完成); 等待下行回填 number
+      if (row && row.number != null) { syncedBack = row; break }
+      if (i > 10 && row && row.number == null) continue
+    }
+  }
+  if (!syncedBack) throw new Error('超时: 未观测到服务端编号回填')
+  console.log(`[verify] 5. 闭环 OK! 服务端分配编号: ${target.code}-${syncedBack.number} (项目 ${target.code})`)
+
+  // 直接从服务端核对 (REST 视角)
+  const serverCheck = await api(`/tasks/${newId}`, { token: login.token })
+  console.log(`[verify] 6. 服务端确认: 任务存在, number=${serverCheck.item.number}, uuid=${serverCheck.item.uuid.slice(0, 13)}...`)
+
+  // 清理验证数据 (本地删除 -> 上行删除)
+  await db.execute('DELETE FROM Task WHERE id = ?', [newId])
+  for (let i = 0; i < 15; i++) {
+    await sleep(2000)
+    const left = await db.getOptional('SELECT 1 AS x FROM ps_crud LIMIT 1')
+    if (!left) break
+  }
+  const gone = await api(`/tasks/${newId}`, { token: login.token }).catch(() => null)
+  if (!gone) console.log('[verify] 7. 验证任务已从服务端清除')
+  else throw new Error('验证任务删除未同步到服务端')
+
+  db.close()
+  fs.rmSync(dbFile, { force: true })
+  fs.rmSync(dbFile + '-journal', { force: true })
+  console.log('[verify] 全部通过: 下行同步 / 离线写入 / 上行幂等 / 服务端编号 / 回填 / 删除同步')
+}
+
+main().catch((e) => {
+  console.error('[verify] 失败:', e.message)
+  process.exit(1)
+})