功能更新

internal/service/sync.go

@@ -0,0 +1,132 @@
+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
+	}
+}