Files
github-auto-sync-golang/engine.go
T

1518 lines
42 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
import (
"bytes"
"encoding/csv"
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"regexp"
"runtime/debug"
"sort"
"strconv"
"strings"
"sync"
"time"
)
// ---------- 事件总线 ----------
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 | push_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"`
PushStrategy string `json:"push_strategy"`
LogFile string `json:"log_file"`
AutoCommit bool `json:"auto_commit"`
AutoSync bool `json:"auto_sync"`
Cron string `json:"cron"`
Remotes []Remote `json:"remotes"`
Running bool `json:"running"`
Status string `json:"status"` // idle | ok | error | conflict | push_conflict
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"`
// 新增:推送冲突信息
PushConflict bool `json:"push_conflict"`
PushConflictRemote string `json:"push_conflict_remote"`
PushConflictBranch string `json:"push_conflict_branch"`
}
type repoState struct {
mu sync.Mutex
cfg *RepoConfig
running bool
syncing bool
status string // idle | ok | error | conflict | push_conflict
ahead int
behind int
dirty int
conflicts []string
lastError string
lastSync *SyncResult
stopCh chan struct{}
// 推送冲突信息
pushConflict bool
pushConflictRemote string
pushConflictBranch string
}
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
}
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+"] ["+level+"] "+msg)
_ = f.Close()
}
}
e.publish(map[string]any{"type": "log", "repo": s.cfg.Name, "level": level, "msg": msg, "time": ts})
}
// statMaxLines 控制单个仓库统计 CSV 的最大行数(含表头),超过则滚动保留最近 N 行
const statMaxLines = 2000
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)
}
// 文件过大时滚动截断,避免长期运行无限膨胀(统计频率低,重写成本可接受)
if rotateStatIfNeeded(path, statMaxLines) {
e.logf(s, "info", "统计文件已滚动截断至最近 %d 行", statMaxLines)
}
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)
}
// rotateStatIfNeeded 当 CSV 行数超过 maxLines 时,保留表头 + 最近 (maxLines-1) 条数据行并重写
func rotateStatIfNeeded(path string, maxLines int) bool {
data, err := os.ReadFile(path)
if err != nil {
return false
}
lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
if len(lines) <= maxLines {
return false
}
keep := append([]string{lines[0]}, lines[len(lines)-(maxLines-1):]...)
if err := os.WriteFile(path, []byte(strings.Join(keep, "\n")+"\n"), 0o644); err != nil {
return false
}
return true
}
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,
PushStrategy: cfg.PushStrategy,
LogFile: cfg.LogFile,
AutoCommit: cfg.AutoCommit,
AutoSync: cfg.AutoSync,
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,
PushConflict: s.pushConflict,
PushConflictRemote: s.pushConflictRemote,
PushConflictBranch: s.pushConflictBranch,
}
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
autoSync := s.cfg.AutoSync
s.mu.Unlock()
// 自动同步关闭:不再自动触发,仅保留监控与手动同步能力
if !autoSync {
e.logf(s, "info", "已关闭自动同步,仅响应手动同步")
e.pushStatus(s)
<-stopCh
return
}
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 把配置中的远程同步到仓库
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:捕获 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)
}
// syncOnce 核心同步逻辑(已调整顺序,并增加 push_conflict 处理)
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()
}()
// 自动 stash 恢复(B4):若本次同步前暂存了改动,结束时尝试恢复
stashed := false
defer func() {
if !stashed {
return
}
if _, perr := gitRun(cfg.RepoPath, 60*time.Second, "stash", "pop"); perr != nil {
e.logf(s, "warn", "同步前自动 stash 的恢复失败(可能与其他改动冲突),请手动执行 git stash pop: %v", perr)
} else {
e.logf(s, "info", "已恢复同步前自动 stash 的未提交改动")
}
}()
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
}
// 自动 stash(B4):未开启自动提交且有未提交改动时先暂存,避免分支切换/推送被脏工作区阻断
if !cfg.AutoCommit && dirtyCount(cfg.RepoPath) > 0 {
if _, serr := gitRun(cfg.RepoPath, 60*time.Second, "stash", "push", "-u", "-m", "git-sync-web auto-stash"); serr == nil {
stashed = true
e.logf(s, "info", "已自动 stash 未提交改动(同步结束后恢复)")
} else {
e.logf(s, "warn", "自动 stash 失败,将尝试直接同步")
}
}
// 分支对齐(安全切换)
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)
// ---- 先拉取 ----
if behind > 0 && (cfg.Mode == "pull" || cfg.Mode == "both") {
if !e.pullRemote(s, cfg, remote, behind) {
return
}
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") {
pushOK, pushErr := e.pushAll(s, cfg, ahead)
if !pushOK {
// 检查是否是 non-fast-forward 错误,按推送策略处理(B3)
if pushErr != nil && strings.Contains(strings.ToLower(pushErr.Error()), "non-fast-forward") {
switch cfg.PushStrategy {
case "force":
e.logf(s, "warn", "按推送策略 force 执行强制推送(--force-with-lease)")
if _, ferr := gitRun(cfg.RepoPath, 300*time.Second, "push", "--force-with-lease", remote, cfg.Branch); ferr != nil {
e.setFail(s, "强制推送失败: "+ferr.Error())
return
}
e.logf(s, "info", "强制推送成功")
e.stat(s, "push", "success", ahead)
s.mu.Lock()
s.pushConflict = false
s.lastError = ""
s.mu.Unlock()
case "rebase":
e.logf(s, "warn", "按推送策略 rebase 后重试推送")
if _, rerr := gitRun(cfg.RepoPath, 300*time.Second, "rebase", remote+"/"+cfg.Branch); rerr != nil {
e.setFail(s, "变基到远程失败(请手动解决冲突): "+rerr.Error())
return
}
if _, perr := gitRun(cfg.RepoPath, 300*time.Second, "push", remote, cfg.Branch); perr != nil {
e.setFail(s, "变基后推送失败: "+perr.Error())
return
}
e.logf(s, "info", "变基后推送成功")
e.stat(s, "push", "success", ahead)
s.mu.Lock()
s.pushConflict = false
s.lastError = ""
s.mu.Unlock()
default:
// manual:保持 push_conflict 状态,等待用户在控制台决策(强制推送 / 放弃)
e.logf(s, "warn", "推送被拒绝(non-fast-forward),等待用户决策")
return
}
} else {
e.setFail(s, "推送失败: "+pushErr.Error())
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.pushConflict = false
s.pushConflictRemote = ""
s.pushConflictBranch = ""
s.mu.Unlock()
e.logf(s, "info", "=== 同步完成(本地领先 %d,落后 %d)===", ahead, behind)
e.pushStatus(s)
}
// pushAll 返回 (是否全部成功, 错误信息)
// 如果遇到 non-fast-forward,会设置 pushConflict 状态并返回 false,错误包含 "non-fast-forward"
func (e *Engine) pushAll(s *repoState, cfg *RepoConfig, n int) (bool, error) {
allOK := true
var firstErr error
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
if firstErr == nil {
firstErr = err
}
// 检测到 non-fast-forward,记录状态并继续尝试其他远程
if strings.Contains(strings.ToLower(err.Error()), "non-fast-forward") {
s.mu.Lock()
s.status = "push_conflict"
s.pushConflict = true
s.pushConflictRemote = r.Name
s.pushConflictBranch = cfg.Branch
s.lastError = "推送被拒绝(non-fast-forward),请选择强制覆盖或放弃"
s.lastSync = &SyncResult{
Time: time.Now().Format("2006-01-02 15:04:05"),
Result: "push_conflict",
Detail: fmt.Sprintf("远程 %s 有新提交,推送被拒绝", r.Name),
}
s.mu.Unlock()
e.pushStatus(s)
}
} else {
e.logf(s, "info", "推送至 %s 成功(%d 个提交)", r.Name, n)
}
}
if allOK {
e.stat(s, "push", "success", n)
return true, nil
}
e.stat(s, "push", "failed", n)
return false, firstErr
}
// pullRemote 已使用重试
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
}
// ---------- 推送冲突处理 ----------
// ForcePush 强制推送(--force-with-lease)到冲突的远程
func (e *Engine) ForcePush(name string) error {
s := e.get(name)
if s == nil {
return fmt.Errorf("配置不存在: %s", name)
}
s.mu.Lock()
if !s.pushConflict {
s.mu.Unlock()
return fmt.Errorf("当前没有待处理的推送冲突")
}
remote := s.pushConflictRemote
branch := s.pushConflictBranch
path := s.cfg.RepoPath
s.mu.Unlock()
e.logf(s, "info", "执行强制推送至 %s %s", remote, branch)
_, err := gitRun(path, 300*time.Second, "push", "--force-with-lease", remote, branch)
if err != nil {
e.logf(s, "error", "强制推送失败: %v", err)
return err
}
e.logf(s, "info", "强制推送成功")
s.mu.Lock()
s.pushConflict = false
s.pushConflictRemote = ""
s.pushConflictBranch = ""
s.status = "ok"
s.lastError = ""
s.lastSync = &SyncResult{Time: time.Now().Format("2006-01-02 15:04:05"), Result: "ok", Detail: "强制推送完成"}
s.mu.Unlock()
e.pushStatus(s)
return nil
}
// AbortPush 放弃强制推送,清除冲突状态
func (e *Engine) AbortPush(name string) error {
s := e.get(name)
if s == nil {
return fmt.Errorf("配置不存在: %s", name)
}
s.mu.Lock()
defer s.mu.Unlock()
if !s.pushConflict {
return fmt.Errorf("当前没有待处理的推送冲突")
}
s.pushConflict = false
s.pushConflictRemote = ""
s.pushConflictBranch = ""
s.status = "ok"
s.lastError = "用户放弃强制推送"
s.lastSync = &SyncResult{Time: time.Now().Format("2006-01-02 15:04:05"), Result: "ok", Detail: "放弃强制推送,状态已清除"}
e.logf(s, "warn", "放弃强制推送")
e.pushStatus(s)
return nil
}
// ---------- 手动操作 ----------
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
}
ok, err := e.pushAll(s, cfg, ahead)
if !ok {
e.logf(s, "error", "推送失败: %v", err)
return err
}
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.pushConflict = false
s.pushConflictRemote = ""
s.pushConflictBranch = ""
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
}
// ---------- 差异与文件编辑(网页端) ----------
const fileEditMaxBytes = 1 << 20 // 1MB,网页编辑单文件上限
// safeRepoPath 将仓库内相对路径解析为绝对路径,并严格校验不越界、不触及 .git
func (e *Engine) safeRepoPath(s *repoState, rel string) (string, error) {
rel = filepath.ToSlash(strings.TrimSpace(rel))
rel = strings.TrimPrefix(rel, "/")
if rel == "" || rel == "." {
return "", fmt.Errorf("无效路径")
}
for _, seg := range strings.Split(rel, "/") {
if seg == ".." {
return "", fmt.Errorf("路径不允许包含 ..")
}
}
root := filepath.Clean(s.cfg.RepoPath)
full := filepath.Clean(filepath.Join(root, filepath.FromSlash(rel)))
if full != root && !strings.HasPrefix(full, root+string(os.PathSeparator)) {
return "", fmt.Errorf("路径越界: %s", rel)
}
relClean := strings.TrimPrefix(full, root+string(os.PathSeparator))
if relClean == ".git" || strings.HasPrefix(relClean, ".git"+string(os.PathSeparator)) {
return "", fmt.Errorf("禁止访问 .git 内部")
}
return full, nil
}
// FileNode 文件树节点(网页端)
type FileNode struct {
Path string `json:"path"`
Status string `json:"status"` // M 已修改 / A 新增 / D 删除 / ? 未跟踪 / 空 未改动
}
// FileTree 列出仓库内可编辑文件(已跟踪 + 未跟踪且未被忽略),天然避开 .git 内部
func (e *Engine) FileTree(name string) []FileNode {
s := e.get(name)
if s == nil {
return []FileNode{}
}
set := map[string]struct{}{}
collect := func(args ...string) {
if out, err := gitRun(s.cfg.RepoPath, 15*time.Second, args...); err == nil {
for _, ln := range strings.Split(strings.TrimSpace(out), "\n") {
ln = strings.TrimSpace(ln)
if ln != "" {
set[ln] = struct{}{}
}
}
}
}
collect("ls-files")
collect("ls-files", "--others", "--exclude-standard")
paths := make([]string, 0, len(set))
for p := range set {
paths = append(paths, p)
}
sort.Strings(paths)
// 附带每个文件的改动状态(M/A/D/?),便于网页端一眼看出哪些文件被动过
status := changedFiles(s.cfg.RepoPath)
nodes := make([]FileNode, 0, len(paths))
for _, p := range paths {
nodes = append(nodes, FileNode{Path: p, Status: status[p]})
}
return nodes
}
// FileContent 文件内容读取结果
type FileContent struct {
Path string `json:"path"`
Content string `json:"content"`
Binary bool `json:"binary"`
Size int `json:"size"`
}
// ReadFileContent 读取仓库内文件内容(带大小上限与二进制检测)
func (e *Engine) ReadFileContent(name, rel string) (*FileContent, error) {
s := e.get(name)
if s == nil {
return nil, fmt.Errorf("配置不存在: %s", name)
}
full, err := e.safeRepoPath(s, rel)
if err != nil {
return nil, err
}
info, err := os.Stat(full)
if err != nil {
return nil, err
}
if info.IsDir() {
return nil, fmt.Errorf("这是一个目录: %s", rel)
}
if info.Size() > fileEditMaxBytes {
return nil, fmt.Errorf("文件过大(>%d 字节),禁止在网页编辑", fileEditMaxBytes)
}
data, err := os.ReadFile(full)
if err != nil {
return nil, err
}
if bytes.Contains(data, []byte{0}) {
return &FileContent{Path: rel, Binary: true, Size: int(info.Size())}, nil
}
return &FileContent{Path: rel, Content: string(data), Size: int(info.Size())}, nil
}
// WriteFileContent 将内容写回仓库内文件(路径校验 + 大小上限 + 二进制拒绝)
func (e *Engine) WriteFileContent(name, rel, content string) error {
s := e.get(name)
if s == nil {
return fmt.Errorf("配置不存在: %s", name)
}
full, err := e.safeRepoPath(s, rel)
if err != nil {
return err
}
if len(content) > fileEditMaxBytes {
return fmt.Errorf("内容过大(>%d 字节)", fileEditMaxBytes)
}
if strings.Contains(content, "\x00") {
return fmt.Errorf("内容含空字符,疑似二进制,禁止写入")
}
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
return err
}
return os.WriteFile(full, []byte(content), 0o644)
}
// CommitFile 仅提交单个被网页端编辑的文件
func (e *Engine) CommitFile(name, rel, message string) error {
s := e.get(name)
if s == nil {
return fmt.Errorf("配置不存在: %s", name)
}
if _, err := e.safeRepoPath(s, rel); err != nil {
return err
}
if strings.TrimSpace(message) == "" {
message = "网页端编辑: " + rel
}
if _, err := gitRun(s.cfg.RepoPath, 30*time.Second, "add", "--", rel); err != nil {
return fmt.Errorf("git add 失败: %w", err)
}
if _, err := gitRun(s.cfg.RepoPath, 60*time.Second, "commit", "-m", message, "--", rel); err != nil {
return fmt.Errorf("git commit 失败: %w", err)
}
e.logf(s, "info", "网页端提交 %s 成功", rel)
return nil
}
// ResolveConflictFile 网页端解决冲突:写入内容 → 暂存 → (可选)结束合并/变基
// finish=true 且已无其它冲突文件时才真正结束,避免把带冲突标记的内容提交进去
func (e *Engine) ResolveConflictFile(name, rel, content string, finish bool) error {
s := e.get(name)
if s == nil {
return fmt.Errorf("配置不存在: %s", name)
}
if err := e.WriteFileContent(name, rel, content); err != nil {
return err
}
if _, err := gitRun(s.cfg.RepoPath, 30*time.Second, "add", "--", rel); err != nil {
return fmt.Errorf("git add 失败: %w", err)
}
e.logf(s, "info", "网页端冲突文件 %s 已暂存", rel)
if !finish {
return nil
}
rest, _ := conflictFiles(s.cfg.RepoPath)
if len(rest) > 0 {
e.pushStatus(s)
return fmt.Errorf("已暂存 %s;但仍有未解决的冲突文件:%s", rel, strings.Join(rest, ", "))
}
return e.ConflictResolve(name)
}
// DiffFile 单文件增删统计
type DiffFile struct {
Path string `json:"path"`
Add int `json:"add"`
Del int `json:"del"`
}
// DiffResult 差异结果(网页端「差异」tab)
type DiffResult struct {
Scope string `json:"scope"`
Stat string `json:"stat"`
Text string `json:"text"`
Files []DiffFile `json:"files"`
VsRemote bool `json:"vs_remote"`
Word bool `json:"word"`
}
// WorkingDiff 返回工作区/暂存/全部/与远程的差异
// rel 非空时只对比单个文件(已做路径安全校验)
// mode = "word" 时改用 --word-diff=plain,便于词级定位改动
func (e *Engine) WorkingDiff(name, scope, rel, mode string) (*DiffResult, error) {
s := e.get(name)
if s == nil {
return nil, fmt.Errorf("配置不存在: %s", name)
}
res := &DiffResult{Scope: scope}
var args []string
remote := "origin"
switch scope {
case "staged":
args = []string{"diff", "--cached", "--no-color"}
case "all":
args = []string{"diff", "HEAD", "--no-color"}
case "remote":
ahead, _, _ := aheadBehind(s.cfg.RepoPath, remote, s.cfg.Branch)
if ahead <= 0 {
res.Text = "本地没有领先远程的提交,无需对比。"
return res, nil
}
args = []string{"diff", "--no-color", remote + "/" + s.cfg.Branch, "HEAD"}
res.VsRemote = true
default: // working(未暂存)
args = []string{"diff", "--no-color"}
}
// 词级差异:--word-diff=plain 用 [-x-] / {+x+} 标记行内改动
// 注意:该模式下 git 输出的上下文行不带前导空格,前端解析需兼容(见 parseUnifiedDiff)
if mode == "word" {
args = append(args, "--word-diff=plain")
res.Word = true
}
statArgs := append([]string{}, args...)
if scope == "remote" {
statArgs = []string{"diff", "--no-color", "--stat", remote + "/" + s.cfg.Branch, "HEAD"}
} else {
statArgs = append(statArgs, "--stat")
}
// 单文件过滤:追加 -- <path>,先做路径安全校验(防穿越 / .git)
if rel != "" {
if _, err := e.safeRepoPath(s, rel); err != nil {
return nil, err
}
args = append(args, "--", rel)
statArgs = append(statArgs, "--", rel)
}
if statOut, serr := gitRun(s.cfg.RepoPath, 30*time.Second, statArgs...); serr == nil {
statOut = stripWarnings(statOut)
res.Stat = statOut
res.Files = parseStatFiles(statOut)
}
if textOut, terr := gitRun(s.cfg.RepoPath, 60*time.Second, args...); terr == nil {
res.Text = stripWarnings(textOut)
} else {
res.Text = terr.Error()
}
if len(res.Text) > 262144 {
res.Text = res.Text[:262144] + "\n…(diff 过长已截断)"
}
return res, nil
}
// stripWarnings 去掉 git 在 stderr 输出的环境告警(如 autocrlf 提示),避免污染差异展示
func stripWarnings(s string) string {
lines := strings.Split(s, "\n")
out := make([]string, 0, len(lines))
for _, l := range lines {
if strings.HasPrefix(l, "warning:") {
continue
}
out = append(out, l)
}
return strings.Join(out, "\n")
}
// parseStatFiles 解析 `git diff --stat` 输出为单文件增删统计
func parseStatFiles(stat string) []DiffFile {
var files []DiffFile
for _, line := range strings.Split(stat, "\n") {
if strings.Contains(line, "files changed") || strings.Contains(line, "file changed") {
continue
}
idx := strings.Index(line, " | ")
if idx < 0 {
continue
}
path := strings.TrimSpace(line[:idx])
rest := line[idx+3:]
add, del := 0, 0
for _, r := range rest {
if r == '+' {
add++
} else if r == '-' {
del++
}
}
if path != "" {
files = append(files, DiffFile{Path: path, Add: add, Del: del})
}
}
return files
}