github-auto-sync-golang/engine.go

package main

import (
	"encoding/csv"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"regexp"
	"runtime/debug"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// ---------- 事件总线（SSE 广播） ----------

type EventBus struct {
	mu   sync.Mutex
	next int
	subs map[int]chan []byte
}

func NewEventBus() *EventBus { return &EventBus{subs: map[int]chan []byte{}} }

func (b *EventBus) Subscribe() (int, chan []byte) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.next++
	ch := make(chan []byte, 128)
	b.subs[b.next] = ch
	return b.next, ch
}

func (b *EventBus) Unsubscribe(id int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if ch, ok := b.subs[id]; ok {
		delete(b.subs, id)
		close(ch)
	}
}

func (b *EventBus) Publish(data []byte) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, ch := range b.subs {
		select {
		case ch <- data:
		default: // 订阅者阻塞时丢弃，避免拖慢同步循环
		}
	}
}

// ---------- 状态 ----------

type SyncResult struct {
	Time   string `json:"time"`
	Result string `json:"result"` // ok | error | conflict
	Detail string `json:"detail"`
}

type RepoSnapshot struct {
	Name             string      `json:"name"`
	RepoPath         string      `json:"repo_path"`
	Branch           string      `json:"branch"`
	Mode             string      `json:"mode"`
	Interval         int         `json:"interval"`
	AuthMethod       string      `json:"auth_method"`
	RetryCount       int         `json:"retry_count"`
	RetryInterval    int         `json:"retry_interval"`
	ConflictStrategy string      `json:"conflict_strategy"`
	LogFile          string      `json:"log_file"`
	AutoCommit       bool        `json:"auto_commit"`
	Cron             string      `json:"cron"`
	Remotes          []Remote    `json:"remotes"`
	Running          bool        `json:"running"`
	Status           string      `json:"status"`
	Ahead            int         `json:"ahead"`
	Behind           int         `json:"behind"`
	Dirty            int         `json:"dirty"`
	Conflicts        []string    `json:"conflicts"`
	LastError        string      `json:"last_error"`
	LastSync         *SyncResult `json:"last_sync"`
}

type repoState struct {
	mu        sync.Mutex
	cfg       *RepoConfig
	running   bool
	syncing   bool
	status    string // idle | ok | error | conflict
	ahead     int
	behind    int
	dirty     int
	conflicts []string
	lastError string
	lastSync  *SyncResult
	stopCh    chan struct{}
}

type Engine struct {
	baseDir  string
	cfgDir   string
	logDir   string
	statsDir string

	mu     sync.Mutex
	states map[string]*repoState
	bus    *EventBus
}

func NewEngine(base string) (*Engine, error) {
	e := &Engine{
		baseDir:  base,
		cfgDir:   filepath.Join(base, "configs"),
		logDir:   filepath.Join(base, "logs"),
		statsDir: filepath.Join(base, "stats"),
		states:   map[string]*repoState{},
		bus:      NewEventBus(),
	}
	for _, d := range []string{e.cfgDir, e.logDir, e.statsDir} {
		if err := os.MkdirAll(d, 0o755); err != nil {
			return nil, err
		}
	}
	return e, nil
}

func (e *Engine) Bus() *EventBus { return e.bus }

func (e *Engine) configNames() []string {
	entries, err := os.ReadDir(e.cfgDir)
	if err != nil {
		return nil
	}
	var names []string
	for _, en := range entries {
		if !en.IsDir() && strings.HasSuffix(en.Name(), ".conf") {
			names = append(names, strings.TrimSuffix(en.Name(), ".conf"))
		}
	}
	sort.Strings(names)
	return names
}

func (e *Engine) get(name string) *repoState {
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.states[name]
}

func (e *Engine) sortedNames() []string {
	e.mu.Lock()
	defer e.mu.Unlock()
	names := make([]string, 0, len(e.states))
	for n := range e.states {
		names = append(names, n)
	}
	sort.Strings(names)
	return names
}

