package service import ( "fmt" "strings" "gorm.io/gorm" "nl-pms-api/internal/commonservice" ) // SyncService 个人待办/工单/记事本的推拉(LWW)。 type SyncService struct { DB *gorm.DB } type syncTableMeta struct { remote string cols []string // 不含 user_id;含 uuid … updated_at deleted hasProject bool hasTimes bool } var syncTables = map[string]syncTableMeta{ "todos": { remote: "sync_todos", hasProject: true, hasTimes: true, cols: []string{"uuid", "title", "content", "due_at", "priority", "status", "history", "team_id", "project_name", "created_at", "updated_at", "deleted"}, }, "tickets": { remote: "sync_tickets", hasProject: true, hasTimes: true, cols: []string{"uuid", "title", "description", "type", "start_at", "due_at", "status", "priority", "history", "team_id", "project_name", "created_at", "updated_at", "deleted"}, }, "notes": { remote: "sync_notes", cols: []string{"uuid", "content", "updated_at", "deleted"}, }, } // Push 对指定表做 LWW upsert;row 中不含 user_id(取自 JWT)。 func (s *SyncService) Push(table string, userID int64, rows []map[string]any) (int, error) { meta, ok := syncTables[table] if !ok { return 0, commonservice.BadRequest("SYNC_BAD_TABLE") } if userID <= 0 { return 0, commonservice.Unauthorized("UNAUTHORIZED") } remoteCols := append([]string{"user_id"}, meta.cols...) set := make([]string, 0, len(meta.cols)) for _, c := range meta.cols { if c == "uuid" || 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)") placeholders := strings.TrimRight(strings.Repeat("?,", len(remoteCols)), ",") q := fmt.Sprintf("INSERT INTO %s(%s) VALUES(%s) ON DUPLICATE KEY UPDATE %s", meta.remote, strings.Join(remoteCols, ","), placeholders, strings.Join(set, ",")) n := 0 for _, row := range rows { uuid := commonservice.MapStr(row, "uuid") if uuid == "" { continue } args := make([]any, 0, len(remoteCols)) args = append(args, userID) for _, c := range meta.cols { switch c { case "deleted", "team_id": args = append(args, commonservice.MapInt64(row, c)) default: args = append(args, commonservice.MapStr(row, c)) } } if err := s.DB.Exec(q, args...).Error; err != nil { return n, commonservice.Internal("SYNC_PUSH_FAILED") } n++ } return n, nil } // Pull 按 updated_at 游标增量拉取,最多 2000 条。 func (s *SyncService) Pull(table string, userID int64, cursor string) ([]map[string]any, error) { meta, ok := syncTables[table] if !ok { return nil, commonservice.BadRequest("SYNC_BAD_TABLE") } if userID <= 0 { return nil, commonservice.Unauthorized("UNAUTHORIZED") } q := fmt.Sprintf("SELECT %s FROM %s WHERE user_id=? AND updated_at>? ORDER BY updated_at LIMIT 2000", strings.Join(meta.cols, ","), meta.remote) rawRows, err := s.DB.Raw(q, userID, cursor).Rows() if err != nil { return nil, commonservice.Internal("SYNC_PULL_FAILED") } defer rawRows.Close() out := []map[string]any{} for rawRows.Next() { vals := make([]any, len(meta.cols)) ptrs := make([]any, len(meta.cols)) for i := range vals { ptrs[i] = &vals[i] } if err := rawRows.Scan(ptrs...); err != nil { return nil, commonservice.Internal("SYNC_PULL_FAILED") } m := map[string]any{} for i, c := range meta.cols { m[c] = coerceSQL(vals[i]) } out = append(out, m) } return out, nil } func coerceSQL(v any) any { switch x := v.(type) { case []byte: return string(x) default: return x } }