iboard/scripts/verify-powersync-client.mjs

// 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, uuid, 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 视角: REST 仍按 int id 寻址, 通过列表接口按 uuid 匹配)
  const plist = await api('/projects', { token: login.token })
  const pj = plist.items.find((p) => p.uuid === target.uuid)
  if (!pj) throw new Error('服务端确认失败: 项目列表中找不到目标项目')
  const tlist = await api(`/projects/${pj.id}/tasks`, { token: login.token })
  const hit = tlist.items.find((t) => t.uuid === newId)
  if (!hit || hit.number == null) throw new Error('服务端确认失败: 任务不存在或未定号')
  console.log(`[verify] 6. 服务端确认: number=${hit.number}, uuid=${hit.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(`/projects/${pj.id}/tasks`, { token: login.token })
  if (!gone.items.some((t) => t.uuid === newId)) console.log('[verify] 7. 验证任务已从服务端清除')
  else throw new Error('验证任务删除未同步到服务端')

  db.close()
  try {
    fs.rmSync(dbFile, { force: true })
    fs.rmSync(dbFile + '-journal', { force: true })
  } catch { /* Windows 下可能被占用, 留在临时目录无害 */ }
  console.log('[verify] 全部通过: 下行同步 / 离线写入 / 上行幂等 / 服务端编号 / 回填 / 删除同步')
}

main().catch((e) => {
  console.error('[verify] 失败:', e.message)
  process.exit(1)
})