// 多端同步引擎:全量差异对比 + 记录级新者胜 + 软删除墓碑 // 实时性:本地变更 3 秒内推送;服务端变更通过 SSE 秒级通知其他端;60 秒轮询兜底 import Dexie, { liveQuery } from 'dexie' import { db } from '../db/schema' import { getSetting, setSetting } from '../db/repo' import { getCloudCreds } from '../db/auth' import { nowLocalISO } from '../lib/date' export const DEFAULT_SYNC_BASE = 'https://daywk.linyux.cn' const TABLES = ['project', 'work_log', 'log_item', 'attendance', 'weekly_note', 'setting', 'memo', 'credential', 'transaction'] // 本地敏感或临时性的设置不参与同步 const EXCLUDE_SETTING_KEYS = new Set(['sync_cursor', 'sync_last_time', 'sync_token', 'last_project', 'last_category']) export const SYNC_TABLES = TABLES interface AnyRow { [key: string]: any } function newUuid(): string { if (typeof crypto !== 'undefined' && crypto.randomUUID) return crypto.randomUUID() return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, c => { const r = (Math.random() * 16) | 0 return (c === 'x' ? r : (r & 0x3) | 0x8).toString(16) }) } function stripId(row: AnyRow): AnyRow { const copy: AnyRow = { ...row } delete copy.id return copy } function rowHash(row: AnyRow): string { return JSON.stringify(row) } async function getTable(name: string): Promise { return db.table(name).toArray() } // 主键访问:setting 表主键为字符串 key,其余表为自增 id function pkOf(t: string, row: AnyRow): any { return t === 'setting' ? row.key : row.id } // 一次性补齐 uuid 与 updated_at,缺失时回填 async function ensureSyncFields(): Promise { for (const t of TABLES) { const table = db.table(t) const rows: AnyRow[] = await table.toArray() for (const row of rows) { if (t === 'setting' && EXCLUDE_SETTING_KEYS.has(row.key)) continue const patch: AnyRow = {} if (!row.uuid) patch.uuid = newUuid() if (!row.updated_at) patch.updated_at = row.created_at || nowLocalISO() if (Object.keys(patch).length) await table.update(pkOf(t, row), patch) } } } export interface SyncResult { pushed: number pulled: number removed: number at: string } export interface SyncConfig { baseUrl: string username: string password: string } // 同步基址:设置页可覆盖,默认使用独立子域名 export async function getSyncBase(): Promise { return (await getSetting('sync_base_url', '')) || DEFAULT_SYNC_BASE } async function loadSyncConfig(): Promise { const creds = getCloudCreds() if (!creds) throw new Error('未登录,无法同步') return { baseUrl: await getSyncBase(), username: creds.username, password: creds.password } } // 使用当前登录凭据执行同步(供设置页与自动同步调用) export async function syncWithStoredConfig(): Promise { return runSync(await loadSyncConfig()) } // 登录后自动绑定云端账号:云端不存在则注册,同名则校验密码一致 export async function ensureCloudBinding(): Promise { const creds = getCloudCreds() if (!creds) throw new Error('未登录') const base = await getSyncBase() const headers = { 'Content-Type': 'application/json' } const body = JSON.stringify({ username: creds.username, password: creds.password }) let resp = await fetch(base + '/api/auth/register', { method: 'POST', headers, body }) if (resp.status === 409) { resp = await fetch(base + '/api/auth/login', { method: 'POST', headers, body }) if (!resp.ok) throw new Error('云端已存在同名账号且密码不一致,请检查') } else if (!resp.ok) { throw new Error('云端绑定失败:' + resp.status) } } export function getCloudUser(): string | null { const c = getCloudCreds() return c ? c.username : null } export async function isSyncConfigured(): Promise { return getCloudCreds() != null } export async function getSyncLastTime(): Promise { return getSetting('sync_last_time', '') } // 云端账号注册(服务端存 PBKDF2 哈希) export async function registerCloudAccount(baseUrl: string, username: string, password: string): Promise { const base = baseUrl.trim().replace(/\/+$/, '') if (!base || !username.trim() || password.length < 6) { throw new Error('请完整填写同步地址、用户名与至少 6 位的密码') } let resp: Response try { resp = await fetch(base + '/api/auth/register', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ username: username.trim(), password }) }) } catch { throw new Error('无法访问同步服务:请检查地址、端口与网络') } if (resp.status === 409) throw new Error('云端用户名已存在,可直接同步') if (!resp.ok) throw new Error('云端注册失败:' + resp.status) } // ---------- 云端令牌(登录换取,本地缓存,过期自动重登) ---------- async function loginCloud(base: string, config: SyncConfig): Promise { const resp = await fetch(base + '/api/auth/login', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ username: config.username.trim(), password: config.password }) }) if (!resp.ok) throw new Error('云端账号或密码不正确') const j = await resp.json() if (!j || !j.token) throw new Error('云端登录异常') return String(j.token) } async function ensureToken(base: string, config: SyncConfig): Promise { let token = await getSetting('sync_token', '') if (token) return token token = await loginCloud(base, config) await setSetting('sync_token', token) return token } // ---------- 并发保护与实时触发 ---------- let syncRunning = false let pendingSync = false let changesSuppressed = false let changeDebounce: number | null = null let sse: EventSource | null = null let sseKey = '' let realtimeAttached = false let realtimeSub: { unsubscribe: () => void } | null = null export async function autoSyncNow(): Promise { if (syncRunning) { pendingSync = true return } try { if (!(await isSyncConfigured())) return await runSync(await loadSyncConfig()) } catch { /* 自动同步静默失败,等待下次触发 */ } } // 实时触发:本地任何数据变更 3 秒后自动推送(Dexie 4 核心 liveQuery,跨全部业务表) export function enableRealtimeSync(): void { if (realtimeAttached) return realtimeAttached = true const sub = liveQuery(async () => { const counts: Record = {} for (const t of TABLES) counts[t] = await db.table(t).count() return counts }) realtimeSub = sub.subscribe({ next: () => { if (changesSuppressed || syncRunning) return if (changeDebounce) globalThis.clearTimeout(changeDebounce) changeDebounce = globalThis.setTimeout(() => { void autoSyncNow() }, 3000) }, error: () => { /* 忽略监听错误,轮询兜底 */ } }) } export function stopRealtime(): void { if (realtimeSub) { realtimeSub.unsubscribe(); realtimeSub = null } if (sse) { sse.close(); sse = null; sseKey = '' } if (changeDebounce) { globalThis.clearTimeout(changeDebounce); changeDebounce = null } } // 服务端 SSE 通知:其他端有变更时立即拉取 function ensureSSE(base: string, token: string): void { const key = base + '|' + token if (sse && sseKey === key) return if (sse) { sse.close(); sse = null } try { sse = new EventSource(base + '/api/events?token=' + encodeURIComponent(token)) sseKey = key sse.onmessage = () => { void autoSyncNow() } } catch { sse = null } } // ---------- 执行同步 ---------- export async function runSync(config: SyncConfig): Promise { if (syncRunning) { pendingSync = true return { pushed: 0, pulled: 0, removed: 0, at: '' } } syncRunning = true changesSuppressed = true try { const result = await doRunSync(config) if (pendingSync) { pendingSync = false globalThis.setTimeout(() => { void autoSyncNow() }, 800) } return result } finally { syncRunning = false changesSuppressed = false } } async function doRunSync(config: SyncConfig): Promise { const base = config.baseUrl.trim().replace(/\/+$/, '') if (!base) throw new Error('请先配置同步服务地址') await ensureSyncFields() // 1. 载入镜像,计算本地差异 const mirrorRows = await db.sync_mirror.toArray() const mirror = new Map() for (const m of mirrorRows) mirror.set(m.table + '|' + m.uuid, m) const pushTables: Record = {} const tombstones: { table: string; uuid: string; deleted_at: string }[] = [] let pushed = 0 for (const t of TABLES) { let rows: AnyRow[] = await getTable(t) if (t === 'setting') rows = rows.filter(r => !EXCLUDE_SETTING_KEYS.has(r.key)) const current = new Map() for (const row of rows) current.set(row.uuid, row) for (const [u, row] of current) { const clean = stripId(row) const h = rowHash(clean) const m = mirror.get(t + '|' + u) if (!m || m.hash !== h) { pushTables[t] = pushTables[t] || [] pushTables[t].push(clean) pushed++ } } for (const [key, m] of mirror) { if (key.startsWith(t + '|') && !current.has(m.uuid)) { tombstones.push({ table: t, uuid: m.uuid, deleted_at: nowLocalISO() }) } } pushTables[t] = pushTables[t] || [] } // 2. 认证并与服务端交换(token 失效自动重登一次) const token = await ensureToken(base, config) const doPush = (tk: string) => fetch(base + '/api/sync', { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + tk }, body: JSON.stringify({ device: 'client', tables: pushTables, tombstones }) }) let resp = await doPush(token) if (resp.status === 401) { await setSetting('sync_token', '') const token2 = await loginCloud(base, config) await setSetting('sync_token', token2) resp = await doPush(token2) } if (!resp.ok) throw new Error('同步服务返回 ' + resp.status) const data = await resp.json() const serverTables: Record = data.tables || {} const serverTombstones: { table: string; uuid: string; deleted_at: string }[] = data.tombstones || [] // 3. 应用服务端权威数据 let pulled = 0 let removed = 0 const finalMirror: { table: string; uuid: string; hash: string }[] = [] for (const t of TABLES) { const table = db.table(t) const localRows: AnyRow[] = await table.toArray() const byUuid = new Map() for (const row of localRows) byUuid.set(row.uuid, row) const serverRows: AnyRow[] = serverTables[t] || [] const serverByUuid = new Map() for (const row of serverRows) serverByUuid.set(row.uuid, row) const dead = new Set(serverTombstones.filter(x => x.table === t).map(x => x.uuid)) for (const [u, row] of byUuid) { if (dead.has(u)) { await table.delete(pkOf(t, row)) removed++ } } for (const srow of serverRows) { if (dead.has(srow.uuid)) continue const local = byUuid.get(srow.uuid) if (!local) { await table.add({ ...srow }) pulled++ } else { const localUa = String(local.updated_at || '') const serverUa = String(srow.updated_at || '') if (serverUa > localUa) { const patch: AnyRow = { ...srow } // setting 表主键为 key,禁止通过 update 修改主键字段 if (t === 'setting') delete patch.key await table.update(pkOf(t, local), patch) pulled++ } } } const finalRows: AnyRow[] = await table.toArray() for (const row of finalRows) { if (t === 'setting' && EXCLUDE_SETTING_KEYS.has(row.key)) continue finalMirror.push({ table: t, uuid: row.uuid, hash: rowHash(stripId(row)) }) } } // 4. 重建镜像 + 建立 SSE 实时通道 await db.sync_mirror.clear() if (finalMirror.length) await db.sync_mirror.bulkAdd(finalMirror.map(m => ({ ...m }))) ensureSSE(base, token) const at = nowLocalISO() await setSetting('sync_last_time', at) return { pushed, pulled, removed, at } }