LocalNetMsg/scripts/selftest/sync-e2e.ts

// 同步盘端到端测试 (两个虚拟节点 A/B, 走真实的 WS + HTTP 通道)
//
// 覆盖的是"用户真正会遇到的路径":
//   对方弹窗同意 -> 建立关系 -> 双向同步 -> 改文件 -> 删文件 -> 删除传播 -> 墓碑防复活
// 和 scripts/selftest/main.ts 的定位不同: 那边是逐条断言, 这边是**完整链路**,
// 用来守住"删了能删掉、删了不会复活"这两条命门。
//
// 运行: npm run selftest:sync
import { mkdirSync, rmSync, writeFileSync, existsSync, readFileSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'

import { ChatServer } from '../../src/main/chat-server'
import { ChatClient } from '../../src/main/chat-client'
import { FileServer } from '../../src/main/file-server'
import { RelationStore } from '../../src/main/relations'
import { ManifestStore } from '../../src/main/sync-manifest'
import { RelationService } from '../../src/main/relation-service'
import { scanSync, applySyncItems } from '../../src/main/sync'
import { listPendingApprovals, replyApproval } from '../../src/main/remote/approval'
import type { DeviceInfo, WsFrame } from '../../src/main/protocol'

const sleep = (ms: number) => new Promise(r => setTimeout(r, ms))
let failures = 0
function check(name: string, cond: any, extra?: string) {
  if (cond) { console.log(`  PASS ${name}`) }
  else { failures++; console.log(`  FAIL ${name}${extra ? ' :: ' + extra : ''}`) }
}

class Node {
  info: DeviceInfo
  server: ChatServer
  client: ChatClient
  fileServer: FileServer
  relations: RelationStore
  manifestStore: ManifestStore
  relationService!: RelationService
  syncDir = ''
  base = ''

  constructor(id: string, name: string) {
    this.info = {
      deviceId: id, name, hostname: name, platform: 'linux', version: 1,
      appVersion: '0.1.0', address: '127.0.0.1', chatPort: 0, filePort: 0, ts: Date.now(),
    }
    this.base = join(tmpdir(), `lnm-smoke-${id.slice(-4)}-${Date.now()}`)
    rmSync(this.base, { recursive: true, force: true })
    mkdirSync(this.base, { recursive: true })
    this.relations = new RelationStore(join(this.base, 'relations.json'))
    this.manifestStore = new ManifestStore(join(this.base, 'manifest.json'))
    this.server = new ChatServer(0)
    this.client = new ChatClient(this.info)
    this.fileServer = new FileServer(0, join(this.base, 'dl'), {
      getDriveRoot: () => join(this.base, 'drive'),
      getDriveToken: () => `tok-${id}`,
      isDriveEnabled: () => true,
      authorize: (token, peerId) => this.relationService
        ? this.relationService.authorizeDrive(token, peerId)
        : { ok: false, reason: '未就绪' },
      getSyncScope: (relationId, peerId) => this.relationService?.getSyncScope(relationId, peerId) ?? null,
      getSyncManifest: ({ relationId }) => this.relationService?.getSyncManifest(relationId) ?? null,
      onSyncManifest: ({ relationId, wire }) => this.relationService
        ? this.relationService.applyPeerManifest(relationId, wire)
        : { applied: [], conflicts: [] },
    })
  }

  async start() {
    const cp = await this.server.start()
    const fp = await this.fileServer.start()
    this.info.chatPort = cp
    this.info.filePort = fp
    mkdirSync(join(this.base, 'drive'), { recursive: true })
    this.syncDir = join(this.base, 'sync')
    mkdirSync(this.syncDir, { recursive: true })
    this.relationService = new RelationService({
      store: this.relations,
      manifestStore: this.manifestStore,
      self: this.info,
      send: (peerId, frame) => this.client.send(frame, peerId),
      notify: () => {},
    })
    const route = (peer: DeviceInfo, frame: WsFrame) => {
      if (frame.type === 'relation:request') {
        this.relationService.handleRelationRequest(peer, frame.payload).catch(() => {})
      } else if (frame.type === 'relation:accept') {
        this.relationService.handleRelationAccept(peer, frame.payload)
      } else if (frame.type === 'relation:reject') {
        this.relationService.handleRelationReject(peer, frame.payload)
      }
    }
    this.server.on('frame', route)
    this.client.on('message', route)
    this.client.on('error', () => {})
  }

  connectTo(other: Node) { this.client.connectTo({ ...other.info }) }
  async stop() { try { await this.server.stop() } catch {} ; try { this.client.stop() } catch {} }
}

async function main() {
  const A = new Node('aaaaaaaa-1111-2222-3333-444444444444', 'A机')
  const B = new Node('bbbbbbbb-1111-2222-3333-444444444444', 'B机')
  await A.start()
  await B.start()
  A.connectTo(B)
  B.connectTo(A)
  await sleep(700)

  // ---- 1) 建立关系 (模拟对方在弹窗里点同意) ----
  const req = A.relationService.requestRelation({
    peerId: B.info.deviceId, peerName: B.info.name,
    peerAddress: '127.0.0.1', peerChatPort: B.info.chatPort, peerFilePort: B.info.filePort,
    syncLocalDir: A.syncDir, syncRemotePath: '', grants: ['drive', 'sync'],
  })
  check('发起关系请求成功', req.ok, JSON.stringify(req))
  await sleep(500)
  const ask = listPendingApprovals().find(x => x.kind === 'relation' && x.peerId === A.info.deviceId)
  check('对方弹出了关系授权请求', !!ask)
  if (ask) {
    replyApproval(ask.requestId, true, { grants: ['drive', 'sync'], syncLocalDir: B.syncDir, syncRemotePath: '' })
  }
  await sleep(700)
  const relA = A.relations.getRelationByPeer(B.info.deviceId)
  const relB = B.relations.getRelationByPeer(A.info.deviceId)
  // 实际同步根 = 顶层目录 + 按关系隔离的子目录 (一个顶层目录管多个聊天备份时各自隔开)
  const aRoot = join(A.syncDir, relA?.syncSubdir || '')
  const bRoot = join(B.syncDir, relB?.syncSubdir || '')
  mkdirSync(aRoot, { recursive: true })
  mkdirSync(bRoot, { recursive: true })
  check('A 侧关系 active', relA?.state === 'active', relA?.state)
  check('B 侧关系 active', relB?.state === 'active', relB?.state)
  check('双方令牌已交换', !!relA?.peerToken && !!relB?.peerToken)
  if (!relA || !relB) { await A.stop(); await B.stop(); process.exit(1) }

  // ---- 2) 同步: A 的文件传到 B ----
  writeFileSync(join(aRoot, 'hello.txt'), 'HELLO-FROM-A')
  writeFileSync(join(bRoot, 'only-b.txt'), 'ONLY-FROM-B')
  const ctxA = {
    peer: B.info, localDir: aRoot, token: relA.peerToken,
    relationId: relA.relationId, store: A.manifestStore,
    selfDeviceId: A.info.deviceId,
  }
  const scan1 = await scanSync(ctxA)
  check('扫描成功', scan1.ok, scan1.reason)
  check('hello.txt 判为待上传', scan1.items.find(i => i.relPath === 'hello.txt')?.action === 'upload',
    JSON.stringify(scan1.items.map(i => [i.relPath, i.action])))
  check('only-b.txt 判为待下载', scan1.items.find(i => i.relPath === 'only-b.txt')?.action === 'download',
    JSON.stringify(scan1.items.map(i => [i.relPath, i.action])))
  const items1 = scan1.items
    .filter(i => i.action === 'upload' || i.action === 'download' || i.action === 'delete-local' || i.action === 'delete-remote')
    .map(i => ({ relPath: i.relPath, action: i.action as any }))
  const ap1 = await applySyncItems({ ...ctxA, items: items1 })
  check('同步执行成功', ap1.ok, JSON.stringify(ap1.results))
  check('B 收到了 hello.txt', existsSync(join(bRoot, 'hello.txt')))
  check('A 收到了 only-b.txt', existsSync(join(aRoot, 'only-b.txt')))
  check('内容正确', readFileSync(join(bRoot, 'hello.txt'), 'utf8') === 'HELLO-FROM-A')

  // ---- 3) 本地改了 -> 应判 upload (有基线了) ----
  writeFileSync(join(aRoot, 'hello.txt'), 'HELLO-FROM-A-V2')
  const scan2 = await scanSync(ctxA)
  check('改过的文件判为待上传(不是冲突)',
    scan2.items.find(i => i.relPath === 'hello.txt')?.action === 'upload',
    JSON.stringify(scan2.items.map(i => [i.relPath, i.action])))

  // ---- 4) 删除传播: A 删了 -> B 那份也要没 ----
  rmSync(join(aRoot, 'hello.txt'))
  const scan3 = await scanSync(ctxA)
  check('删掉的文件判为"将删除对方"',
    scan3.items.find(i => i.relPath === 'hello.txt')?.action === 'delete-remote',
    JSON.stringify(scan3.items.map(i => [i.relPath, i.action])))
  const ap3 = await applySyncItems({
    ...ctxA,
    items: scan3.items.filter(i => i.action === 'delete-remote' || i.action === 'delete-local' || i.action === 'upload' || i.action === 'download')
      .map(i => ({ relPath: i.relPath, action: i.action as any })),
  })
  check('删除传播执行成功', ap3.ok, JSON.stringify(ap3.results))
  check('B 那份也被删了', !existsSync(join(bRoot, 'hello.txt')))

  // ---- 5) 墓碑: 再扫一次不该把它当成"对方新增"重新下载 ----
  const scan4 = await scanSync(ctxA)
  const again = scan4.items.find(i => i.relPath === 'hello.txt')
  check('墓碑生效: 不会被重新下载', !again || again.action === 'same',
    JSON.stringify(scan4.items.map(i => [i.relPath, i.action])))

  // ---- 6) 授权边界: 错令牌被拒 ----
  const bad = await scanSync({ ...ctxA, token: 'wrong-token' })
  check('错令牌被拒', !bad.ok, bad.reason)

  // ---- 7) 反向删除传播: B 删了 -> A 那份也要没 ----
  rmSync(join(bRoot, 'only-b.txt'))
  const ctxB = {
    peer: A.info, localDir: bRoot, token: relB.peerToken,
    relationId: relB.relationId, store: B.manifestStore,
    selfDeviceId: B.info.deviceId,
  }
  const scanB = await scanSync(ctxB)
  check('B 侧识别出"我删了 only-b.txt, 要传播给对方"',
    scanB.items.find(i => i.relPath === 'only-b.txt')?.action === 'delete-remote',
    JSON.stringify(scanB.items.map(i => [i.relPath, i.action])))
  await applySyncItems({
    ...ctxB,
    items: scanB.items.filter(i => i.action !== 'same').map(i => ({ relPath: i.relPath, action: i.action as any })),
  })
  check('A 那份 only-b.txt 也被删了', !existsSync(join(aRoot, 'only-b.txt')))

  // ---- 7b) 完整删除传播链路: A 删 -> A 推关系表 -> B 那边自动删掉本地那份 ----
  writeFileSync(join(aRoot, 'both.txt'), 'SHARED')
  await applySyncItems({ ...ctxA, items: [{ relPath: 'both.txt', action: 'upload' }] })
  const scanB2 = await scanSync(ctxB)                     // B 拉一次, 建立基线
  await applySyncItems({
    ...ctxB,
    items: scanB2.items.filter(i => i.action === 'download')
      .map(i => ({ relPath: i.relPath, action: i.action as any })),
  })
  check('both.txt 同步到了 B', existsSync(join(bRoot, 'both.txt')))
  rmSync(join(aRoot, 'both.txt'))
  const scanDel = await scanSync(ctxA)
  check('A 侧判为"将删除对方"',
    scanDel.items.find(i => i.relPath === 'both.txt')?.action === 'delete-remote',
    JSON.stringify(scanDel.items.map(i => [i.relPath, i.action])))
  await applySyncItems({ ...ctxA, items: [{ relPath: 'both.txt', action: 'delete-remote' }] })
  await sleep(400)                                       // 等对方收到并应用
  check('删除传播到 B: B 本地那份也没了', !existsSync(join(bRoot, 'both.txt')))

  // ---- 8) 冲突: 两边都改同一个文件 -> 判 conflict, 不静默覆盖 ----
  writeFileSync(join(aRoot, 'conf.txt'), 'A-VERSION')
  await applySyncItems({ ...ctxA, items: [{ relPath: 'conf.txt', action: 'upload' }] })
  writeFileSync(join(aRoot, 'conf.txt'), 'A-VERSION-2')
  writeFileSync(join(bRoot, 'conf.txt'), 'B-VERSION-2')
  const scanC = await scanSync(ctxA)
  check('两边都改过 -> 判冲突',
    scanC.items.find(i => i.relPath === 'conf.txt')?.action === 'conflict',
    JSON.stringify(scanC.items.map(i => [i.relPath, i.action])))

  // ---- 9) 复活防护: 对方拿更早的 seq 声称文件还在 -> 墓碑赢 ----
  const M = require('../../src/main/sync-manifest')
  const man = { relationId: 'r1', clock: 0, remoteSeq: 0, entries: {} as any }
  const snap = (relPath: string, hash: string) => ({ relPath, size: 10, mtime: Date.now(), hash })
  // 先建立基线: 两边一致
  M.reconcileLocal(man, [snap('z.txt', 'H1')])
  const peerMap = new Map([['z.txt', { fileId: 'f1', relPath: 'z.txt', hash: 'H1', size: 10, mtime: 1, state: 'present', seq: 1, deletedAt: 0 }]])
  M.mergePeer(man, peerMap)
  M.updateBaseline(man, 'z.txt', 'H1', 10, 1)
  // 本地删除 -> 墓碑
  M.reconcileLocal(man, [])
  const tombSeq = man.entries['z.txt'].mySeq
  check('本地删除记成了墓碑', man.entries['z.txt'].myState === 'deleted' && tombSeq > 0)
  // 对方用一个更早的 seq 声称文件还在 -> 墓碑应该赢
  const stalePeer = new Map([['z.txt', { fileId: 'f1', relPath: 'z.txt', hash: 'H1', size: 10, mtime: 1, state: 'present', seq: 1, deletedAt: 0 }]])
  M.mergePeer(man, stalePeer)
  const planStale = M.planSync(man, [], stalePeer)
  check('墓碑赢: 不会让已删文件复活',
    planStale.items.find(i => i.relPath === 'z.txt')?.action === 'delete-remote',
    JSON.stringify(planStale.items.map(i => [i.relPath, i.action])))
  // 对方用更新的 seq 声称文件还在 -> 接受复活
  const freshPeer = new Map([['z.txt', { fileId: 'f1', relPath: 'z.txt', hash: 'H2', size: 10, mtime: 2, state: 'present', seq: tombSeq + 5, deletedAt: 0 }]])
  M.mergePeer(man, freshPeer)
  const planFresh = M.planSync(man, [], freshPeer)
  check('对方 seq 更新 -> 接受复活(下载)',
    planFresh.items.find(i => i.relPath === 'z.txt')?.action === 'download',
    JSON.stringify(planFresh.items.map(i => [i.relPath, i.action])))

  await A.stop()
  await B.stop()
  console.log(failures === 0 ? '\n全部通过' : `\n${failures} 条失败`)
  process.exit(failures === 0 ? 0 : 1)
}

main().catch((e) => { console.error('崩溃:', e); process.exit(1) })