// FIX: LoadAll 加锁，避免并发读写 map
func (e *Engine) LoadAll() {
	e.mu.Lock()
	defer e.mu.Unlock()
	for _, name := range e.configNames() {
		if _, ok := e.states[name]; ok {
			continue
		}
		cfg, err := loadRepoConfig(e.cfgDir, name)
		if err != nil {
			log.Printf("加载配置 %s 失败: %v", name, err)
			continue
		}
		e.states[name] = &repoState{cfg: cfg, status: "idle"}
	}
}

func (e *Engine) publish(v any) {
	data, err := json.Marshal(v)
	if err != nil {
		return
	}
	e.bus.Publish(data)
}

func (e *Engine) logf(s *repoState, level, format string, args ...any) {
	msg := fmt.Sprintf(format, args...)
	ts := time.Now().Format("2006-01-02 15:04:05")
	logPath := s.cfg.LogFile
	if logPath == "" {
		logPath = filepath.Join(e.logDir, s.cfg.Name+".log")
	}
	if logPath != "" {
		_ = os.MkdirAll(filepath.Dir(logPath), 0o755)
		if f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644); err == nil {
			fmt.Fprintln(f, "["+ts+"] "+msg)
			_ = f.Close()
		}
	}
	e.publish(map[string]any{"type": "log", "repo": s.cfg.Name, "level": level, "msg": msg, "time": ts})
}

func (e *Engine) stat(s *repoState, action, result string, commits int) {
	path := filepath.Join(e.statsDir, s.cfg.Name+".csv")
	if _, err := os.Stat(path); os.IsNotExist(err) {
		_ = os.WriteFile(path, []byte("timestamp,action,result,commits\n"), 0o644)
	}
	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
	if err != nil {
		return
	}
	defer f.Close()
	fmt.Fprintf(f, "%s,%s,%s,%d\n", time.Now().Format("2006-01-02 15:04:05"), action, result, commits)
}

func firstRemoteName(cfg *RepoConfig) string {
	if len(cfg.Remotes) > 0 {
		return cfg.Remotes[0].Name
	}
	return ""
}

func (s *repoState) snapshot(live bool) RepoSnapshot {
	s.mu.Lock()
	cfg := s.cfg
	snap := RepoSnapshot{
		Name:             cfg.Name,
		RepoPath:         cfg.RepoPath,
		Branch:           cfg.Branch,
		Mode:             cfg.Mode,
		Interval:         cfg.Interval,
		AuthMethod:       cfg.AuthMethod,
		RetryCount:       cfg.RetryCount,
		RetryInterval:    cfg.RetryInterval,
		ConflictStrategy: cfg.ConflictStrategy,
		LogFile:          cfg.LogFile,
		AutoCommit:       cfg.AutoCommit,
		Cron:             cfg.Cron,
		Remotes:          cfg.Remotes,
		Running:          s.running,
		Status:           s.status,
		Ahead:            s.ahead,
		Behind:           s.behind,
		Dirty:            s.dirty,
		Conflicts:        append([]string(nil), s.conflicts...),
		LastError:        s.lastError,
		LastSync:         s.lastSync,
	}
	s.mu.Unlock()
	if live {
		if b, err := currentBranch(cfg.RepoPath); err == nil {
			snap.Branch = b
		}
		snap.Dirty = dirtyCount(cfg.RepoPath)
		if remote := firstRemoteName(cfg); remote != "" {
			if a, be, err := aheadBehind(cfg.RepoPath, remote, snap.Branch); err == nil {
				snap.Ahead, snap.Behind = a, be
			}
		}
		if files, err := conflictFiles(cfg.RepoPath); err == nil && len(files) > 0 {
			snap.Conflicts = files
			snap.Status = "conflict"
		}
	}
	return snap
}

func (e *Engine) Snapshots(live bool) []RepoSnapshot {
	var out []RepoSnapshot
	for _, name := range e.sortedNames() {
		if s := e.get(name); s != nil {
			out = append(out, s.snapshot(live))
		}
	}
	return out
}

func (e *Engine) pushStatus(s *repoState) {
	e.publish(map[string]any{"type": "status", "repo": s.cfg.Name, "data": s.snapshot(false)})
}

// ---------- 启动 / 停止 ----------

