nl-pms-api/internal/service/sync.go

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
	}
}