feat: 准实时同步（变更 3 秒推送 + SSE 秒级拉取）并修复云端认证

intern-log/src/App.vue

@@ -11,8 +11,7 @@ import VaultView from './views/VaultView.vue'
 import SettingsView from './views/SettingsView.vue'
 import LoginView from './views/LoginView.vue'
 import { getSession, logoutUser, openUserDatabase, type Session } from './db/auth'
-import { getSetting, } from './db/repo'
-import { isSyncConfigured, runSync } from './sync/engine'
+import { autoSyncNow, enableRealtimeSync, stopRealtime } from './sync/engine'
 import { clockIn, getAttendance, getWorkRule, getWeeklyNote, searchLogItems, searchMemos, smartClockOut, type WorkRule } from './db/repo'
 import { addDays, hhmm, todayStr, weekStart } from './lib/date'
 import { useToasts } from './lib/toast'
@@ -38,6 +37,8 @@ async function onLogin(s: Session) {
   }
 }
 async function doLogout() {
+  stopRealtime()
+  if (syncTimer !== null) { window.clearInterval(syncTimer); syncTimer = null }
   await logoutUser()
   session.value = null
   route.value = 'calendar'
@@ -45,31 +46,13 @@ async function doLogout() {
   showToast('已退出登录')
 }
 
-// ---- 自动同步：登录后与每 5 分钟，静默执行 ----
+// ---- 实时同步：本地变更 3 秒推送 + SSE 秒级拉取 + 60 秒兜底 ----
 let syncTimer: number | null = null
-let lastSyncToast = 0
-async function maybeAutoSync() {
-  if (!session.value) return
-  try {
-    if (!(await isSyncConfigured())) return
-    const result = await runSync({
-      baseUrl: await getSetting('sync_base_url', ''),
-      username: await getSetting('sync_username', ''),
-      password: await getSetting('sync_password', '')
-    })
-    if (result.pushed || result.pulled || result.removed) showToast('云同步完成：推送 ' + result.pushed + ' · 拉取 ' + result.pulled)
-  } catch (e) {
-    const now = Date.now()
-    if (now - lastSyncToast > 10 * 60 * 1000) {
-      lastSyncToast = now
-      showToast(e instanceof Error ? e.message : '自动同步失败')
-    }
-  }
-}
 function startSyncTimer() {
+  void enableRealtimeSync()
   if (syncTimer !== null) window.clearInterval(syncTimer)
-  syncTimer = window.setInterval(() => { maybeAutoSync() }, 5 * 60 * 1000)
-  maybeAutoSync()
+  syncTimer = window.setInterval(() => { void autoSyncNow() }, 60 * 1000)
+  void autoSyncNow()
 }
 
 // ---- 全局搜索 ----

intern-log/src/sync/engine.ts

@@ -1,4 +1,5 @@
 // 多端同步引擎：全量差异对比 + 记录级新者胜 + 软删除墓碑
+// 实时性：本地变更 3 秒内推送；服务端变更通过 SSE 秒级通知其他端；60 秒轮询兜底
 import { db } from '../db/schema'
 import { getSetting, setSetting } from '../db/repo'
 import { nowLocalISO } from '../lib/date'
@@ -6,7 +7,7 @@ import { nowLocalISO } from '../lib/date'
 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', 'last_project', 'last_category'])
+const EXCLUDE_SETTING_KEYS = new Set(['sync_cursor', 'sync_last_time', 'sync_token', 'last_project', 'last_category'])
 
 export const SYNC_TABLES = TABLES
 
@@ -95,8 +96,105 @@ export async function registerCloudAccount(baseUrl: string, username: string, pa
   if (!resp.ok) throw new Error('云端注册失败：' + resp.status)
 }
 
-// 执行一次完整同步
+// ---------- 云端令牌（登录换取，本地缓存，过期自动重登） ----------
+
+async function loginCloud(base: string, config: SyncConfig): Promise<string> {
+  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<string> {
+  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
+
+export async function autoSyncNow(): Promise<void> {
+  if (syncRunning) {
+    pendingSync = true
+    return
+  }
+  try {
+    if (!(await isSyncConfigured())) return
+    const values = await Promise.all([
+      getSetting('sync_base_url', ''),
+      getSetting('sync_username', ''),
+      getSetting('sync_password', '')
+    ])
+    await runSync({ baseUrl: values[0], username: values[1], password: values[2] })
+  } catch { /* 自动同步静默失败，等待下次触发 */ }
+}
+
+// 实时触发：本地任何数据变更 3 秒后自动推送
+export function enableRealtimeSync(): void {
+  if (realtimeAttached) return
+  realtimeAttached = true
+  ;(db as unknown as AnyRow).on('changes', () => {
+    if (changesSuppressed || syncRunning) return
+    if (changeDebounce) window.clearTimeout(changeDebounce)
+    changeDebounce = window.setTimeout(() => { void autoSyncNow() }, 3000)
+  })
+}
+
+export function stopRealtime(): void {
+  if (sse) { sse.close(); sse = null; sseKey = '' }
+  if (changeDebounce) { window.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<SyncResult> {
+  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
+      window.setTimeout(() => { void autoSyncNow() }, 800)
+    }
+    return result
+  } finally {
+    syncRunning = false
+    changesSuppressed = false
+  }
+}
+
+async function doRunSync(config: SyncConfig): Promise<SyncResult> {
   const base = config.baseUrl.trim().replace(/\/+$/, '')
   if (!base) throw new Error('请先配置同步服务地址')
   await ensureSyncFields()
@@ -134,21 +232,23 @@ export async function runSync(config: SyncConfig): Promise<SyncResult> {
     pushTables[t] = pushTables[t] || []
   }
 
-  // 2. 与服务端交换
-  let resp: Response
-  try {
-    resp = await fetch(base + '/api/sync', {
+  // 2. 认证并与服务端交换（token 失效自动重登一次）
+  const token = await ensureToken(base, config)
+  let resp = await fetch(base + '/api/sync', {
     method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        Authorization: 'Bearer ' + config.username.trim() + ':' + config.password
-      },
+    headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + token },
+    body: JSON.stringify({ device: 'client', tables: pushTables, tombstones })
+  })
+  if (resp.status === 401) {
+    await setSetting('sync_token', '')
+    const token2 = await loginCloud(base, config)
+    await setSetting('sync_token', token2)
+    resp = await fetch(base + '/api/sync', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', Authorization: 'Bearer ' + token2 },
       body: JSON.stringify({ device: 'client', tables: pushTables, tombstones })
     })
-  } catch {
-    throw new Error('无法访问同步服务：请检查地址、端口与网络')
   }
-  if (resp.status === 401) throw new Error('云端账号或密码不正确')
   if (!resp.ok) throw new Error('同步服务返回 ' + resp.status)
   const data = await resp.json()
   const serverTables: Record<string, AnyRow[]> = data.tables || {}
@@ -166,6 +266,8 @@ export async function runSync(config: SyncConfig): Promise<SyncResult> {
     for (const row of localRows) byUuid.set(row.uuid, row)
 
     const serverRows: AnyRow[] = serverTables[t] || []
+    const serverByUuid = new Map<string, AnyRow>()
+    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) {
@@ -196,9 +298,10 @@ export async function runSync(config: SyncConfig): Promise<SyncResult> {
     }
   }
 
-  // 4. 重建镜像
+  // 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 }

server/index.js

@@ -76,6 +76,24 @@ function base64url(s) {
   return Buffer.from(s).toString('base64url')
 }
 
+// ---------- SSE 长连接：按用户分组，数据变更时实时通知 ----------
+const sseClients = new Map()
+function broadcast(userId, event) {
+  const set = sseClients.get(userId)
+  if (!set) return
+  for (const res of set) {
+    try { res.write('event: ' + event + '\ndata: {}\n\n') } catch { /* 连接已断开 */ }
+  }
+}
+function addSse(userId, res) {
+  if (!sseClients.has(userId)) sseClients.set(userId, new Set())
+  sseClients.get(userId).add(res)
+}
+function removeSse(userId, res) {
+  const set = sseClients.get(userId)
+  if (set) { set.delete(res); if (!set.size) sseClients.delete(userId) }
+}
+
 // ---------- HTTP 基础 ----------
 function json(res, status, data) {
   const body = JSON.stringify(data)
@@ -113,6 +131,25 @@ async function handle(req, res) {
 
   if (req.method === 'GET' && route === '/api/health') { json(res, 200, { ok: true }); return }
 
+  // SSE 事件流：token 经查询参数传递（EventSource 无法自定义请求头）
+  if (req.method === 'GET' && route === '/api/events') {
+    const auth = verifyToken(url.searchParams.get('token'))
+    if (!auth) { json(res, 401, { error: '未认证' }); return }
+    res.writeHead(200, {
+      'Content-Type': 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive',
+      'Access-Control-Allow-Origin': '*'
+    })
+    res.write('event: hello\ndata: {}\n\n')
+    addSse(auth.userId, res)
+    const ka = setInterval(() => {
+      try { res.write(': keepalive\n\n') } catch { /* 忽略 */ }
+    }, 25000)
+    req.on('close', () => { clearInterval(ka); removeSse(auth.userId, res) })
+    return
+  }
+
   if (req.method === 'POST' && route === '/api/auth/register') {
     const body = await readBody(req)
     const username = String(body.username || '').trim()
@@ -149,6 +186,7 @@ async function handle(req, res) {
 
     // 应用推送的变更（新者胜）
     const incomingTables = body.tables || {}
+    let changed = 0
     for (const t of TABLES) {
       const rows = incomingTables[t] || []
       userData.tables[t] = userData.tables[t] || {}
@@ -159,6 +197,7 @@ async function handle(req, res) {
         const existing = userData.tables[t][row.uuid]
         if (!existing || String(clean.updated_at || '') >= String(existing.updated_at || '')) {
           userData.tables[t][row.uuid] = clean
+          changed++
         }
       }
     }
@@ -168,7 +207,10 @@ async function handle(req, res) {
       if (!TABLES.includes(tb.table) || !tb.uuid) continue
       userData.tombstones[tb.table] = userData.tombstones[tb.table] || {}
       const prev = userData.tombstones[tb.table][tb.uuid]
-      if (!prev || tb.deleted_at > prev) userData.tombstones[tb.table][tb.uuid] = tb.deleted_at
+      if (!prev || tb.deleted_at > prev) {
+        userData.tombstones[tb.table][tb.uuid] = tb.deleted_at
+        changed++
+      }
     }
     // 墓碑生效：清除被删数据
     for (const t of TABLES) {
@@ -179,6 +221,7 @@ async function handle(req, res) {
       }
     }
     saveUserData(auth.userId, userData)
+    if (changed) broadcast(auth.userId, 'change')
 
     // 返回权威全量
     const outTables = {}

实习工作日志系统-设计方案.md

@@ -533,3 +533,4 @@ src/
 - v1.13：多端同步与安卓应用——自建零依赖同步服务端（Docker 部署、按用户 JSON 隔离、HMAC token）；客户端同步引擎（uuid 与 updated_at 补齐、镜像哈希差异、墓碑传播、记录级新者胜）；设置页同步配置与自动同步；Capacitor 安卓工程脚手架与移动端 UI 适配（安全区、输入字号）
 - v1.13.1：登录会话持久化（localStorage，关闭浏览器后免登录，无需每次使用都登录）；部署方案调整为 Docker + Caddy 自动 HTTPS（独立子域名与现有域名路径两种反代方式）；移除 30 分钟无操作自动退出（密码库仍保留 5 分钟自动锁定与主密码门槛）
 - v1.13.2：同步服务独立子域名部署 daywk.linyux.cn（Caddy 新站点块，客户端同步地址 https://daywk.linyux.cn）
+- v1.13.3：准实时同步——本地数据变更 3 秒内自动推送（Dexie 变更事件触发）；服务端新增 SSE 事件流，其他端变更秒级拉取；60 秒轮询兜底；修复引擎与服务端认证方式不匹配问题（登录换取 HMAC token 并缓存，失效自动重登）；同步并发加锁防重入