// 同步引擎:服务器是唯一数据权威,本机只是实时缓存。 // 写路径:界面修改本地缓存 → 立即把差异推给服务器 → 服务器返回权威快照 → 本地缓存全量重建 // 读路径:启动 / SSE 通知 / 60 秒兜底 / 任何写操作之后 → 全量刷新 // 因此本地与服务器的数据永远强一致:没有合并、没有冲突收敛、没有双端分歧。 import { ref } from 'vue' import Dexie, { liveQuery } from 'dexie' import { db } from '../db/schema' import { getCloudCreds } from '../db/auth' import { nowLocalISO } from '../lib/date' import { notifyAttendanceChanged } from '../lib/bus' import { DEFAULT_SYNC_BASE, getSyncBase as baseOf, getToken, setToken, setSyncBaseOverride } from '../lib/server' // 最近一次自动同步的错误(设置页展示用) export const syncLastError = ref('') const TABLES = ['project', 'work_log', 'log_item', 'attendance', 'weekly_note', 'setting', 'memo', 'credential', 'goal', 'transaction', 'plan'] // 本地临时性键不参与同步,也不能从服务端写回本地缓存 const EXCLUDE_SETTING_KEYS = new Set(['sync_cursor', 'sync_last_time', 'sync_token', 'last_project', 'last_category']) // 自然键:同一逻辑记录在并发写入时的服务端去重依据(保留 updated_at 较新者) const NATURAL_KEYS: Record = { project: 'name', work_log: 'date', attendance: 'date', weekly_note: 'week_start', setting: 'key' } export const SYNC_TABLES = TABLES export { DEFAULT_SYNC_BASE, setSyncBaseOverride } export const getSyncBase = baseOf interface AnyRow { [key: string]: any } 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 = genUuid() if (!row.updated_at && row.created_at) patch.updated_at = row.created_at if (Object.keys(patch).length) await table.update(pkOf(t, row), patch) } } } function genUuid(): string { if (typeof crypto !== 'undefined' && crypto.randomUUID) return crypto.randomUUID() return 'u' + Date.now().toString(36) + Math.random().toString(36).slice(2, 10) } export interface SyncResult { pushed: number pulled: number removed: number errors: number at: string } export interface SyncConfig { baseUrl: string username: string password: string } export async function isSyncConfigured(): Promise { return getCloudCreds() != null } export async function getSyncLastTime(): Promise { try { return localStorage.getItem('dsh_last_sync') || '' } catch { return '' } } async function putSyncLastTime(at: string): Promise { try { localStorage.setItem('dsh_last_sync', at) } catch { /* 忽略 */ } } // 云端账号注册(服务端存 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 = getToken() if (token) return token token = await loginCloud(base, config) setToken(token) return token } // 登录后自动绑定云端账号:云端存在则校验密码一致(同用户名即同一数据集) export async function ensureCloudBinding(): Promise { const creds = getCloudCreds() if (!creds) throw new Error('未登录') const base = await baseOf() const resp = await fetch(base + '/api/auth/login', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ username: creds.username, password: creds.password }) }) if (!resp.ok) throw new Error('云端账号校验失败:' + resp.status) } export function getCloudUser(): string | null { const c = getCloudCreds() return c ? c.username : null } // 使用当前登录凭据执行同步(设置页与写后推送共用) export async function syncWithStoredConfig(): Promise { const creds = getCloudCreds() if (!creds) throw new Error('未登录,无法同步') return runSync({ baseUrl: await baseOf(), username: creds.username, password: creds.password }) } // 界面写操作完成后调用:立即推送差异并以服务器快照重建缓存 // 注意:绝不向界面抛出——写操作本身已在本地提交成功,同步失败只记录状态 // (设置页可见 + 下一轮自动同步续传),否则一次服务端故障会把每次点击都变成全屏报错 export async function flushAfterWrite(): Promise { try { await syncWithStoredConfig() } catch (e) { syncLastError.value = e instanceof Error ? e.message : String(e) try { await syncWithStoredConfig() } catch { /* 服务器不可达:本地暂存修改,下次自动推送或全量刷新时对齐 */ } } } // ---------- 推送规范化:剥离本地 id,外键转换为 uuid 引用 ---------- interface RefMaps { workLog: Map project: Map goal: Map } function normalizeRow(t: string, row: AnyRow, refs: RefMaps): AnyRow { const clean: AnyRow = {} for (const k of Object.keys(row)) { if (k === 'id') continue clean[k] = row[k] } if (t === 'log_item') { delete clean.work_log_id delete clean.project_id const wl = row.work_log_id ? refs.workLog.get(row.work_log_id) : null if (wl && wl.uuid) clean.wuuid = wl.uuid const pj = row.project_id ? refs.project.get(row.project_id) : null if (pj && pj.uuid) clean.puuid = pj.uuid } if (t === 'plan') { delete clean.goal_id const g = row.goal_id ? refs.goal.get(row.goal_id) : null if (g && g.uuid) clean.guuid = g.uuid } return clean } async function buildRefMaps(): Promise { const wlRows = await db.work_log.toArray() const pjRows = await db.project.toArray() const gRows = await db.goal.toArray() return { workLog: new Map(wlRows.map(r => [r.id as number, r])), project: new Map(pjRows.map(p => [p.id as number, p])), goal: new Map(gRows.map(r => [r.id as number, r])) } } // ---------- 执行同步 ---------- 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 configOf()) syncLastError.value = '' } catch (e) { syncLastError.value = e instanceof Error ? e.message : String(e) } } async function configOf(): Promise { const creds = getCloudCreds() if (!creds) throw new Error('未登录,无法同步') return { baseUrl: await baseOf(), username: creds.username, password: creds.password } } // 实时触发:本地任何数据变更 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 } realtimeAttached = false } // 服务端 SSE 通知:其他端有变更时立即刷新(服务端广播 change 即表示权威数据有变化) 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, errors: 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, epochRetry = false): Promise { const base = config.baseUrl.trim().replace(/\/+$/, '') if (!base) throw new Error('请先配置同步服务地址') await ensureSyncFields() const refs = await buildRefMaps() // 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 = normalizeRow(t, row, refs) 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] || [] } // 服务端残留临时键的清理墓碑(如早前同步上去的 sync_token 等) tombstones.push(...drainServerCleanup()) // 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) { setToken('') const token2 = await loginCloud(base, config) setToken(token2) resp = await doPush(token2) } if (!resp.ok) throw new Error('同步服务返回 ' + resp.status) const data = await resp.json() // 服务器数据纪元检测:纪元变化说明服务端数据存储被重建(如误删 data 目录后重启)。 // 此时本地镜像不可信(镜像认为「早已同步」而不推送,随后会拉到空快照把本地缓存清空)。 // 处置:作废镜像并以本地全量数据重推一次,让服务器从本设备恢复。 const epoch = String(data.dataEpoch || '') const epochKey = 'dsh_sync_epoch_' + (getCloudCreds()?.username || 'anon') const prevEpoch = epoch ? localStorage.getItem(epochKey) : null if (!epochRetry && prevEpoch && epoch && prevEpoch !== epoch) { await db.sync_mirror.clear() localStorage.setItem(epochKey, epoch) return doRunSync(config, true) } const serverTables: Record = data.tables || {} const serverTombstones: { table: string; uuid: string; deleted_at: string }[] = data.tombstones || [] // 二道防线(不依赖纪元存储,首次升级即可生效): // 本地镜像认为服务器应有 N 行,但服务器实际行数更少 —— 说明服务器数据发生过丢失/重建。 // 此时若照常拉快照,会把本地缓存也清空。处置:作废镜像并以本地全量数据重推自救。 const serverCount = TABLES.reduce((s, tn) => s + ((serverTables[tn] || []).length), 0) const expectedMin = mirrorRows.length - tombstones.length if (!epochRetry && serverCount < expectedMin) { await db.sync_mirror.clear() return doRunSync(config, true) } // 3. 用服务器快照全量重建本地缓存(本地不再有映射逻辑,服务器永远是对的) const applyResult = await applySnapshot(serverTables, serverTombstones) // 4. 建立 SSE 实时通道 + 记录同步时间 + 广播考勤变更 ensureSSE(base, getToken() || token) const at = nowLocalISO() await putSyncLastTime(at) if (epoch) localStorage.setItem(epochKey, epoch) if (applyResult.changed > 0 || applyResult.removed > 0 || pushed > 0) notifyAttendanceChanged() return { pushed, pulled: applyResult.changed, removed: applyResult.removed, errors: 0, at } } // ---------- 快照重建:清空重灌 + 稳定本地 id(uuid -> id 映射保存在本地) ---------- interface IdMap { [table: string]: { [uuid: string]: number } } function idMapKey(): string { const user = getCloudCreds()?.username || 'anon' return 'dsh_idmap_' + user } function loadIdMap(): IdMap { try { const raw = localStorage.getItem(idMapKey()) if (raw) return JSON.parse(raw) as IdMap } catch { /* 忽略 */ } return {} } function saveIdMap(map: IdMap): void { try { localStorage.setItem(idMapKey(), JSON.stringify(map)) } catch { /* 忽略 */ } } async function applySnapshot( serverTables: Record, serverTombstones: { table: string; uuid: string; deleted_at: string }[] ): Promise<{ changed: number; removed: number }> { const idMap = loadIdMap() const dead = new Set(serverTombstones.map(x => x.table + '|' + x.uuid)) const mirrorNow = new Map() for (const m of await db.sync_mirror.toArray()) mirrorNow.set(m.table + '|' + m.uuid, m) const mirrorUuids = new Set(Array.from(mirrorNow.keys()).map(k => k.split('|')[1])) let changed = 0 let removed = 0 const wlIds: { [uuid: string]: number } = {} const pjIds: { [uuid: string]: number } = {} const goalIds: { [uuid: string]: number } = {} for (const t of TABLES) { const table = db.table(t) const prevRows: AnyRow[] = await table.toArray() const prevById = new Map(prevRows.map(r => [pkOf(t, r), r])) const prevByUuid = new Map(prevRows.filter(r => r.uuid).map(r => [r.uuid as string, r])) const live: AnyRow[] = (serverTables[t] || []) .filter(r => r && typeof r === 'object' && r.uuid) .filter(r => !dead.has(t + '|' + r.uuid)) // 设置表的临时键:不写回本地缓存;顺手在服务端也清掉(下一轮推送墓碑) const filtered = live.filter(r => !(t === 'setting' && EXCLUDE_SETTING_KEYS.has(r.key))) const tmpKeys = live.filter(r => t === 'setting' && EXCLUDE_SETTING_KEYS.has(r.key)) for (const r of tmpKeys) tombstoneLater(t, r.uuid) // 在途写入保护:本次界面写操作尚未推送完成时,快照重建不得抹掉这些行 const inflight = t === 'setting' ? [] : prevRows.filter(r => r.uuid && !mirrorUuids.has(r.uuid) && !dead.has(t + '|' + r.uuid)) void inflight // 本地 id:优先沿用既有映射,保证视图引用稳定 const nextId = { v: Math.max(0, ...prevRows.map(r => Number(r.id) || 0)) } const importance = filtered.length const hydrated: AnyRow[] = [] for (const srow0 of filtered) { const srow = { ...srow0 } if (t === 'log_item') { const wlId = srow.wuuid ? wlIds[srow.wuuid] : undefined if (!wlId) continue srow.work_log_id = wlId delete srow.wuuid srow.project_id = srow.puuid ? (pjIds[srow.puuid] ?? null) : null delete srow.puuid } if (t === 'plan') { if (srow.guuid) { const gid = goalIds[srow.guuid] srow.goal_id = gid ?? null delete srow.guuid } else { srow.goal_id = null } } let id: any = pkOf(t, srow0) if (t === 'setting') id = srow0.key else { const byUuid = idMap[t] && idMap[t][srow0.uuid] const prevRow = prevByUuid.get(srow0.uuid) if (prevRow && prevRow.id != null) id = prevRow.id else if (typeof byUuid === 'number') id = byUuid else id = ++nextId.v } const rowOut: AnyRow = { ...srow, id } if (t === 'setting' && EXCLUDE_SETTING_KEYS.has(rowOut.key)) continue hydrated.push(rowOut) if (t === 'work_log') wlIds[srow0.uuid] = id if (t === 'project') pjIds[srow0.uuid] = id if (t === 'goal') goalIds[srow0.uuid] = id } void importance const outRows = hydrated.concat(inflight) // 增删差异统计:内容或 id 变化都算变更 for (const r of outRows) { const p = prevById.get(pkOf(t, r)) if (!p || rowHash(p) !== rowHash(r)) changed++ } removed += prevRows.filter(r => r.uuid && !hydrated.some(x => x.uuid === r.uuid) && !dead.has(t + '|' + r.uuid)).length // 全量重建:先清空后批量写入;在途行粘回缓存(它们会在随后的推送中上 JOIN 服务器) await table.clear() if (outRows.length) await table.bulkPut(outRows) // 更新 uuid -> id 映射 const m2: { [uuid: string]: number } = {} for (const r of outRows) { if (r.uuid) m2[r.uuid] = r.id } idMap[t] = m2 } saveIdMap(idMap) // 清理镜像与本地墓碑(快照重建后镜像直接按快照重建) const refs = await buildRefMaps() const finalMirror: { table: string; uuid: string; hash: string }[] = [] for (const t of TABLES) { const rows: AnyRow[] = await db.table(t).toArray() for (const row of rows) { if (t === 'setting' && EXCLUDE_SETTING_KEYS.has(row.key)) continue if (!row.uuid) continue finalMirror.push({ table: t, uuid: row.uuid, hash: rowHash(normalizeRow(t, row, refs)) }) } } await db.sync_mirror.clear() if (finalMirror.length) await db.sync_mirror.bulkAdd(finalMirror.map(m => ({ ...m }))) await db.tombstone.clear() return { changed, removed: removed + pendingTombstones.length } } // 服务端不应存在的临时设置键:推墓碑到服务端做一次清理 const pendingTombstones: { table: string; uuid: string; deleted_at: string }[] = [] function tombstoneLater(t: string, uuid: string): void { if (!pendingTombstones.some(x => x.table === t && x.uuid === uuid)) { pendingTombstones.push({ table: t, uuid, deleted_at: nowLocalISO() }) } } // 每轮推送时把待清理的临时键墓碑捎带上去 export function drainServerCleanup(): { table: string; uuid: string; deleted_at: string }[] { return pendingTombstones.splice(0) }