sync: 各端互通引擎就位 —— wire 应用/拉取（LWW+FK uuid 映射）接入 /sync/push|pull，桌面与网页同库

2026-10-16 15:46:51 +08:00
parent ae278379af
commit 9d97384aaf
4 changed files with 189 additions and 4 deletions
@@ -80,6 +80,35 @@ export function buildSearchText(title, description, result, tags, extras) {
  return [title, description || '', result || '', (tags || []).join(' '), ...extraFlat]
    .filter(Boolean).join('\n');
 }
+
+// ---------- Wire 协议（各端互通：uuid 引用 FK，LWW 仲裁） ----------
+
+function parseHlc(h) {
+  const m = /^([0-9]+):([0-9]+):(.+)$/.exec(String(h || ''));
+  return m ? { ms: Number(m[1]), cnt: Number(m[2]), dev: m[3] } : null;
+}
+
+function hlcNewer(a, b) {
+  if (a.ms !== b.ms) return a.ms > b.ms;
+  if (a.cnt !== b.cnt) return a.cnt > b.cnt;
+  return a.dev > b.dev;
+}
+
+// 表 → [列, FK引用目标表]
+const WIRE_TABLES = {
+  footprint: { fks: { type_id: 'footprint_type', parent_id: 'footprint' } },
+  footprint_type: { fks: {} },
+  taxonomy: { fks: {} },
+  dimension: { fks: {} },
+  grade_band: { fks: {} },
+  metric_def: { fks: {} },
+  semester: { fks: {} },
+  course_grade: { fks: { semester_id: 'semester' } },
+  saved_view: { fks: {} },
+  attachment: { fks: { owner_id: 'footprint' } },
+  footprint_dimension: { fks: { footprint_id: 'footprint', dimension_id: 'dimension' } }
+};
+
 // 老库迁移：补列 + 回填 uuid/hlc/origin（幂等）
 const SYNC_TABLES = ['footprint', 'footprint_type', 'taxonomy', 'dimension', 'attachment',
  'semester', 'course_grade', 'metric_def', 'saved_view', 'grade_band'];
@@ -977,13 +1006,168 @@ function listFootprints(d, query) {
      return { ok: true };
    }

+    case 'sync_apply_wire':
+      return applyWireBatch(payload.changes || []);
+
+    case 'sync_pull_wire':
+      return pullWire(payload.cursor, payload.limit);
+
    default:
      throw new Error('未知命令：' + cmd);
  }
 }

 // 工厂闭合