func (e *Engine) Start(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	s.mu.Lock()
	if s.running {
		s.mu.Unlock()
		return nil
	}
	if !isRepo(s.cfg.RepoPath) {
		s.mu.Unlock()
		return fmt.Errorf("仓库路径无效: %s", s.cfg.RepoPath)
	}
	s.stopCh = make(chan struct{})
	s.running = true
	s.status = "ok"
	ch := s.stopCh
	s.mu.Unlock()
	go e.loop(s, ch)
	e.logf(s, "info", "监控已启动")
	e.pushStatus(s)
	return nil
}

func (e *Engine) loop(s *repoState, stopCh chan struct{}) {
	s.mu.Lock()
	interval := s.cfg.Interval
	cronStr := s.cfg.Cron
	s.mu.Unlock()
	if interval < 5 {
		interval = 5
	}

	var spec *cronSpec
	if cronStr != "" {
		sp, err := parseCron(cronStr)
		if err != nil {
			e.logf(s, "error", "cron 表达式无效: %v（回退为固定间隔 %d 秒）", err, interval)
		} else {
			spec = sp
			e.logf(s, "info", "按 cron 计划运行: %s", cronStr)
		}
	}
	if spec == nil {
		e.syncOnce(s, "启动检查")
	}
	for {
		var wait time.Duration
		trigger := "定时检查"
		if spec != nil {
			wait = time.Until(cronNext(spec, time.Now()))
			trigger = "cron 触发"
		} else {
			wait = time.Duration(interval) * time.Second
		}
		if wait < time.Second {
			wait = time.Second
		}
		timer := time.NewTimer(wait)
		select {
		case <-stopCh:
			timer.Stop()
			e.logf(s, "info", "监控已停止")
			e.pushStatus(s)
			return
		case <-timer.C:
			e.syncOnce(s, trigger)
		}
	}
}

func (e *Engine) Stop(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.running {
		return nil
	}
	s.running = false
	s.status = "idle"
	close(s.stopCh)
	return nil
}

func (e *Engine) StartAll() {
	for _, name := range e.sortedNames() {
		if err := e.Start(name); err != nil {
			log.Printf("自动启动 %s 失败: %v", name, err)
		}
	}
}

func (e *Engine) SyncNow(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if !s.running {
		return fmt.Errorf("监控未启动，请先启动")
	}
	if s.syncing {
		return fmt.Errorf("正在同步中，请稍候")
	}
	go e.safeSync(s, "手动触发")
	return nil
}

// ---------- 同步核心 ----------

func (e *Engine) withRetry(s *repoState, label string, fn func() (string, error)) (string, error) {
	maxAttempts, sleep, retryCount := 1, 0, 0
	s.mu.Lock()
	cfg := s.cfg
	s.mu.Unlock()
	if cfg.AuthMethod == "https" && cfg.RetryCount > 0 {
		maxAttempts = cfg.RetryCount + 1
		sleep = cfg.RetryInterval
		retryCount = cfg.RetryCount
	}
	var lastErr error
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		if attempt > 1 {
			e.logf(s, "warn", "%s 失败，%d 秒后重试（第 %d/%d 次）", label, sleep, attempt-1, retryCount)
			time.Sleep(time.Duration(sleep) * time.Second)
		}
		out, err := fn()
		if err == nil {
			return out, nil
		}
		lastErr = err
	}
	return "", lastErr
}

func (e *Engine) setFail(s *repoState, msg string) {
	now := time.Now().Format("2006-01-02 15:04:05")
	s.mu.Lock()
	s.status = "error"
	s.lastError = msg
	s.lastSync = &SyncResult{Time: now, Result: "error", Detail: msg}
	s.mu.Unlock()
	e.logf(s, "error", "%s", msg)
	e.pushStatus(s)
}

func (e *Engine) setOK(s *repoState, detail string) {
	now := time.Now().Format("2006-01-02 15:04:05")
	s.mu.Lock()
	s.status = "ok"
	s.lastError = ""
	s.lastSync = &SyncResult{Time: now, Result: "ok", Detail: detail}
	s.mu.Unlock()
	e.pushStatus(s)
}

