MyGoNavi/internal/resultdiff/session.go

package resultdiff

import (
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
)

// Session 保存一次比对的中间数据与结果。
type Session struct {
	ID              string
	CreatedAt       time.Time
	KeyColumns      []string
	CompareColumns  []string
	IgnoreColumns   []string
	Options         CompareOptions
	MaxRowsPerSide  int
	IncludeSameRows bool

	mu sync.RWMutex

	// 装载缓冲（rows 模式）
	leftColumns  []string
	rightColumns []string
	leftRows     []map[string]interface{}
	rightRows    []map[string]interface{}
	leftDone     bool
	rightDone    bool

	// 计算结果
	computed bool
	summary  Summary
	rows     []DiffRow
	err      error
}

// Manager 管理内存中的 diff 会话。
type Manager struct {
	mu       sync.Mutex
	sessions map[string]*Session
	ttl      time.Duration
}

// NewManager 创建会话管理器。
func NewManager(ttl time.Duration) *Manager {
	if ttl <= 0 {
		ttl = 30 * time.Minute
	}
	m := &Manager{
		sessions: make(map[string]*Session),
		ttl:      ttl,
	}
	return m
}

// Create 新建会话。
func (m *Manager) Create(req StartRequest) *Session {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.gcLocked()

	id := strings.TrimSpace(req.JobID)
	if id == "" {
		id = "rdiff-" + uuid.NewString()
	}
	maxRows := req.MaxRowsPerSide
	if maxRows <= 0 {
		maxRows = DefaultMaxRowsPerSide
	}
	s := &Session{
		ID:              id,
		CreatedAt:       time.Now(),
		KeyColumns:      normalizeColumnList(req.KeyColumns),
		CompareColumns:  normalizeColumnList(req.CompareColumns),
		IgnoreColumns:   normalizeColumnList(req.IgnoreColumns),
		Options:         req.Options,
		MaxRowsPerSide:  maxRows,
		IncludeSameRows: req.IncludeSameRows,
		leftRows:        make([]map[string]interface{}, 0),
		rightRows:       make([]map[string]interface{}, 0),
	}
	m.sessions[id] = s
	return s
}

// Get 获取会话。
func (m *Manager) Get(jobID string) (*Session, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.gcLocked()
	s, ok := m.sessions[strings.TrimSpace(jobID)]
	if !ok {
		return nil, fmt.Errorf("result diff job not found: %s", jobID)
	}
	return s, nil
}

// Close 释放会话。
func (m *Manager) Close(jobID string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	delete(m.sessions, strings.TrimSpace(jobID))
}

func (m *Manager) gcLocked() {
	if m.ttl <= 0 {
		return
	}
	now := time.Now()
	for id, s := range m.sessions {
		if now.Sub(s.CreatedAt) > m.ttl {
			delete(m.sessions, id)
		}
	}
}

// AppendRows 向一侧追加行。
func (s *Session) AppendRows(side string, columns []string, rows []map[string]interface{}, done bool) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.computed {
		return fmt.Errorf("job already computed")
	}
	side = strings.TrimSpace(strings.ToLower(side))
	switch side {
	case "left":
		if len(columns) > 0 && len(s.leftColumns) == 0 {
			s.leftColumns = normalizeColumnList(columns)
		}
		if err := s.appendSideLocked(&s.leftRows, rows); err != nil {
			return err
		}
		if done {
			s.leftDone = true
		}
	case "right":
		if len(columns) > 0 && len(s.rightColumns) == 0 {
			s.rightColumns = normalizeColumnList(columns)
		}
		if err := s.appendSideLocked(&s.rightRows, rows); err != nil {
			return err
		}
		if done {
			s.rightDone = true
		}
	default:
		return fmt.Errorf("invalid side: %s", side)
	}
	return nil
}

func (s *Session) appendSideLocked(dest *[]map[string]interface{}, rows []map[string]interface{}) error {
	if len(rows) == 0 {
		return nil
	}
	next := len(*dest) + len(rows)
	if next > s.MaxRowsPerSide {
		return fmt.Errorf("row count exceeds maxRowsPerSide=%d (would be %d)", s.MaxRowsPerSide, next)
	}
	*dest = append(*dest, rows...)
	return nil
}

// SetLoaded 直接设置两侧已装载数据（SQL 模式）。
func (s *Session) SetLoaded(
	leftCols []string, leftRows []map[string]interface{},
	rightCols []string, rightRows []map[string]interface{},
) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(leftRows) > s.MaxRowsPerSide {
		return fmt.Errorf("left row count %d exceeds maxRowsPerSide=%d", len(leftRows), s.MaxRowsPerSide)
	}
	if len(rightRows) > s.MaxRowsPerSide {
		return fmt.Errorf("right row count %d exceeds maxRowsPerSide=%d", len(rightRows), s.MaxRowsPerSide)
	}
	s.leftColumns = normalizeColumnList(leftCols)
	s.rightColumns = normalizeColumnList(rightCols)
	s.leftRows = leftRows
	s.rightRows = rightRows
	s.leftDone = true
	s.rightDone = true
	return nil
}

// Compute 执行 diff。
func (s *Session) Compute() (Summary, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.computed {
		return s.summary, s.err
	}
	if !s.leftDone || !s.rightDone {
		return Summary{}, fmt.Errorf("both sides must be fully loaded before compute")
	}
	summary, rows, err := ComputeDiff(
		s.leftRows, s.leftColumns,
		s.rightRows, s.rightColumns,
		s.KeyColumns, s.CompareColumns, s.IgnoreColumns,
		s.Options, s.IncludeSameRows,
	)
	s.summary = summary
	s.rows = rows
	s.err = err
	s.computed = err == nil
	// 释放原始行以降低内存（结果 rows 仍保留 left/right 引用）
	// 注意：DiffRow 引用了原 map，不能清空 leftRows/rightRows 内容
	return summary, err
}

// Page 分页。
func (s *Session) Page(req PageRequest) (PageResult, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if !s.computed {
		return PageResult{}, fmt.Errorf("job not computed yet")
	}
	includeSame := req.IncludeSameRows || s.IncludeSameRows
	page := FilterPage(s.rows, req.Kinds, req.ChangedColumn, req.Offset, req.Limit, includeSame)
	page.JobID = s.ID
	return page, nil
}

// Summary 返回汇总。
func (s *Session) Summary() (Summary, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.summary, s.computed
}