-  return { handleCommand, getDb };
+    // ---------- Wire 协议：出站序列化 / 入站应用 ----------
+
+  const SKIP_COLS = ['id', 'uuid', 'hlc', 'origin', 'created_at', 'updated_at', 'search_text'];
+
+  function uuidOf(table, id) {
+    if (id == null) return null;
+    const r = db.prepare('SELECT uuid FROM ' + table + ' WHERE id=?').get(id);
+    return r ? r.uuid : null;
+  }
+
+  function idOfUuid(table, uuid, mayCreate) {
+    if (uuid == null) return null;
+    const r = db.prepare('SELECT id FROM ' + table + ' WHERE uuid=?').get(uuid);
+    if (r) return r.id;
+    if (!mayCreate) return null;
+    const h2 = hlcNext(db);
+    if (table === 'footprint_type') {
+      const rr = db.prepare("INSERT INTO footprint_type (name, icon, color, sort, form_template, is_builtin, uuid, hlc, origin) VALUES (?,?,?,?,?,0,?,?,?)")
+        .run('同步类型 ' + uuid.slice(0, 8), 'tag', '#64748b', 500, '[]', uuid, h2, 'sync');
+      return Number(rr.lastInsertRowid);
+    }
+    if (table === 'dimension') {
+      const rr = db.prepare("INSERT INTO dimension (name, color, sort, uuid, hlc, origin) VALUES (?,?,?,?,?,?)")
+        .run('同步维度 ' + uuid.slice(0, 8), '#64748b', 500, uuid, h2, 'sync');
+      return Number(rr.lastInsertRowid);
+    }
+    if (table === 'grade_band') {
+      const rr = db.prepare("INSERT INTO grade_band (label, color, min, max, sort, uuid, hlc, origin) VALUES (?,?,?,?,?,?,?,?)")
+        .run('同步梯度 ' + uuid.slice(0, 8), '#64748b', 60, 70, 500, uuid, h2, 'sync');
+      return Number(rr.lastInsertRowid);
+    }
+    if (table === 'metric_def') {
+      const rr = db.prepare("INSERT INTO metric_def (key, label, unit, agg, section, uuid, hlc, origin) VALUES (?,?,?,?,?,?,?,?)")
+        .run('sync_' + uuid.slice(0, 8), '同步指标 ' + uuid.slice(0, 8), null, 'sum', null, uuid, h2, 'sync');
+      return Number(rr.lastInsertRowid);
+    }
+    if (table === 'taxonomy') {
+      const rr = db.prepare("INSERT INTO taxonomy (kind, value, color, sort, uuid, hlc, origin) VALUES (?,?,?,0,?,?,?,?)")
+        .run('level', '同步条目 ' + uuid.slice(0, 8), '#64748b', uuid, h2, 'sync');
+      return Number(rr.lastInsertRowid);
+    }
+    return null;
+  }
+
+  function serializeRowOf(table) {
+    const spec = WIRE_TABLES[table];
+    return (row) => {
+      const out = { uuid: row.uuid, hlc: row.hlc, deleted_at: row.deleted_at || null };
+      for (const k of Object.keys(row)) {
+        if (SKIP_COLS.includes(k)) continue;
+        if (spec.fks[k] !== undefined) { out[k.replace('_id', '_uuid')] = uuidOf(spec.fks[k], row[k]); continue; }
+        out[k] = row[k];
+      }
+      return out;
+    };
+  }
+
+  function softDeleteById(table, id) {
+    db.prepare('UPDATE ' + table + " SET deleted_at=datetime('now','localtime') WHERE id=?").run(id);
+  }
+
+  function applyWire(change) {
+    const spec = WIRE_TABLES[change.table];
+    if (!spec) return { accepted: 0, stale: 0, invalid: 1 };
+    const rowUuid = String(change.row_uuid || '');
+    const incoming = parseHlc(change.hlc);
+    if (!incoming || (change.op !== 'upsert' && change.op !== 'delete')) return { accepted: 0, stale: 0, invalid: 1 };
+    const existing = db.prepare('SELECT id, hlc FROM ' + change.table + ' WHERE uuid=?').get(rowUuid);
+    if (existing) {
+      const e = parseHlc(existing.hlc);
+      if (e && !hlcNewer(incoming, e)) return { accepted: 0, stale: 1, invalid: 0 };
+    }
+    let res;
+    try {
+      res = withTx(db, () => {
+      if (change.op === 'delete') {
+        if (existing) softDeleteById(change.table, existing.id);
+        touchOplog(db, change.table, rowUuid, 'delete', change.hlc);
+        return 1;
+      }
+      const data = change.data || {};
+      const fv = {};
+      for (const fk of Object.keys(spec.fks)) {
+        const uuidVal = data[fk.replace('_id', '_uuid')];
+        fv[fk] = idOfUuid(spec.fks[fk], uuidVal, true);
+      }
+      const cols = [];
+      const vals = [];
+      for (const k of Object.keys(data)) {
+        if (['uuid', 'hlc', 'deleted_at'].includes(k)) continue;
+        if (k.endsWith('_uuid') && spec.fks[k.replace('_uuid', '_id')] !== undefined) continue;
+        cols.push(k);
+        vals.push(data[k]);
+      }
+      for (const fk of Object.keys(fv)) { cols.push(fk); vals.push(fv[fk]); }
+      cols.push('uuid', 'hlc', 'origin');
+      vals.push(rowUuid, change.hlc, 'sync');
+      if (change.table === 'footprint') {
+        cols.push('search_text');
+        vals.push(buildSearchText(String(data.title || ''), String(data.description || ''), String(data.result || ''), data.tags || [], data.extras || {}));
+      }
+      if (existing) {
+        const setSql = cols.map(c => c + '=?').join(', ');
+        db.prepare('UPDATE ' + change.table + ' SET ' + setSql + ' WHERE id=?').run(...vals, existing.id);
+      } else {
+        db.prepare('INSERT INTO ' + change.table + ' (' + cols.join(',') + ') VALUES (' + cols.map(() => '?').join(',') + ')').run(...vals);
+      }
+      touchOplog(db, change.table, rowUuid, 'upsert', change.hlc);
+      return 1;
+    });
+    } catch (ex) {
+      throw ex;
+    }
+    return res === 1 ? { accepted: 1, stale: 0, invalid: 0 } : { accepted: 0, stale: 0, invalid: 0 };
+  }
+
+  function applyWireBatch(changes) {
+    let accepted = 0; let stale = 0; let invalid = 0;
+    for (const ch of (changes || [])) {
+      const r = applyWire(ch);
+      accepted += r.accepted; stale += r.stale; invalid += r.invalid;
+    }
+    return { accepted, stale, invalid };
+  }
+
+  function pullWire(cursorRaw, limitRaw) {
+    const cursor = Math.max(0, Number(cursorRaw) || 0);
+    const limit = Math.min(500, Math.max(1, Number(limitRaw) || 200));
+    const opRows = db.prepare('SELECT id, table_name, row_uuid, op, hlc FROM sync_oplog WHERE id>? ORDER BY id LIMIT ?').all(cursor, limit);
+    const changes = [];
+    for (const op of opRows) {
+      if (!WIRE_TABLES[op.table_name]) continue;
+      if (op.op === 'delete') {
+        changes.push({ seq: op.id, table: op.table_name, row_uuid: op.row_uuid, op: 'delete', hlc: op.hlc, data: null });
+        continue;
+      }
+      const row = db.prepare('SELECT * FROM ' + op.table_name + ' WHERE uuid=?').get(op.row_uuid);
+      if (!row) continue;
+      changes.push({ seq: op.id, table: op.table_name, row_uuid: op.row_uuid, op: 'upsert', hlc: op.hlc, data: serializeRowOf(op.table_name)(row) });
+    }
+    const maxSeq = db.prepare('SELECT COALESCE(MAX(id),0) AS m FROM sync_oplog').get().m;
+    const hasMore = opRows.length === limit;
+    return {
+      changes,
+      cursor: hasMore && opRows.length ? opRows[opRows.length - 1].id : Math.max(cursor, maxSeq),
+      has_more: hasMore
+    };
+  }
+
+  return { handleCommand, getDb, pullWire, applyWireBatch };
 }

 // ---------- 默认单例（开发桥接 / seed 兼容） ----------
@@ -999,4 +1183,4 @@ export function getDb() {

 export function handleCommand(cmd, payload) {
  return getDefault().handleCommand(cmd, payload);
-}
+}