package main // sync.go 实现登录 + MySQL 同步引擎(本地优先): // 推送本地 dirty 行 → 按 updated_at 增量拉取远端 → Last-Write-Wins 冲突合并。 // 触发时机:登录成功、每 5 分钟巡检(自带断网恢复探测)、设置页手动同步。 import ( "context" "database/sql" "encoding/hex" "encoding/json" "errors" "fmt" "path/filepath" "strconv" "strings" "time" "github.com/go-sql-driver/mysql" "golang.org/x/crypto/bcrypt" ) const syncTimeout = 10 * time.Second // 内置默认连接:本地 MySQL(root/root)的 code_count 库。 // 用户零配置即可登录同步;meta 中存过自定义配置时仍以 meta 为准。 const ( defaultSyncHost = "127.0.0.1" defaultSyncUser = "root" defaultSyncPassword = "root" defaultSyncDatabase = "code_count" ) // ---------- 配置与状态 ---------- func (a *App) syncConfig() SyncConfig { port, _ := strconv.Atoi(a.store.Meta("sync_port")) if port <= 0 || port > 65535 { port = 3306 } c := SyncConfig{ Host: a.store.Meta("sync_host"), Port: port, User: a.store.Meta("sync_user"), Password: a.store.Meta("sync_password"), Database: a.store.Meta("sync_database"), } if c.Host == "" && c.User == "" && c.Database == "" { c.Host, c.User, c.Password, c.Database = defaultSyncHost, defaultSyncUser, defaultSyncPassword, defaultSyncDatabase } return c } func syncConfigComplete(c SyncConfig) bool { return c.Host != "" && c.User != "" && c.Database != "" } func syncDSN(c SyncConfig) string { cfg := mysql.NewConfig() cfg.User = c.User cfg.Passwd = c.Password cfg.Net = "tcp" cfg.Addr = fmt.Sprintf("%s:%d", c.Host, c.Port) cfg.DBName = c.Database cfg.Timeout = 5 * time.Second cfg.ReadTimeout = syncTimeout cfg.WriteTimeout = syncTimeout cfg.Params = map[string]string{"charset": "utf8mb4"} return cfg.FormatDSN() } 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 } if c.Port <= 0 || c.Port > 65535 { c.Port = 3306 } for k, v := range map[string]string{ "sync_host": strings.TrimSpace(c.Host), "sync_port": strconv.Itoa(c.Port), "sync_user": strings.TrimSpace(c.User), "sync_password": c.Password, "sync_database": strings.TrimSpace(c.Database), } { if e := a.store.SetMeta(k, v); e != nil { return e } } return nil } 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, 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 } // ---------- 远端连接 ---------- // openRemote 建立 MySQL 连接。应用与前端不执行任何建表/迁移(DDL), // 数据库与数据表须提前用仓库根目录的 init.sql 初始化。 func (a *App) openRemote(ctx context.Context) (*sql.DB, error) { c := a.syncConfig() if !syncConfigComplete(c) { return nil, errors.New("SYNC_NOT_CONFIGURED") } db, e := sql.Open("mysql", syncDSN(c)) if e != nil { return nil, errors.New("SYNC_OFFLINE") } db.SetMaxOpenConns(2) db.SetConnMaxLifetime(time.Minute) if e = db.PingContext(ctx); e != nil { db.Close() if isSchemaMissing(e) { return nil, errors.New("SYNC_SCHEMA_MISSING") } return nil, errors.New("SYNC_OFFLINE") } return db, nil } // isSchemaMissing 识别"库/表不存在"类错误(1049 未知数据库、1146 表不存在)。 func isSchemaMissing(e error) bool { var me *mysql.MySQLError return errors.As(e, &me) && (me.Number == 1049 || me.Number == 1146) } // mapSyncErr 将底层 MySQL 错误翻译为稳定错误码,未识别的原样返回。 func mapSyncErr(e error) error { if e == nil { return nil } if isSchemaMissing(e) { return errors.New("SYNC_SCHEMA_MISSING") } // 1054 未知列:远端表结构落后(如缺 history 列),提示重跑 init.sql。 var me *mysql.MySQLError if errors.As(e, &me) && me.Number == 1054 { return errors.New("SYNC_SCHEMA_OUTDATED") } return e } // ---------- 注册 / 登录 / 登出 ---------- 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 } ctx, cancel := context.WithTimeout(a.ctx, syncTimeout) defer cancel() db, e := a.openRemote(ctx) if e != nil { return e } defer db.Close() // 先查重给出明确报错;并发注册的极端情况由 username 唯一索引兜底(下方 1062)。 var exists int if e := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM users WHERE username=?`, strings.TrimSpace(username)).Scan(&exists); e == nil && exists > 0 { return errors.New("SYNC_USER_EXISTS") } hash, e := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost) if e != nil { return e } _, e = db.ExecContext(ctx, `INSERT INTO users(username,password_hash,created_at) VALUES(?,?,?)`, strings.TrimSpace(username), string(hash), nowRFC()) var me *mysql.MySQLError if errors.As(e, &me) && me.Number == 1062 { return errors.New("SYNC_USER_EXISTS") } if e == nil { // 新账号默认用应用 logo 当头像(base64);注册后的自动登录同步会推成账号头像。 a.ensureDefaultAvatar() } return mapSyncErr(e) } func (a *App) SyncLogin(username, password string) (SyncStatus, error) { if e := a.ready(); e != nil { return SyncStatus{}, e } username = strings.TrimSpace(username) ctx, cancel := context.WithTimeout(a.ctx, syncTimeout) defer cancel() db, e := a.openRemote(ctx) if e != nil { return SyncStatus{}, e } defer db.Close() var id int64 var hash string e = db.QueryRowContext(ctx, `SELECT id,password_hash FROM users WHERE username=?`, username).Scan(&id, &hash) if e == sql.ErrNoRows || (e == nil && bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) != nil) { return SyncStatus{}, errors.New("SYNC_BAD_CREDENTIALS") } if e != nil { return SyncStatus{}, mapSyncErr(e) } // 切换账号时重置拉取游标,并把本地数据整体标脏,合并进新账号。 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_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(ctx, db, id, 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 Key 密文先用旧密钥解密(失败则回退本地明文重建,再不行就删除该行), // 再用新密钥重加密,与 password_hash 更新放在同一事务里提交。 // 其他已登录设备的旧密钥会失效,需用新密码重新登录。 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") } ctx, cancel := context.WithTimeout(a.ctx, syncTimeout) defer cancel() db, e := a.openRemote(ctx) if e != nil { return e } defer db.Close() var hash string if e = db.QueryRowContext(ctx, `SELECT password_hash FROM users WHERE id=?`, userID).Scan(&hash); e != nil { return mapSyncErr(e) } if bcrypt.CompareHashAndPassword([]byte(hash), []byte(oldPassword)) != nil { return errors.New("SYNC_OLD_PASSWORD_WRONG") } newHash, e := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost) if e != nil { return e } // 盐保持不变,新密码 + 原盐派生新密钥;缺盐(从未用过加密同步)时跳过密钥轮换。 var newKey []byte var salt string se := db.QueryRowContext(ctx, `SELECT value FROM sync_settings WHERE user_id=? AND name='enc_salt'`, userID).Scan(&salt) if se == nil { if raw, de := hex.DecodeString(strings.TrimSpace(salt)); de == nil && len(raw) >= 8 { newKey = deriveEncKey(newPassword, raw) } } else if se != sql.ErrNoRows && !isSchemaMissing(se) { return mapSyncErr(se) } tx, e := db.BeginTx(ctx, nil) if e != nil { return mapSyncErr(e) } defer tx.Rollback() if _, e = tx.ExecContext(ctx, `UPDATE users SET password_hash=? WHERE id=?`, string(newHash), userID); e != nil { return mapSyncErr(e) } if newKey != nil { var blob string be := tx.QueryRowContext(ctx, `SELECT value FROM sync_settings WHERE user_id=? AND name='api_keys'`, userID).Scan(&blob) if be != nil && be != sql.ErrNoRows { return mapSyncErr(be) } if be == nil { 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 重建密文;本地也没有就删除云端行。 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: // 内容未变只换密文:保留原时间戳,LWW 语义完全不受影响。 enc, ee := encryptWithKey(newKey, plain) if ee != nil { return ee } if _, e = tx.ExecContext(ctx, `UPDATE sync_settings SET value=? WHERE user_id=? AND name='api_keys'`, enc, userID); e != nil { return mapSyncErr(e) } case de == nil: // 用本地内容重建:算一次内容更新,打新时间戳。 enc, ee := encryptWithKey(newKey, plain) if ee != nil { return ee } if _, e = tx.ExecContext(ctx, `UPDATE sync_settings SET value=?,updated_at=? WHERE user_id=? AND name='api_keys'`, enc, nowRFC(), userID); e != nil { return mapSyncErr(e) } default: if _, e = tx.ExecContext(ctx, `DELETE FROM sync_settings WHERE user_id=? AND name='api_keys'`, userID); e != nil { return mapSyncErr(e) } } } } if e = tx.Commit(); e != nil { return mapSyncErr(e) } if newKey != nil { _ = 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.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 启动时立即尝试连接 MySQL 同步一次(在线优先),失败自动降级为本地模式; // 之后每 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 { return 0, 0, errors.New("SYNC_NOT_LOGGED_IN") } base := a.ctx if base == nil { base = context.Background() // 测试等无窗口环境下没有应用级 ctx } ctx, cancel := context.WithTimeout(base, 60*time.Second) defer cancel() wasOnline := a.syncOnline.Load() db, e := a.openRemote(ctx) 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 0, 0, e } defer db.Close() a.syncOnline.Store(true) for _, t := range []syncTable{todoSync, ticketSync, noteSync} { p, e := a.pushTable(ctx, db, userID, t) if e != nil { e = mapSyncErr(e) a.setSyncErr(e.Error()) a.emitSyncDone() return pushed, pulled, e } pushed += p g, e := a.pullTable(ctx, db, userID, t) if e != nil { e = mapSyncErr(e) a.setSyncErr(e.Error()) a.emitSyncDone() return pushed, pulled, e } pulled += g } // API Key 加密同步(设置里开启后才执行)。 if st, se := a.store.Settings(); se == nil && st.SyncAPIKeys { kp, kg, ke := a.syncAPIKeys(ctx, db, userID, st) if ke != nil { ke = mapSyncErr(ke) a.setSyncErr(ke.Error()) a.emitSyncDone() return pushed, pulled, ke } pushed += kp pulled += kg } // 头像同步:老服务器缺 sync_settings 表时静默跳过,不影响其余同步。 if ap, ag, ae := a.syncAvatar(ctx, db, userID); ae != nil { if !isSchemaMissing(ae) { ae = mapSyncErr(ae) a.setSyncErr(ae.Error()) a.emitSyncDone() return pushed, pulled, ae } a.store.Log("warning", "同步", "头像同步已跳过", "服务器缺少 sync_settings 表,请重跑 init.sql") } 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(ctx, db, userID, d.name, d.gen, d.apply) if de != nil { if isSchemaMissing(de) { a.store.Log("warning", "同步", "配置同步已跳过", "服务器缺少 sync_settings 表,请重跑 init.sql") break } de = mapSyncErr(de) a.setSyncErr(de.Error()) a.emitSyncDone() return pushed, pulled, de } pushed += dp pulled += dg } // 节日背景图同步:管理员(id=1)推送,全员拉取;缺表静默跳过。 if fp, fg, fe := a.syncFestivalImages(ctx, db); fe != nil { if !isSchemaMissing(fe) { fe = mapSyncErr(fe) a.setSyncErr(fe.Error()) a.emitSyncDone() return pushed, pulled, fe } a.store.Log("warning", "同步", "节日图片同步已跳过", "服务器缺少 sync_settings 表,请重跑 init.sql") } else { pushed += fp pulled += fg } // 全局文件存储配置:管理员(id=1)推送,全员拉取;缺表静默跳过。 if sp, sg, se := a.syncFileStorageConfig(ctx, db); se != nil { if !isSchemaMissing(se) { se = mapSyncErr(se) a.setSyncErr(se.Error()) a.emitSyncDone() return pushed, pulled, se } a.store.Log("warning", "同步", "文件存储配置同步已跳过", "服务器缺少 sync_settings 表,请重跑 init.sql") } else { pushed += sp pulled += sg } // 公开资料推拉 + 团队通知拉取 + 团队指派数缓存:服务器未升级团队表时静默跳过。 a.syncUserProfile(ctx, db, userID) pulled += a.pullTeamNotices(ctx, db, userID) a.refreshTeamAssignedCache(ctx, db, userID) _ = 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(ctx context.Context, db *sql.DB, userID int64, password string) error { var salt string e := db.QueryRowContext(ctx, `SELECT value FROM sync_settings WHERE user_id=? AND name='enc_salt'`, userID).Scan(&salt) if e == sql.ErrNoRows { if salt, e = newSaltHex(); e != nil { return e } if _, e = db.ExecContext(ctx, `INSERT IGNORE INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?)`, userID, "enc_salt", salt, nowRFC()); e != nil { return e } // 并发写入时以真正落库的盐为准。 if e = db.QueryRowContext(ctx, `SELECT value FROM sync_settings WHERE user_id=? AND name='enc_salt'`, userID).Scan(&salt); e != nil { return e } } else if e != nil { return e } 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(ctx context.Context, db *sql.DB, userID int64, 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") // 开启功能前就配置过 Key 的老用户:补一个本地时间戳,让首轮同步能推送。 if localAt == "" && (st.SparkKey != "" || st.DeepSeekKey != "") { localAt = nowRFC() _ = a.store.SetMeta("api_keys_updated_at", localAt) } var remoteVal, remoteAt string e := db.QueryRowContext(ctx, `SELECT value,updated_at FROM sync_settings WHERE user_id=? AND name='api_keys'`, userID).Scan(&remoteVal, &remoteAt) if e != nil && e != sql.ErrNoRows { return 0, 0, e } hasRemote := e == nil 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 = db.ExecContext(ctx, `INSERT INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE value=IF(VALUES(updated_at)>updated_at,VALUES(value),value), updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)`, userID, "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 } } // 覆盖 SaveSettings 里自动打的时间戳,与远端保持一致,避免下一轮回推。 _ = a.store.SetMeta("api_keys_updated_at", remoteAt) pulled = 1 } return pushed, pulled, nil } // syncAvatar 同步头像(sync_settings 的 name='avatar' 行,明文 JSON {mode,value},LWW)。 // path 模式是设备本地行为:既不推送,也不参与拉取覆盖。 func (a *App) syncAvatar(ctx context.Context, db *sql.DB, userID int64) (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) } var remoteVal, remoteAt string e = db.QueryRowContext(ctx, `SELECT value,updated_at FROM sync_settings WHERE user_id=? AND name='avatar'`, userID).Scan(&remoteVal, &remoteAt) if e != nil && e != sql.ErrNoRows { return 0, 0, e } hasRemote := e == nil 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 = db.ExecContext(ctx, `INSERT INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE value=IF(VALUES(updated_at)>updated_at,VALUES(value),value), updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)`, userID, "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 } // ---------- 项目/分组/收藏/排除规则文档同步 ---------- // // 复用 sync_settings 表的两个文档行(name='projects' / name='rules'),整文档 LWW: // 本地文档与远端 value 不一致才推送(省去在十几个写入口埋点); // 远端更新则先套用(项目只增改不删,规则替换自定义部分)再把合并结果回推。 // projectDocEntry 是项目身份文档的元素。v2 起 path 不再入云(跨机器路径互不相同), // 保留字段用于解析旧客户端推送的 v1 文档(含 path)。 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"` } // projectsDoc 生成本地项目身份文档(v2:不含本机路径,按 name 排序保证内容稳定可比较)。 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 } // ensureGroupID 按名取分组 id,不存在则创建;空名归入默认分组(id=1)。 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 } // applyProjectsDoc 套用远端项目身份文档:按 name 匹配本地项目并更新描述/分组/收藏; // 本机没有的项目不再落库(路径属于机器私有,由 paths: 行或用户手动绑定补齐), // 而是记入待绑定清单;远端缺失的本地项目也不删除(避免误删统计数据)。 // 兼容 v1 旧文档:元素含 path 时用它做二次匹配(项目改名场景),但不再采用其路径。 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 // 本机没有该项目:进待绑定清单(见下方 rebuildCloudPending) 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 } // rulesDoc 生成本地自定义排除规则文档(内置规则不参与同步)。 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 } // applyRulesDoc 用远端文档整体替换本地自定义规则(内置规则保持不动)。 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" } // 与内置规则撞名时忽略,pattern 全局唯一。 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(ctx context.Context, db *sql.DB, userID int64, 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 } var remoteVal, remoteAt string e = db.QueryRowContext(ctx, `SELECT value,updated_at FROM sync_settings WHERE user_id=? AND name=?`, userID, name).Scan(&remoteVal, &remoteAt) if e != nil && e != sql.ErrNoRows { return 0, 0, e } hasRemote := e == nil 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 = db.ExecContext(ctx, `INSERT INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE value=IF(VALUES(updated_at)>updated_at,VALUES(value),value), updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)`, userID, 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:<节日名>,统一挂在管理员(id=1)名下。 // 管理员账号推送本地 dirty 行;所有账号都从 id=1 拉取,实现"管理员配置、全员可见"。 func (a *App) syncFestivalImages(ctx context.Context, db *sql.DB) (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 := db.ExecContext(ctx, `INSERT INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE value=IF(VALUES(updated_at)>updated_at,VALUES(value),value), updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)`, festivalAdminID, prefix+r.k, r.v, r.at); e != nil { return pushed, pulled, e } // 推送期间若管理员又改了同一节日(updated_at 变化),保留 dirty 待下轮再推。 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++ } } cursor := a.store.Meta("sync_pull_fest_imgs") rows, e := db.QueryContext(ctx, `SELECT name,value,updated_at FROM sync_settings WHERE user_id=? AND name LIKE 'fest\_img:%' AND updated_at>? ORDER BY updated_at LIMIT 500`, festivalAdminID, cursor) if e != nil { return pushed, pulled, e } defer rows.Close() maxSeen := cursor for rows.Next() { var name, val, at string if e := rows.Scan(&name, &val, &at); e != nil { return pushed, pulled, e } if at > maxSeen { maxSeen = at } key := strings.TrimPrefix(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 // 本地相同或更新(管理员本机 dirty 行下轮推送) } 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, val, at); e != nil { return pushed, pulled, e } pulled++ } if e := rows.Err(); e != nil { return pushed, pulled, e } if maxSeen != cursor { _ = a.store.SetMeta("sync_pull_fest_imgs", maxSeen) } return pushed, pulled, nil } // syncFileStorageConfig 同步全局文件存储配置(sync_settings 的 name='file_storage' 行, // 统一挂在管理员 id=1 名下):管理员按 LWW 推送本地修改,所有账号拉取, // 实现"管理员配置存储方式、全员生效"(见 filestorage.go)。 func (a *App) syncFileStorageConfig(ctx context.Context, db *sql.DB) (pushed, pulled int, err error) { localAt := a.store.Meta(fileStorageAtKey) var remoteVal, remoteAt string e := db.QueryRowContext(ctx, `SELECT value,updated_at FROM sync_settings WHERE user_id=? AND name=?`, festivalAdminID, fileStorageKey).Scan(&remoteVal, &remoteAt) if e != nil && e != sql.ErrNoRows { return 0, 0, e } hasRemote := e == nil switch { case a.syncUserID() == festivalAdminID && localAt != "" && (!hasRemote || localAt > remoteAt): b, e := json.Marshal(a.fileStorageConfig()) if e != nil { return 0, 0, e } if _, e = db.ExecContext(ctx, `INSERT INTO sync_settings(user_id,name,value,updated_at) VALUES(?,?,?,?) ON DUPLICATE KEY UPDATE value=IF(VALUES(updated_at)>updated_at,VALUES(value),value), updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)`, festivalAdminID, fileStorageKey, string(b), localAt); e != 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, remote string // localCols 顺序即 scan/insert 顺序;project_id 与 project_name 的转换单独处理。 hasProject bool hasTimes bool // 有 created_at(notes 没有) extraCols []string } var ( todoSync = syncTable{local: "todos", remote: "sync_todos", hasProject: true, hasTimes: true, extraCols: []string{"title", "content", "due_at", "priority", "status", "history", "team_id"}} ticketSync = syncTable{local: "tickets", remote: "sync_tickets", hasProject: true, hasTimes: true, extraCols: []string{"title", "description", "type", "start_at", "due_at", "status", "priority", "history", "team_id"}} noteSync = syncTable{local: "notes", remote: "sync_notes", extraCols: []string{"content"}} ) // pushTable 上传本地 dirty 行(LWW:仅当本地 updated_at 更新时覆盖远端)。 func (a *App) pushTable(ctx context.Context, db *sql.DB, userID int64, 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 } // 远端列:uuid + extra + project_name? + created_at? + updated_at + deleted remoteCols := append([]string{"user_id", "uuid"}, t.extraCols...) if t.hasProject { remoteCols = append(remoteCols, "project_name") } if t.hasTimes { remoteCols = append(remoteCols, "created_at") } remoteCols = append(remoteCols, "updated_at", "deleted") // LWW 覆盖:所有列都用 IF(VALUES(updated_at)>updated_at,...),updated_at 必须最后赋值。 set := []string{} for _, c := range remoteCols[2:] { if c == "updated_at" { continue } set = append(set, fmt.Sprintf("%s=IF(VALUES(updated_at)>updated_at,VALUES(%s),%s)", c, c, c)) } set = append(set, "updated_at=IF(VALUES(updated_at)>updated_at,VALUES(updated_at),updated_at)") q := fmt.Sprintf("INSERT INTO %s(%s) VALUES(%s) ON DUPLICATE KEY UPDATE %s", t.remote, strings.Join(remoteCols, ","), strings.TrimRight(strings.Repeat("?,", len(remoteCols)), ","), strings.Join(set, ",")) n := 0 for _, r := range recs { args := []any{userID} for i, c := range cols { if c == "project_id" { args = append(args, a.projectNameByID(asInt64(r.vals[i]))) continue } args = append(args, r.vals[i]) } if _, e := db.ExecContext(ctx, q, args...); e != nil { return n, e } if _, e := a.store.db.Exec(`UPDATE `+t.local+` SET dirty=0 WHERE uuid=? AND dirty=1`, r.uuid); e != nil { return n, e } n++ } return n, nil } // pullTable 增量拉取远端更新(LWW:远端 updated_at 更新才覆盖本地)。 func (a *App) pullTable(ctx context.Context, db *sql.DB, userID int64, t syncTable) (int, error) { cursorKey := "sync_pull_" + t.local cursor := a.store.Meta(cursorKey) remoteCols := append([]string{"uuid"}, t.extraCols...) if t.hasProject { remoteCols = append(remoteCols, "project_name") } if t.hasTimes { remoteCols = append(remoteCols, "created_at") } remoteCols = append(remoteCols, "updated_at", "deleted") rows, e := db.QueryContext(ctx, `SELECT `+strings.Join(remoteCols, ",")+` FROM `+t.remote+` WHERE user_id=? AND updated_at>? ORDER BY updated_at LIMIT 2000`, userID, cursor) if e != nil { return 0, e } defer rows.Close() n, maxSeen := 0, cursor for rows.Next() { vals := make([]any, len(remoteCols)) ptrs := make([]any, len(remoteCols)) for i := range vals { ptrs[i] = &vals[i] } if e := rows.Scan(ptrs...); e != nil { return n, e } m := map[string]any{} for i, c := range remoteCols { m[c] = vals[i] } 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 e := rows.Err(); e != nil { return n, e } 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 // 本地相同或更新(dirty 行下轮推送) } 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": // 多条笔记:按 uuid 标准 LWW upsert(历史单条时代不同设备 uuid 不同, // 拉取后各成一条笔记,内容互不覆盖)。 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 } // projectNameByID / projectIDByName 用项目名在设备间映射项目绑定。 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 []byte: n, _ := strconv.ParseInt(string(x), 10, 64) return n case string: n, _ := strconv.ParseInt(x, 10, 64) return n default: return 0 } }