package main // sync.go 实现登录 + HTTP 同步引擎(本地优先): // 推送本地 dirty 行 → 按 updated_at 增量拉取远端 → Last-Write-Wins 冲突合并。 // 远端经 nl-pms-api(JWT);本地仍用 SQLite。触发:登录成功、每 5 分钟巡检、手动同步。 import ( "database/sql" "encoding/hex" "encoding/json" "errors" "fmt" "net/http" "net/url" "path/filepath" "strconv" "strings" "time" ) const syncTimeout = 10 * time.Second // ---------- 配置与状态 ---------- // syncConfig 解析 API 基址。启动时写入本地 meta(见 applyPackagedSyncConfig); // 运行期以 meta 为准,meta 为空时回退到打包嵌入配置。 func (a *App) syncConfig() SyncConfig { c := SyncConfig{BaseURL: strings.TrimRight(strings.TrimSpace(a.store.Meta("sync_base_url")), "/")} if syncConfigComplete(c) { return c } pkg := packagedSyncDefaults() if syncConfigComplete(pkg) { return pkg } return SyncConfig{} } // applyPackagedSyncConfig 启动时用打包配置覆盖本地 SQLite 中的 API 基址, // 避免旧安装残留的 MySQL 连接 meta 继续生效。 func (a *App) applyPackagedSyncConfig() { if a.store == nil { return } pkg := packagedSyncDefaults() if !syncConfigComplete(pkg) { return } _ = a.store.SetMeta("sync_base_url", strings.TrimRight(strings.TrimSpace(pkg.BaseURL), "/")) } func syncConfigComplete(c SyncConfig) bool { return strings.TrimSpace(c.BaseURL) != "" } func (a *App) GetSyncConfig() (SyncConfig, error) { if e := a.ready(); e != nil { return SyncConfig{}, e } return a.syncConfig(), nil } func (a *App) SaveSyncConfig(c SyncConfig) error { if e := a.ready(); e != nil { return e } u := strings.TrimRight(strings.TrimSpace(c.BaseURL), "/") return a.store.SetMeta("sync_base_url", u) } func (a *App) syncUserID() int64 { id, _ := strconv.ParseInt(a.store.Meta("sync_user_id"), 10, 64) return id } func (a *App) GetSyncStatus() (SyncStatus, error) { if e := a.ready(); e != nil { return SyncStatus{}, e } a.syncMu.Lock() lastErr := a.syncLastErr a.syncMu.Unlock() var pending int _ = a.store.db.QueryRow(`SELECT (SELECT COUNT(*) FROM todos WHERE dirty=1) +(SELECT COUNT(*) FROM tickets WHERE dirty=1) +(SELECT COUNT(*) FROM notes WHERE dirty=1)`).Scan(&pending) return SyncStatus{ Configured: syncConfigComplete(a.syncConfig()), LoggedIn: a.syncUserID() > 0 && a.store.Meta("sync_access_token") != "", UserID: a.syncUserID(), Username: a.store.Meta("sync_username"), LastSyncAt: a.store.Meta("sync_last_at"), Syncing: a.syncing.Load(), Online: a.syncOnline.Load(), LastError: lastErr, Pending: pending, }, nil } // ---------- 注册 / 登录 / 登出 ---------- func validAccount(username, password string) error { username = strings.TrimSpace(username) if len(username) < 3 || len(username) > 64 { return errors.New("SYNC_USERNAME_INVALID") } if len(password) < 6 { return errors.New("SYNC_PASSWORD_TOO_SHORT") } return nil } func (a *App) SyncRegister(username, password string) error { if e := a.ready(); e != nil { return e } if e := validAccount(username, password); e != nil { return e } if e := a.apiDecode(http.MethodPost, "/api/v1/auth/register", map[string]string{ "username": strings.TrimSpace(username), "password": password, }, nil, false); e != nil { return e } a.ensureDefaultAvatar() return nil } func (a *App) SyncLogin(username, password string) (SyncStatus, error) { if e := a.ready(); e != nil { return SyncStatus{}, e } username = strings.TrimSpace(username) var out struct { AccessToken string `json:"accessToken"` RefreshToken string `json:"refreshToken"` UserID int64 `json:"userId"` Username string `json:"username"` } if e := a.apiDecode(http.MethodPost, "/api/v1/auth/login", map[string]string{ "username": username, "password": password, }, &out, false); e != nil { return SyncStatus{}, e } if out.AccessToken == "" || out.UserID <= 0 { return SyncStatus{}, errors.New("SYNC_BAD_CREDENTIALS") } id := out.UserID if uname := strings.TrimSpace(out.Username); uname != "" { username = uname } // 切换账号时重置拉取游标,并把本地数据整体标脏,合并进新账号。 if prev := a.store.Meta("sync_last_user_id"); prev != "" && prev != strconv.FormatInt(id, 10) { for _, t := range []string{"todos", "tickets", "notes"} { _ = a.store.SetMeta("sync_pull_"+t, "") _, _ = a.store.db.Exec(`UPDATE ` + t + ` SET dirty=1`) } } _ = a.store.SetMeta("sync_access_token", out.AccessToken) _ = a.store.SetMeta("sync_refresh_token", out.RefreshToken) _ = a.store.SetMeta("sync_user_id", strconv.FormatInt(id, 10)) _ = a.store.SetMeta("sync_last_user_id", strconv.FormatInt(id, 10)) _ = a.store.SetMeta("sync_username", username) // 派生 API Key 加密密钥(可选功能,失败不阻断登录,只记录日志)。 if ee := a.ensureEncKey(password); ee != nil { _ = a.store.SetMeta("sync_enc_key", "") a.store.Log("warning", "同步", "API Key 加密密钥派生失败", ee.Error()) } a.store.Log("info", "同步", "登录成功", username) a.RefreshShell() // 原生账号菜单切换为已登录形态 go a.syncOnce(true) return a.GetSyncStatus() } // SyncChangePassword 在线修改登录密码。加密密钥由密码派生,因此会同步轮换: // 先调 API 改密,再用 HTTP 读写 settings 把 api_keys 密文换成新密钥。 // 其他已登录设备的旧密钥会失效,需用新密码重新登录。 func (a *App) SyncChangePassword(oldPassword, newPassword string) error { if e := a.ready(); e != nil { return e } userID := a.syncUserID() if userID <= 0 { return errors.New("SYNC_NOT_LOGGED_IN") } if len(newPassword) < 6 { return errors.New("SYNC_PASSWORD_TOO_SHORT") } if a.store.Meta("sync_access_token") == "" { return errors.New("SYNC_NOT_LOGGED_IN") } if e := a.apiDecode(http.MethodPost, "/api/v1/auth/change-password", map[string]string{ "oldPassword": oldPassword, "newPassword": newPassword, }, nil, true); e != nil { return e } // 盐保持不变,新密码 + 原盐派生新密钥;缺盐(从未用过加密同步)时跳过密钥轮换。 var newKey []byte saltVal, _, saltOK, se := a.apiGetSetting("enc_salt") if se != nil { return se } if saltOK { if raw, de := hex.DecodeString(strings.TrimSpace(saltVal)); de == nil && len(raw) >= 8 { newKey = deriveEncKey(newPassword, raw) } } if newKey != nil { blob, _, hasKeys, be := a.apiGetSetting("api_keys") if be != nil { return be } if hasKeys { plain, de := "", errors.New("SYNC_DECRYPT_FAILED") fromRemote := false if old := a.encKey(); old != nil { plain, de = decryptWithKey(old, blob) fromRemote = de == nil } if de != nil { // 旧密钥不可用时用本地明文 Key 重建密文;本地也没有就删除该行(写空并新时间戳无意义,跳过 PUT)。 if st, le := a.store.Settings(); le == nil && (st.SparkKey != "" || st.DeepSeekKey != "") { b, _ := json.Marshal(map[string]string{"sparkKey": st.SparkKey, "deepSeekKey": st.DeepSeekKey}) plain, de = string(b), nil } } switch { case de == nil && fromRemote: enc, ee := encryptWithKey(newKey, plain) if ee != nil { return ee } // 内容未变只换密文:保留原时间戳,LWW 语义完全不受影响。 _, at, _, _ := a.apiGetSetting("api_keys") if at == "" { at = nowRFC() } if e := a.apiPutSetting("api_keys", enc, at); e != nil { return e } case de == nil: enc, ee := encryptWithKey(newKey, plain) if ee != nil { return ee } if e := a.apiPutSetting("api_keys", enc, nowRFC()); e != nil { return e } default: // 无法解密也无法本地重建:用空值覆盖,避免旧密文永久卡死。 if e := a.apiPutSetting("api_keys", "", nowRFC()); e != nil { return e } } } _ = a.store.SetMeta("sync_enc_key", hex.EncodeToString(newKey)) } else { _ = a.store.SetMeta("sync_enc_key", "") } a.store.Log("info", "同步", "密码修改成功", a.store.Meta("sync_username")) return nil } func (a *App) SyncLogout() error { if e := a.ready(); e != nil { return e } _ = a.store.SetMeta("sync_user_id", "") _ = a.store.SetMeta("sync_username", "") _ = a.store.SetMeta("sync_enc_key", "") _ = a.store.SetMeta("sync_access_token", "") _ = a.store.SetMeta("sync_refresh_token", "") a.syncOnline.Store(false) a.RefreshShell() // 原生账号菜单切换为未登录形态 return nil } // ---------- 同步引擎 ---------- // SyncNow 手动触发一次同步,同步完成后返回最新状态。 func (a *App) SyncNow() (SyncStatus, error) { if e := a.ready(); e != nil { return SyncStatus{}, e } if a.syncUserID() <= 0 { return SyncStatus{}, errors.New("SYNC_NOT_LOGGED_IN") } pushed, pulled, e := a.syncOnce(false) st, _ := a.GetSyncStatus() st.Pushed, st.Pulled = pushed, pulled if e != nil { return st, e } return st, nil } // runSyncLoop 启动时立即尝试同步一次(在线优先),失败自动降级为本地模式; // 之后每 5 分钟巡检:连接成功即视为在线,失败静默跳过(断网恢复后自然补同步)。 func (a *App) runSyncLoop() { if a.bootstrap.State == BootstrapReady && a.syncUserID() > 0 { _, _, _ = a.syncOnce(true) } t := time.NewTicker(5 * time.Minute) defer t.Stop() for { select { case <-a.ctx.Done(): return case <-t.C: if a.bootstrap.State == BootstrapReady && a.syncUserID() > 0 { _, _, _ = a.syncOnce(true) } } } } // syncOnce 执行一轮推送+拉取。silent 模式下网络错误不写消息中心。 func (a *App) syncOnce(silent bool) (pushed, pulled int, err error) { if !a.syncing.CompareAndSwap(false, true) { return 0, 0, errors.New("SYNC_IN_PROGRESS") } defer a.syncing.Store(false) userID := a.syncUserID() if userID <= 0 || a.store.Meta("sync_access_token") == "" { return 0, 0, errors.New("SYNC_NOT_LOGGED_IN") } if !syncConfigComplete(a.syncConfig()) { a.syncOnline.Store(false) e := errors.New("SYNC_NOT_CONFIGURED") a.setSyncErr(e.Error()) a.emitSyncDone() return 0, 0, e } wasOnline := a.syncOnline.Load() for _, t := range []syncTable{todoSync, ticketSync, noteSync} { p, e := a.pushTable(t) if e != nil { a.syncOnline.Store(false) a.setSyncErr(e.Error()) if !silent { a.pushMessage("sync", a.syncText("fail"), e.Error(), "", 0, false) } a.emitSyncDone() return pushed, pulled, e } pushed += p g, e := a.pullTable(t) if e != nil { a.syncOnline.Store(false) a.setSyncErr(e.Error()) if !silent { a.pushMessage("sync", a.syncText("fail"), e.Error(), "", 0, false) } a.emitSyncDone() return pushed, pulled, e } pulled += g } a.syncOnline.Store(true) // API Key 加密同步(设置里开启后才执行)。 if st, se := a.store.Settings(); se == nil && st.SyncAPIKeys { kp, kg, ke := a.syncAPIKeys(st) if ke != nil { a.setSyncErr(ke.Error()) a.emitSyncDone() return pushed, pulled, ke } pushed += kp pulled += kg } if ap, ag, ae := a.syncAvatar(); ae != nil { a.setSyncErr(ae.Error()) a.emitSyncDone() return pushed, pulled, ae } else { pushed += ap pulled += ag } // paths: 行先于身份文档处理:本机路径先落库,身份合并后待绑定清单才准确。 for _, d := range []struct { name string gen func() (string, error) apply func(string) error }{ {"paths:" + a.machineID(), a.machinePathsDoc, a.applyMachinePaths}, {"projects", a.projectsDoc, a.applyProjectsDoc}, {"rules", a.rulesDoc, a.applyRulesDoc}, } { dp, dg, de := a.syncDoc(d.name, d.gen, d.apply) if de != nil { a.setSyncErr(de.Error()) a.emitSyncDone() return pushed, pulled, de } pushed += dp pulled += dg } if fp, fg, fe := a.syncFestivalImages(); fe != nil { a.setSyncErr(fe.Error()) a.emitSyncDone() return pushed, pulled, fe } else { pushed += fp pulled += fg } if sp, sg, se := a.syncFileStorageConfig(); se != nil { a.setSyncErr(se.Error()) a.emitSyncDone() return pushed, pulled, se } else { pushed += sp pulled += sg } a.syncUserProfile() pulled += a.pullTeamNotices() a.refreshTeamAssignedCache() _ = a.store.SetMeta("sync_last_at", nowRFC()) a.setSyncErr("") if pushed > 0 || pulled > 0 || !wasOnline { a.store.Log("info", "同步", "同步完成", fmt.Sprintf("推送 %d 条 | 拉取 %d 条", pushed, pulled)) if pushed > 0 || pulled > 0 { a.pushMessage("sync", a.syncText("done"), fmt.Sprintf(a.syncText("body"), pushed, pulled), "", 0, false) } } a.emitSyncDone() return pushed, pulled, nil } func (a *App) setSyncErr(v string) { a.syncMu.Lock() a.syncLastErr = v a.syncMu.Unlock() } func (a *App) emitSyncDone() { if st, e := a.GetSyncStatus(); e == nil { a.emit("sync:done", st) } a.RefreshTrayStatus() } func (a *App) syncText(k string) string { locale := "zh-CN" if st, e := a.store.Settings(); e == nil { locale = st.Locale } if locale == "en" { return map[string]string{"done": "Sync finished", "fail": "Sync failed", "body": "pushed %d, pulled %d"}[k] } return map[string]string{"done": "同步完成", "fail": "同步失败", "body": "推送 %d 条,拉取 %d 条"}[k] } // ---------- API Key 加密同步(可选) ---------- // ensureEncKey 登录时从远端取(或初始化)每用户加密盐,用登录密码派生 AES 密钥并缓存到本地。 func (a *App) ensureEncKey(password string) error { salt, _, ok, e := a.apiGetSetting("enc_salt") if e != nil { return e } if !ok { if salt, e = newSaltHex(); e != nil { return e } if e = a.apiPutSetting("enc_salt", salt, nowRFC()); e != nil { return e } // 并发写入时以真正落库的盐为准。 salt, _, ok, e = a.apiGetSetting("enc_salt") if e != nil { return e } if !ok { return errors.New("SYNC_DECRYPT_FAILED") } } raw, e := hex.DecodeString(strings.TrimSpace(salt)) if e != nil || len(raw) < 8 { return errors.New("SYNC_DECRYPT_FAILED") } return a.store.SetMeta("sync_enc_key", hex.EncodeToString(deriveEncKey(password, raw))) } // encKey 返回本地缓存的加密密钥;未登录或未派生成功时为 nil。 func (a *App) encKey() []byte { raw, e := hex.DecodeString(a.store.Meta("sync_enc_key")) if e != nil || len(raw) != 32 { return nil } return raw } // syncAPIKeys 双向同步加密后的 AI API Key(单行 JSON 载荷,LWW 按 RFC3339 时间戳判定)。 func (a *App) syncAPIKeys(st AppSettings) (pushed, pulled int, err error) { key := a.encKey() if key == nil { return 0, 0, errors.New("SYNC_DECRYPT_FAILED") } localAt := a.store.Meta("api_keys_updated_at") if localAt == "" && (st.SparkKey != "" || st.DeepSeekKey != "") { localAt = nowRFC() _ = a.store.SetMeta("api_keys_updated_at", localAt) } remoteVal, remoteAt, hasRemote, e := a.apiGetSetting("api_keys") if e != nil { return 0, 0, e } switch { case localAt != "" && (!hasRemote || localAt > remoteAt): payload, e := json.Marshal(map[string]string{"sparkKey": st.SparkKey, "deepSeekKey": st.DeepSeekKey}) if e != nil { return 0, 0, e } enc, e := encryptWithKey(key, string(payload)) if e != nil { return 0, 0, e } if e = a.apiPutSetting("api_keys", enc, localAt); e != nil { return 0, 0, e } pushed = 1 case hasRemote && remoteAt > localAt: plain, e := decryptWithKey(key, remoteVal) if e != nil { return 0, 0, errors.New("SYNC_DECRYPT_FAILED") } var m map[string]string if e := json.Unmarshal([]byte(plain), &m); e != nil { return 0, 0, errors.New("SYNC_DECRYPT_FAILED") } if m["sparkKey"] != st.SparkKey || m["deepSeekKey"] != st.DeepSeekKey { st.SparkKey, st.DeepSeekKey = m["sparkKey"], m["deepSeekKey"] if e := a.store.SaveSettings(st); e != nil { return 0, 0, e } } _ = a.store.SetMeta("api_keys_updated_at", remoteAt) pulled = 1 } return pushed, pulled, nil } // syncAvatar 同步头像(settings name='avatar',明文 JSON {mode,value},LWW)。 // path 模式是设备本地行为:既不推送,也不参与拉取覆盖。 func (a *App) syncAvatar() (pushed, pulled int, err error) { st, e := a.store.Settings() if e != nil { return 0, 0, e } if st.AvatarMode == "path" { return 0, 0, nil } localAt := a.store.Meta("avatar_updated_at") if localAt == "" && st.AvatarMode != "" { localAt = nowRFC() _ = a.store.SetMeta("avatar_updated_at", localAt) } remoteVal, remoteAt, hasRemote, e := a.apiGetSetting("avatar") if e != nil { return 0, 0, e } switch { case localAt != "" && (!hasRemote || localAt > remoteAt): payload, e := json.Marshal(map[string]string{"mode": st.AvatarMode, "value": st.AvatarValue}) if e != nil { return 0, 0, e } if e = a.apiPutSetting("avatar", string(payload), localAt); e != nil { return 0, 0, e } pushed = 1 case hasRemote && remoteAt > localAt: var m map[string]string if json.Unmarshal([]byte(remoteVal), &m) != nil { return 0, 0, nil } mode := m["mode"] if mode != "" && mode != "base64" && mode != "url" { return 0, 0, nil } if mode != st.AvatarMode || m["value"] != st.AvatarValue { st.AvatarMode, st.AvatarValue = mode, m["value"] if e := a.store.SaveSettings(st); e != nil { return 0, 0, e } } _ = a.store.SetMeta("avatar_updated_at", remoteAt) pulled = 1 } return pushed, pulled, nil } // ---------- 项目/分组/收藏/排除规则文档同步 ---------- type projectDocEntry struct { Name string `json:"name"` Path string `json:"path,omitempty"` Description string `json:"description,omitempty"` Group string `json:"group,omitempty"` Favorite bool `json:"favorite,omitempty"` } type ruleDocEntry struct { Pattern string `json:"pattern"` Category string `json:"category"` } func (a *App) projectsDoc() (string, error) { rows, e := a.store.db.Query(`SELECT p.name,p.description,COALESCE(g.name,''), EXISTS(SELECT 1 FROM favorites f WHERE f.project_id=p.id) FROM projects p LEFT JOIN project_groups g ON g.id=p.group_id ORDER BY p.name`) if e != nil { return "", e } defer rows.Close() list := []projectDocEntry{} for rows.Next() { var it projectDocEntry if e := rows.Scan(&it.Name, &it.Description, &it.Group, &it.Favorite); e != nil { return "", e } list = append(list, it) } b, e := json.Marshal(list) return string(b), e } func (a *App) ensureGroupID(group string) (int64, error) { gid := int64(1) g := strings.TrimSpace(group) if g == "" { return gid, nil } e := a.store.db.QueryRow(`SELECT id FROM project_groups WHERE name=?`, g).Scan(&gid) if e == sql.ErrNoRows { now := nowRFC() r, e2 := a.store.db.Exec(`INSERT INTO project_groups(name,created_at,updated_at) VALUES(?,?,?)`, g, now, now) if e2 != nil { return 0, e2 } gid, _ = r.LastInsertId() return gid, nil } return gid, e } func (a *App) applyProjectsDoc(doc string) error { var list []projectDocEntry if json.Unmarshal([]byte(doc), &list) != nil { return nil } for _, it := range list { name := strings.TrimSpace(it.Name) if name == "" { if p := strings.TrimSpace(it.Path); p != "" { name = filepath.Base(p) } else { continue } } gid, e := a.ensureGroupID(it.Group) if e != nil { return e } var pid int64 e = a.store.db.QueryRow(`SELECT id FROM projects WHERE name=? LIMIT 1`, name).Scan(&pid) if e == sql.ErrNoRows { if p := strings.TrimSpace(it.Path); p != "" { e = a.store.db.QueryRow(`SELECT id FROM projects WHERE path=? LIMIT 1`, p).Scan(&pid) } } switch { case e == sql.ErrNoRows: continue case e != nil: return e } if _, e := a.store.db.Exec(`UPDATE projects SET name=?,description=?,group_id=?,updated_at=? WHERE id=? AND (name!=? OR description!=? OR group_id!=?)`, name, it.Description, gid, nowRFC(), pid, name, it.Description, gid); e != nil { return e } if it.Favorite { _, _ = a.store.db.Exec(`INSERT OR IGNORE INTO favorites(project_id,created_at) VALUES(?,?)`, pid, nowRFC()) } else { _, _ = a.store.db.Exec(`DELETE FROM favorites WHERE project_id=?`, pid) } } a.rebuildCloudPending(list) return nil } func (a *App) rulesDoc() (string, error) { rows, e := a.store.db.Query(`SELECT pattern,category FROM exclusion_rules WHERE builtin=0 ORDER BY pattern`) if e != nil { return "", e } defer rows.Close() list := []ruleDocEntry{} for rows.Next() { var it ruleDocEntry if e := rows.Scan(&it.Pattern, &it.Category); e != nil { return "", e } list = append(list, it) } b, e := json.Marshal(list) return string(b), e } func (a *App) applyRulesDoc(doc string) error { var list []ruleDocEntry if json.Unmarshal([]byte(doc), &list) != nil { return nil } tx, e := a.store.db.Begin() if e != nil { return e } defer tx.Rollback() if _, e := tx.Exec(`DELETE FROM exclusion_rules WHERE builtin=0`); e != nil { return e } for _, it := range list { p := strings.TrimSpace(it.Pattern) if p == "" { continue } c := it.Category if c == "" { c = "custom" } if _, e := tx.Exec(`INSERT OR IGNORE INTO exclusion_rules(pattern,category,builtin) VALUES(?,?,0)`, p, c); e != nil { return e } } return tx.Commit() } // syncDoc 单个文档行的双向同步:远端较新先套用合并,本地与远端内容不一致再回推。 func (a *App) syncDoc(name string, gen func() (string, error), apply func(string) error) (pushed, pulled int, err error) { doc, e := gen() if e != nil { return 0, 0, e } remoteVal, remoteAt, hasRemote, e := a.apiGetSetting(name) if e != nil { return 0, 0, e } metaKey := "sync_doc_" + name + "_at" lastAt := a.store.Meta(metaKey) if hasRemote && remoteAt > lastAt { if e := apply(remoteVal); e != nil { return 0, 0, e } if doc, e = gen(); e != nil { return 0, 0, e } lastAt = remoteAt pulled = 1 } if (hasRemote && doc != remoteVal) || (!hasRemote && doc != "[]" && doc != "{}") { now := nowRFC() if e = a.apiPutSetting(name, doc, now); e != nil { return 0, 0, e } lastAt = now pushed = 1 } if lastAt != "" { _ = a.store.SetMeta(metaKey, lastAt) } return pushed, pulled, nil } // syncFestivalImages 同步节日背景图:云端键 fest_img:<节日名>。 // 管理员推送本地 dirty 行;所有账号拉取(API 会从管理员名下读全局前缀)。 func (a *App) syncFestivalImages() (pushed, pulled int, err error) { const prefix = "fest_img:" if a.syncUserID() == festivalAdminID { rows, e := a.store.db.Query(`SELECT fest_key,value,updated_at FROM festival_images WHERE dirty=1`) if e != nil { return 0, 0, e } type rec struct{ k, v, at string } recs := []rec{} for rows.Next() { var r rec if e := rows.Scan(&r.k, &r.v, &r.at); e != nil { rows.Close() return 0, 0, e } recs = append(recs, r) } rows.Close() for _, r := range recs { if e := a.apiPutSettingWithStepUp(prefix+r.k, r.v, r.at, true); e != nil { // 无动态码时跳过推送,保留 dirty,待管理员验证后再同步 if e.Error() == "ADMIN_STEPUP_REQUIRED" || e.Error() == "ADMIN_IP_CHANGED" { continue } return pushed, pulled, e } if _, e := a.store.db.Exec(`UPDATE festival_images SET dirty=0 WHERE fest_key=? AND updated_at=?`, r.k, r.at); e != nil { return pushed, pulled, e } pushed++ } } items, e := a.apiListSettings(prefix) if e != nil { return pushed, pulled, e } cursor := a.store.Meta("sync_pull_fest_imgs") maxSeen := cursor for _, it := range items { at := it.UpdatedAt if at <= cursor { continue } if at > maxSeen { maxSeen = at } key := strings.TrimPrefix(it.Name, prefix) var localAt string ge := a.store.db.QueryRow(`SELECT updated_at FROM festival_images WHERE fest_key=?`, key).Scan(&localAt) if ge == nil && localAt >= at { continue } if _, e := a.store.db.Exec(`INSERT INTO festival_images(fest_key,value,updated_at,dirty) VALUES(?,?,?,0) ON CONFLICT(fest_key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at,dirty=0`, key, it.Value, at); e != nil { return pushed, pulled, e } pulled++ } if maxSeen != cursor { _ = a.store.SetMeta("sync_pull_fest_imgs", maxSeen) } return pushed, pulled, nil } // syncFileStorageConfig 同步全局文件存储配置(settings name='file_storage')。 func (a *App) syncFileStorageConfig() (pushed, pulled int, err error) { localAt := a.store.Meta(fileStorageAtKey) remoteVal, remoteAt, hasRemote, e := a.apiGetGlobalSetting(fileStorageKey) if e != nil { return 0, 0, e } switch { case a.syncUserID() == festivalAdminID && localAt != "" && (!hasRemote || localAt > remoteAt): b, e := json.Marshal(a.fileStorageConfig()) if e != nil { return 0, 0, e } if e = a.apiPutSettingWithStepUp(fileStorageKey, string(b), localAt, true); e != nil { if e.Error() == "ADMIN_STEPUP_REQUIRED" || e.Error() == "ADMIN_IP_CHANGED" { return 0, 0, nil } return 0, 0, e } pushed = 1 case hasRemote && remoteAt > localAt: var c FileStorageConfig if json.Unmarshal([]byte(remoteVal), &c) != nil { return 0, 0, nil } b, e := json.Marshal(normalizeFileStorage(c)) if e != nil { return 0, 0, e } if e := a.store.SetMeta(fileStorageKey, string(b)); e != nil { return 0, 0, e } _ = a.store.SetMeta(fileStorageAtKey, remoteAt) pulled = 1 } return pushed, pulled, nil } // syncTable 描述一张可同步表的本地列映射。 type syncTable struct { local string // localCols 顺序即 scan 顺序;project_id 与 project_name 的转换单独处理。 hasProject bool hasTimes bool // 有 created_at(notes 没有) extraCols []string } var ( todoSync = syncTable{local: "todos", hasProject: true, hasTimes: true, extraCols: []string{"title", "content", "due_at", "priority", "status", "history", "team_id"}} ticketSync = syncTable{local: "tickets", hasProject: true, hasTimes: true, extraCols: []string{"title", "description", "type", "start_at", "due_at", "status", "priority", "history", "team_id"}} noteSync = syncTable{local: "notes", extraCols: []string{"content"}} ) // pushTable 收集本地 dirty 行,POST /sync/push 后清 dirty。 func (a *App) pushTable(t syncTable) (int, error) { cols := append([]string{"uuid"}, t.extraCols...) if t.hasProject { cols = append(cols, "project_id") } if t.hasTimes { cols = append(cols, "created_at") } cols = append(cols, "updated_at", "deleted") rows, e := a.store.db.Query(`SELECT ` + strings.Join(cols, ",") + ` FROM ` + t.local + ` WHERE dirty=1`) if e != nil { return 0, e } type rec struct { vals []any uuid string } recs := []rec{} for rows.Next() { vals := make([]any, len(cols)) ptrs := make([]any, len(cols)) for i := range vals { ptrs[i] = &vals[i] } if e := rows.Scan(ptrs...); e != nil { rows.Close() return 0, e } recs = append(recs, rec{vals: vals, uuid: asString(vals[0])}) } rows.Close() if len(recs) == 0 { return 0, nil } payload := make([]map[string]any, 0, len(recs)) uuids := make([]string, 0, len(recs)) for _, r := range recs { m := map[string]any{} for i, c := range cols { if c == "project_id" { m["project_name"] = a.projectNameByID(asInt64(r.vals[i])) continue } if c == "deleted" || c == "team_id" { m[c] = asInt64(r.vals[i]) continue } m[c] = asString(r.vals[i]) } payload = append(payload, m) uuids = append(uuids, r.uuid) } var resp struct { Pushed int `json:"pushed"` } if e := a.apiDecode(http.MethodPost, "/api/v1/sync/push", map[string]any{ "table": t.local, "rows": payload, }, &resp, true); e != nil { return 0, e } for _, u := range uuids { if _, e := a.store.db.Exec(`UPDATE `+t.local+` SET dirty=0 WHERE uuid=? AND dirty=1`, u); e != nil { return 0, e } } return len(uuids), nil } // pullTable 增量拉取远端更新(LWW:远端 updated_at 更新才覆盖本地)。 func (a *App) pullTable(t syncTable) (int, error) { cursorKey := "sync_pull_" + t.local cursor := a.store.Meta(cursorKey) q := "/api/v1/sync/pull?table=" + url.QueryEscape(t.local) + "&cursor=" + url.QueryEscape(cursor) var resp struct { Rows []map[string]any `json:"rows"` } if e := a.apiDecode(http.MethodGet, q, nil, &resp, true); e != nil { return 0, e } n, maxSeen := 0, cursor for _, m := range resp.Rows { updated := asString(m["updated_at"]) if updated > maxSeen { maxSeen = updated } applied, e := a.applyRemoteRow(t, m) if e != nil { return n, e } if applied { n++ } } if maxSeen != cursor { _ = a.store.SetMeta(cursorKey, maxSeen) } return n, nil } // applyRemoteRow 将一条远端行按 LWW 合并进本地表。 func (a *App) applyRemoteRow(t syncTable, m map[string]any) (bool, error) { uuid := asString(m["uuid"]) var localID int64 var localUpdated string e := a.store.db.QueryRow(`SELECT id,updated_at FROM `+t.local+` WHERE uuid=?`, uuid).Scan(&localID, &localUpdated) exists := e == nil remoteUpdated := asString(m["updated_at"]) if exists && remoteUpdated <= localUpdated { return false, nil } deleted := asInt64(m["deleted"]) switch t.local { case "todos": pid := a.projectIDByName(asString(m["project_name"])) if exists { _, e = a.store.db.Exec(`UPDATE todos SET title=?,content=?,project_id=?,due_at=?,priority=?,status=?,history=?,team_id=?,reminded=0,updated_at=?,deleted=?,dirty=0 WHERE id=?`, asString(m["title"]), asString(m["content"]), pid, asString(m["due_at"]), asString(m["priority"]), asString(m["status"]), asString(m["history"]), asInt64(m["team_id"]), remoteUpdated, deleted, localID) } else { _, e = a.store.db.Exec(`INSERT INTO todos(uuid,title,content,project_id,due_at,priority,status,history,team_id,reminded,created_at,updated_at,deleted,dirty) VALUES(?,?,?,?,?,?,?,?,?,0,?,?,?,0)`, uuid, asString(m["title"]), asString(m["content"]), pid, asString(m["due_at"]), asString(m["priority"]), asString(m["status"]), asString(m["history"]), asInt64(m["team_id"]), asString(m["created_at"]), remoteUpdated, deleted) } case "tickets": pid := a.projectIDByName(asString(m["project_name"])) if exists { _, e = a.store.db.Exec(`UPDATE tickets SET title=?,description=?,type=?,project_id=?,start_at=?,due_at=?,status=?,priority=?,history=?,team_id=?,reminded=0,updated_at=?,deleted=?,dirty=0 WHERE id=?`, asString(m["title"]), asString(m["description"]), asString(m["type"]), pid, asString(m["start_at"]), asString(m["due_at"]), asString(m["status"]), asString(m["priority"]), asString(m["history"]), asInt64(m["team_id"]), remoteUpdated, deleted, localID) } else { _, e = a.store.db.Exec(`INSERT INTO tickets(uuid,title,description,type,project_id,start_at,due_at,status,priority,history,team_id,reminded,created_at,updated_at,deleted,dirty) VALUES(?,?,?,?,?,?,?,?,?,?,?,0,?,?,?,0)`, uuid, asString(m["title"]), asString(m["description"]), asString(m["type"]), pid, asString(m["start_at"]), asString(m["due_at"]), asString(m["status"]), asString(m["priority"]), asString(m["history"]), asInt64(m["team_id"]), asString(m["created_at"]), remoteUpdated, deleted) } case "notes": if exists { _, e = a.store.db.Exec(`UPDATE notes SET content=?,updated_at=?,deleted=?,dirty=0 WHERE id=?`, asString(m["content"]), remoteUpdated, deleted, localID) } else { _, e = a.store.db.Exec(`INSERT INTO notes(uuid,content,updated_at,deleted,dirty) VALUES(?,?,?,?,0)`, uuid, asString(m["content"]), remoteUpdated, deleted) } } return e == nil, e } func (a *App) projectNameByID(id int64) string { if id <= 0 { return "" } var name string _ = a.store.db.QueryRow(`SELECT name FROM projects WHERE id=?`, id).Scan(&name) return name } func (a *App) projectIDByName(name string) int64 { if name == "" { return 0 } var id int64 _ = a.store.db.QueryRow(`SELECT id FROM projects WHERE name=?`, name).Scan(&id) return id } func asString(v any) string { switch x := v.(type) { case nil: return "" case string: return x case []byte: return string(x) default: return fmt.Sprint(x) } } func asInt64(v any) int64 { switch x := v.(type) { case int64: return x case float64: return int64(x) case json.Number: n, _ := x.Int64() return n case []byte: n, _ := strconv.ParseInt(string(x), 10, 64) return n case string: n, _ := strconv.ParseInt(x, 10, 64) return n default: return 0 } }