// ensureRemotes 把配置中的远程同步到仓库（缺失则添加，URL 变化则更新）
func (e *Engine) ensureRemotes(s *repoState, cfg *RepoConfig) {
	for _, r := range cfg.Remotes {
		out, err := gitRun(cfg.RepoPath, 15*time.Second, "remote", "get-url", r.Name)
		if err != nil {
			if _, aerr := gitRun(cfg.RepoPath, 15*time.Second, "remote", "add", r.Name, r.URL); aerr != nil {
				e.logf(s, "warn", "添加远程 %s 失败: %v", r.Name, aerr)
			} else {
				e.logf(s, "info", "已添加远程 %s → %s", r.Name, r.URL)
			}
		} else if strings.TrimSpace(out) != r.URL {
			if _, uerr := gitRun(cfg.RepoPath, 15*time.Second, "remote", "set-url", r.Name, r.URL); uerr != nil {
				e.logf(s, "warn", "更新远程 %s URL 失败: %v", r.Name, uerr)
			} else {
				e.logf(s, "info", "已更新远程 %s URL → %s", r.Name, r.URL)
			}
		}
	}
}

func (e *Engine) refreshCounts(s *repoState, cfg *RepoConfig, remote string) {
	ahead, behind, err := aheadBehind(cfg.RepoPath, remote, cfg.Branch)
	dirty := dirtyCount(cfg.RepoPath)
	s.mu.Lock()
	if err == nil {
		s.ahead, s.behind = ahead, behind
	}
	s.dirty = dirty
	s.mu.Unlock()
	e.pushStatus(s)
}

// safeSync 包装 syncOnce：捕获后台 goroutine 的 panic，避免整个进程被杀死。
func (e *Engine) safeSync(s *repoState, trigger string) {
	defer func() {
		if r := recover(); r != nil {
			stack := string(debug.Stack())
			if len(stack) > 4096 {
				stack = stack[:4096] + "\n…（堆栈已截断）"
			}
			log.Printf("[panic] syncOnce(%s): %v\n%s", s.cfg.Name, r, stack)
			e.logf(s, "error", "同步发生内部异常（已恢复，进程继续运行）: %v", r)
		}
	}()
	e.syncOnce(s, trigger)
}

// FIX: 调整同步顺序：先拉取后推送；分支切换避免强制覆盖；pull 使用重试
func (e *Engine) syncOnce(s *repoState, trigger string) {
	s.mu.Lock()
	if s.syncing {
		s.mu.Unlock()
		return
	}
	s.syncing = true
	cfg := s.cfg
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		s.syncing = false
		s.mu.Unlock()
	}()

	if !isRepo(cfg.RepoPath) {
		e.setFail(s, "仓库路径无效: "+cfg.RepoPath)
		return
	}
	e.ensureRemotes(s, cfg)
	remote := firstRemoteName(cfg)
	if remote == "" {
		e.setFail(s, "未配置远程仓库")
		return
	}

	e.logf(s, "info", "=== 开始同步（%s，模式 %s）===", trigger, cfg.Mode)

	if _, err := e.withRetry(s, "fetch", func() (string, error) {
		return gitRun(cfg.RepoPath, 90*time.Second, "fetch", remote)
	}); err != nil {
		e.setFail(s, "fetch 失败: "+err.Error())
		return
	}

	// FIX: 分支对齐逻辑 – 避免使用 -B 强制覆盖，本地存在则普通切换，不存在则创建
	if cur, cerr := currentBranch(cfg.RepoPath); cerr == nil && cur != cfg.Branch {
		tip, remoteHasBranch := remoteTip(cfg.RepoPath, remote, cfg.Branch)
		localExists := branchExists(cfg.RepoPath, cfg.Branch)
		if localExists {
			// 本地分支存在，普通切换（可能因未提交改动而失败）
			if _, err := gitRun(cfg.RepoPath, 30*time.Second, "checkout", cfg.Branch); err != nil {
				e.setFail(s, fmt.Sprintf("切换到配置分支 %s 失败，请先提交或暂存改动: %v", cfg.Branch, err))
				return
			}
			e.logf(s, "info", "本地分支与配置不一致（%s → %s），已自动切换", cur, cfg.Branch)
		} else {
			// 本地分支不存在，创建（不会覆盖任何内容）
			if remoteHasBranch {
				if _, err := gitRun(cfg.RepoPath, 30*time.Second, "checkout", "-b", cfg.Branch, tip); err != nil {
					e.setFail(s, fmt.Sprintf("基于远程 %s/%s 创建分支失败: %v", remote, cfg.Branch, err))
					return
				}
				_, _ = gitRun(cfg.RepoPath, 15*time.Second, "branch", "--set-upstream-to", remote+"/"+cfg.Branch, cfg.Branch)
				e.logf(s, "info", "本地无分支 %s，已基于远程 %s（%s）创建并跟踪", cfg.Branch, cfg.Branch, tip[:8])
			} else {
				if _, err := gitRun(cfg.RepoPath, 30*time.Second, "checkout", "-b", cfg.Branch); err != nil {
					e.setFail(s, fmt.Sprintf("创建分支 %s 失败: %v", cfg.Branch, err))
					return
				}
				e.logf(s, "info", "本地与远程均无分支 %s，已基于当前 HEAD（%s）创建", cfg.Branch, cur)
			}
		}
	}

	// 自动提交（若有未提交改动）
	dirty := dirtyCount(cfg.RepoPath)
	if dirty > 0 {
		if cfg.AutoCommit {
			e.logf(s, "info", "检测到 %d 个未提交改动，执行自动提交", dirty)
			if _, cerr := gitRun(cfg.RepoPath, 60*time.Second, "add", "-A"); cerr != nil {
				e.setFail(s, "自动提交（git add）失败: "+cerr.Error())
				return
			}
			cmsg := "git-sync-web 自动提交 " + time.Now().Format("2006-01-02 15:04:05")
			if _, cerr := gitRun(cfg.RepoPath, 120*time.Second, "commit", "-m", cmsg); cerr != nil {
				e.setFail(s, "自动提交失败: "+cerr.Error())
				return
			}
			e.logf(s, "info", "自动提交完成，将随本轮同步推送")
			dirty = 0
		} else {
			e.logf(s, "warn", "工作区有 %d 个未提交改动（未开启自动提交）", dirty)
		}
	}

	// 计算初始 ahead/behind
	var ahead, behind int
	var err error
	tip, remoteHasBranch := remoteTip(cfg.RepoPath, remote, cfg.Branch)
	if remoteHasBranch {
		ahead, behind, err = aheadBehindSha(cfg.RepoPath, tip)
		if err != nil {
			e.setFail(s, "获取领先/落后信息失败: "+err.Error())
			return
		}
	} else {
		// 远程分支不存在：认为所有本地提交领先
		if out, cerr := gitRun(cfg.RepoPath, 15*time.Second, "rev-list", "--count", "HEAD"); cerr == nil {
			ahead, _ = strconv.Atoi(strings.TrimSpace(out))
		}
		e.logf(s, "warn", "远程分支 %s/%s 不存在，将推送本地 %d 个提交以建立远程分支", remote, cfg.Branch, ahead)
	}
	s.mu.Lock()
	s.ahead, s.behind, s.dirty = ahead, behind, dirty
	s.mu.Unlock()
	e.pushStatus(s)

	// FIX: 先拉取（pull），再推送（push）
	if behind > 0 && (cfg.Mode == "pull" || cfg.Mode == "both") {
		if !e.pullRemote(s, cfg, remote, behind) {
			return
		}
		// 拉取后重新计算 ahead/behind
		ahead, behind, _ = aheadBehind(cfg.RepoPath, remote, cfg.Branch)
		s.mu.Lock()
		s.ahead, s.behind = ahead, behind
		s.mu.Unlock()
		e.pushStatus(s)
	}

	if ahead > 0 && (cfg.Mode == "push" || cfg.Mode == "both") {
		if !e.pushAll(s, cfg, ahead) {
			e.setFail(s, "推送失败")
			return
		}
	}

	// 最终刷新状态
	ahead, behind, _ = aheadBehind(cfg.RepoPath, remote, cfg.Branch)
	now := time.Now().Format("2006-01-02 15:04:05")
	s.mu.Lock()
	s.ahead, s.behind = ahead, behind
	s.status = "ok"
	s.lastError = ""
	s.lastSync = &SyncResult{Time: now, Result: "ok", Detail: "同步完成"}
	s.mu.Unlock()
	e.logf(s, "info", "=== 同步完成（本地领先 %d，落后 %d）===", ahead, behind)
	e.pushStatus(s)
}

func (e *Engine) pushAll(s *repoState, cfg *RepoConfig, n int) bool {
	allOK := true
	for _, r := range cfg.Remotes {
		_, err := e.withRetry(s, "push → "+r.Name, func() (string, error) {
			return gitRun(cfg.RepoPath, 300*time.Second, "push", r.Name, cfg.Branch)
		})
		if err != nil {
			e.logf(s, "error", "推送至 %s 失败: %v", r.Name, err)
			allOK = false
		} else {
			e.logf(s, "info", "推送至 %s 成功（%d 个提交）", r.Name, n)
		}
	}
	if allOK {
		e.stat(s, "push", "success", n)
	} else {
		e.stat(s, "push", "failed", n)
	}
	return allOK
}

// FIX: pullRemote 使用 withRetry 包裹主 pull 命令
func (e *Engine) pullRemote(s *repoState, cfg *RepoConfig, remote string, n int) bool {
	args := []string{"pull"}
	if cfg.ConflictStrategy == "rebase" {
		args = append(args, "--rebase")
	}
	args = append(args, remote, cfg.Branch, "--no-edit")
	_, err := e.withRetry(s, "pull", func() (string, error) {
		return gitRun(cfg.RepoPath, 300*time.Second, args...)
	})
	if err == nil {
		e.stat(s, "pull", "success", n)
		e.logf(s, "info", "拉取成功（%d 个提交）", n)
		return true
	}
	// 处理冲突
	files, cerr := conflictFiles(cfg.RepoPath)
	if cerr == nil && len(files) > 0 {
		if cfg.ConflictStrategy == "abort" {
			_, _ = abortMerge(cfg.RepoPath)
			e.stat(s, "pull", "failed", n)
			e.setFail(s, "拉取产生冲突，已按策略（abort）自动放弃合并")
			return false
		}
		now := time.Now().Format("2006-01-02 15:04:05")
		s.mu.Lock()
		s.status = "conflict"
		s.conflicts = files
		s.lastError = "拉取产生冲突，等待在冲突处理台解决"
		s.lastSync = &SyncResult{Time: now, Result: "conflict", Detail: s.lastError}
		s.mu.Unlock()
		e.stat(s, "pull", "conflict", n)
		e.logf(s, "error", "拉取产生冲突：%s（请在详情 → 冲突中处理）", strings.Join(files, ", "))
		e.pushStatus(s)
		return false
	}
	e.stat(s, "pull", "failed", n)
	e.setFail(s, "拉取失败: "+err.Error())
	return false
}

// ---------- 手动操作 ----------

func (e *Engine) ManualGit(name, action string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	if !isRepo(s.cfg.RepoPath) {
		return fmt.Errorf("仓库路径无效: %s", s.cfg.RepoPath)
	}
	s.mu.Lock()
	if s.syncing {
		s.mu.Unlock()
		return fmt.Errorf("正在同步中，请稍候")
	}
	s.syncing = true
	cfg := s.cfg
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		s.syncing = false
		s.mu.Unlock()
	}()

	remote := firstRemoteName(cfg)
	if remote == "" {
		return fmt.Errorf("未配置远程仓库")
	}
	e.ensureRemotes(s, cfg)

	switch action {
	case "fetch":
		e.logf(s, "info", "手动 fetch…")
		if _, err := gitRun(cfg.RepoPath, 90*time.Second, "fetch", remote); err != nil {
			e.logf(s, "error", "fetch 失败: %v", err)
			return err
		}
		e.logf(s, "info", "fetch 完成")
		e.refreshCounts(s, cfg, remote)
		return nil

	case "push":
		e.logf(s, "info", "手动 push…")
		ahead, behind, _ := aheadBehind(cfg.RepoPath, remote, cfg.Branch)
		s.mu.Lock()
		s.behind = behind
		s.mu.Unlock()
		if ahead == 0 {
			e.logf(s, "info", "本地无领先提交，无需推送")
			e.refreshCounts(s, cfg, remote)
			return nil
		}
		if !e.pushAll(s, cfg, ahead) {
			e.setFail(s, "推送失败")
			return fmt.Errorf("推送失败")
		}
		e.refreshCounts(s, cfg, remote)
		e.setOK(s, "手动推送完成")
		return nil

	case "pull":
		e.logf(s, "info", "手动 pull…")
		if _, err := gitRun(cfg.RepoPath, 90*time.Second, "fetch", remote); err != nil {
			e.logf(s, "error", "fetch 失败: %v", err)
			return err
		}
		_, behind, _ := aheadBehind(cfg.RepoPath, remote, cfg.Branch)
		if behind == 0 {
			e.logf(s, "info", "远程无新提交，已是最新")
			e.refreshCounts(s, cfg, remote)
			return nil
		}
		if !e.pullRemote(s, cfg, remote, behind) {
			return fmt.Errorf("拉取失败或存在冲突")
		}
		e.refreshCounts(s, cfg, remote)
		e.setOK(s, "手动拉取完成")
		return nil
	}
	return fmt.Errorf("未知操作: %s", action)
}

// ---------- 冲突处理 ----------

func (e *Engine) ConflictAbort(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	if _, err := abortMerge(s.cfg.RepoPath); err != nil {
		e.logf(s, "error", "放弃合并失败: %v", err)
		return err
	}
	s.mu.Lock()
	s.conflicts = nil
	s.status = "ok"
	s.lastError = ""
	s.mu.Unlock()
	e.logf(s, "warn", "已放弃本次合并/变基，仓库回到冲突前状态")
	e.pushStatus(s)
	return nil
}

func (e *Engine) ConflictResolve(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	s.mu.Lock()
	rebase := s.cfg.ConflictStrategy == "rebase"
	path := s.cfg.RepoPath
	s.mu.Unlock()

	if err := resolveDone(path, rebase); err != nil {
		e.logf(s, "error", "继续合并/变基失败: %v", err)
		return err
	}
	files, _ := conflictFiles(path)
	if len(files) > 0 {
		s.mu.Lock()
		s.conflicts = files
		s.mu.Unlock()
		e.logf(s, "error", "仍有未解决的冲突：%s", strings.Join(files, ", "))
		e.pushStatus(s)
		return fmt.Errorf("仍有未解决的冲突，请在本地标记已解决后再试")
	}
	s.mu.Lock()
	s.conflicts = nil
	s.status = "ok"
	s.lastError = ""
	s.mu.Unlock()
	e.logf(s, "info", "冲突已全部解决，合并流程完成")
	e.pushStatus(s)
	return nil
}

// ---------- 配置 CRUD ----------

func (e *Engine) Create(c *RepoConfig) error {
	if err := validateRepoConfig(e.cfgDir, c, true); err != nil {
		return err
	}
	if err := saveRepoConfig(e.cfgDir, c); err != nil {
		return err
	}
	s := &repoState{cfg: c, status: "idle"}
	e.mu.Lock()
	e.states[c.Name] = s
	e.mu.Unlock()
	e.logf(s, "info", "配置已创建（路径 %s）", c.RepoPath)
	e.pushStatus(s)
	return nil
}

func (e *Engine) Update(c *RepoConfig) error {
	s := e.get(c.Name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", c.Name)
	}
	if err := validateRepoConfig(e.cfgDir, c, false); err != nil {
		return err
	}
	s.mu.Lock()
	wasRunning := s.running
	s.mu.Unlock()
	if wasRunning {
		_ = e.Stop(c.Name)
	}
	if err := saveRepoConfig(e.cfgDir, c); err != nil {
		return err
	}
	s.mu.Lock()
	s.cfg = c
	s.status = "idle"
	s.conflicts = nil
	s.lastError = ""
	s.ahead, s.behind, s.dirty = 0, 0, 0
	s.lastSync = nil
	s.mu.Unlock()
	e.logf(s, "info", "配置已更新")
	if wasRunning {
		return e.Start(c.Name)
	}
	e.pushStatus(s)
	return nil
}

func (e *Engine) Delete(name string) error {
	s := e.get(name)
	if s == nil {
		return fmt.Errorf("配置不存在: %s", name)
	}
	_ = e.Stop(name)
	s.mu.Lock()
	s.running = false
	s.mu.Unlock()
	_ = os.Remove(filepath.Join(e.cfgDir, name+".conf"))
	e.mu.Lock()
	delete(e.states, name)
	e.mu.Unlock()
	log.Printf("配置 %s 已删除", name)
	return nil
}

// ---------- 日志 / 统计 / 冲突 diff 读取 ----------

type StatRow struct {
	Time    string `json:"time"`
	Action  string `json:"action"`
	Result  string `json:"result"`
	Commits int    `json:"commits"`
}

type ConflictFile struct {
	File string `json:"file"`
	Diff string `json:"diff"`
}

func (e *Engine) ReadLog(name string, n int) []string {
	s := e.get(name)
	if s == nil {
		return []string{}
	}
	logPath := s.cfg.LogFile
	if logPath == "" {
		logPath = filepath.Join(e.logDir, name+".log")
	}
	data, err := os.ReadFile(logPath)
	if err != nil {
		return []string{}
	}
	lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
	if len(lines) > n {
		lines = lines[len(lines)-n:]
	}
	return lines
}

func (e *Engine) ReadStats(name string) []StatRow {
	f, err := os.Open(filepath.Join(e.statsDir, name+".csv"))
	if err != nil {
		return []StatRow{}
	}
	defer f.Close()
	records, err := csv.NewReader(f).ReadAll()
	if err != nil {
		return []StatRow{}
	}
	rows := []StatRow{}
	for i, rec := range records {
		if i == 0 || len(rec) < 4 {
			continue
		}
		c, _ := strconv.Atoi(rec[3])
		rows = append(rows, StatRow{Time: rec[0], Action: rec[1], Result: rec[2], Commits: c})
	}
	return rows
}

func (e *Engine) ConflictDiffs(name string) []ConflictFile {
	s := e.get(name)
	if s == nil {
		return []ConflictFile{}
	}
	s.mu.Lock()
	files := append([]string(nil), s.conflicts...)
	path := s.cfg.RepoPath
	s.mu.Unlock()
	if len(files) == 0 {
		files, _ = conflictFiles(path)
	}
	out := []ConflictFile{}
	for _, f := range files {
		d, err := gitRun(path, 30*time.Second, "diff", "--", f)
		if err != nil {
			d = err.Error()
		}
		if len(d) > 65536 {
			d = d[:65536] + "\n…（diff 过长已截断）"
		}
		out = append(out, ConflictFile{File: f, Diff: d})
	}
	return out
}

// ---------- 提交历史 ----------

var hashRe = regexp.MustCompile(`^[0-9a-fA-F]{6,40}$`)

type Commit struct {
	Hash    string `json:"hash"`
	Author  string `json:"author"`
	Date    string `json:"date"`
	Subject string `json:"subject"`
}

func (e *Engine) ReadCommits(name string, limit int) []Commit {
	s := e.get(name)
	if s == nil {
		return []Commit{}
	}
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	out, err := gitRun(s.cfg.RepoPath, 30*time.Second,
		"log", "-n", strconv.Itoa(limit),
		"--date=format:%Y-%m-%d %H:%M",
		"--pretty=format:%H%x1f%an%x1f%ad%x1f%s")
	if err != nil {
		return []Commit{}
	}
	commits := []Commit{}
	for _, line := range strings.Split(out, "\n") {
		parts := strings.Split(line, "\x1f")
		if len(parts) != 4 {
			continue
		}
		commits = append(commits, Commit{Hash: parts[0], Author: parts[1], Date: parts[2], Subject: parts[3]})
	}
	return commits
}

func (e *Engine) ReadCommitDiff(name, hash string) (string, error) {
	s := e.get(name)
	if s == nil {
		return "", fmt.Errorf("配置不存在: %s", name)
	}
	if !hashRe.MatchString(hash) {
		return "", fmt.Errorf("无效的提交哈希")
	}
	out, err := gitRun(s.cfg.RepoPath, 60*time.Second, "show", "--no-color", hash)
	if err != nil {
		return "", err
	}
	if len(out) > 131072 {
		out = out[:131072] + "\n…（内容过长已截断）"
	}
	return out, nil
}