Blame for internal/syncer/syncer.go

package syncer

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"
)

// Mode 决定这一轮允许往哪个方向写。
//
// 注意这里**没有** "status/dry-run" 这一档：只报告不写入是「算完不落地」，
// 不是「不许算」——两者混在一个枚举里会让 status 把该有的差异也算成"无事可做"，
// 从而漏报真正的不一致（实现初期正是这么错的）。只读由 Apply 的 dryRun 参数表达。
type Mode string

const (
	// ModeSync 双向对齐：缺的补上、冲突按判据取新的一方。
	ModeSync Mode = "sync"
	// ModePull 只往下拉（远端 → 本地）。本地独有的文件不动。
	ModePull Mode = "pull"
	// ModePush 只往上推（本地 → 远端）。远端独有的文件不动。
	ModePush Mode = "push"
)

// ParseCommand 把子命令解析为 (同步模式, 是否只报告不写入)。
//
// "status" / "diff" / "plan" 一律按**双向**算一遍，只是不落地——这样它报出来的
// 差异和真正 sync 会做的事完全一致，可以直接当"预演"用。
func ParseCommand(s string) (Mode, bool, error) {
	switch strings.ToLower(strings.TrimSpace(s)) {
	case "", "sync", "all":
		return ModeSync, false, nil
	case "pull", "down", "download":
		return ModePull, false, nil
	case "push", "up", "upload":
		return ModePush, false, nil
	case "status", "diff", "plan", "dry-run", "dryrun":
		return ModeSync, true, nil
	default:
		return "", false, fmt.Errorf("未知子命令 %q（可选 sync/pull/push/status/watch）", s)
	}
}

// allowsUpload 报告该模式是否允许写远端。
func (m Mode) allowsUpload() bool { return m == ModeSync || m == ModePush }

// allowsDownload 报告该模式是否允许写本地。
func (m Mode) allowsDownload() bool { return m == ModeSync || m == ModePull }

// Op 是一个同步动作。
type Op string

const (
	// OpNoop 两端已一致或按模式不动。
	OpNoop Op = "noop"
	// OpUpload 本地 → 远端。
	OpUpload Op = "upload"
	// OpDownload 远端 → 本地。
	OpDownload Op = "download"
)

// Action 是计划里的一条动作。
type Action struct {
	Op       Op     `json:"op"`
	Path     string `json:"path"`
	Category string `json:"category"`
	Reason   string `json:"reason"`
	// Conflict 表示两端都有且内容不同（真正的冲突，而不是单边缺失）。
	Conflict bool       `json:"conflict"`
	Local    *FileState `json:"local,omitempty"`
	Remote   *FileState `json:"remote,omitempty"`

	// Applied / Err 由 Apply 回填。
	Applied bool   `json:"applied"`
	Err     string `json:"err,omitempty"`
}

// Plan 是一次对账的结果。
type Plan struct {
	Mode    Mode
	Actions []Action

	// Bucket / Writer 随计划带上，方便输出与排障时一眼看清"对的是哪个桶、会用哪条写入通道"。
	Bucket string
	Writer string

	LocalFiles  int
	RemoteFiles int
}

// Uploads 返回所有上传动作。
func (p *Plan) Uploads() []Action { return p.filter(OpUpload) }

// Downloads 返回所有下载动作。
func (p *Plan) Downloads() []Action { return p.filter(OpDownload) }

// Conflicts 返回所有真冲突（两端都有、内容不同、需要裁决）。
func (p *Plan) Conflicts() []Action {
	var out []Action
	for _, a := range p.Actions {
		if a.Conflict {
			out = append(out, a)
		}
	}
	return out
}

// Changes 报告计划里是否有实际改动。
func (p *Plan) Changes() int { return len(p.Uploads()) + len(p.Downloads()) }

func (p *Plan) filter(op Op) []Action {
	var out []Action
	for _, a := range p.Actions {
		if a.Op == op {
			out = append(out, a)
		}
	}
	return out
}

// Result 是一次执行的统计。
type Result struct {
	Mode       Mode
	Uploaded   int
	Downloaded int
	Skipped    int
	Failed     int
	Actions    []Action
	Errors     []string
}

// Config 是同步配置。字段与 config.json 的 "sync" 段一一对应。
type Config struct {
	// Bucket 桶 ID，形如 "a3216/wb2api-storage"。
	Bucket string `json:"bucket"`
	// Token 具写权限的 HF token。为空时回落到 TokenFile。
	Token string `json:"token"`
	// TokenFile 存放 token 的文件（token 不进 config.json 时用）。
	TokenFile string `json:"token_file"`
	// Endpoint HF 端点，默认 https://huggingface.co。
	Endpoint string `json:"endpoint"`
	// AuthDir 本地凭证目录，与网关 auth_dir 同值。
	AuthDir string `json:"auth_dir"`
	// DataDir 本地数据目录（state_file 所在目录）。
	DataDir string `json:"data_dir"`
	// IncludeData 除 auths/ 之外还要同步的 data/ 顶层文件名白名单。
	// 默认空 = 只同步凭证。理由见 docs：state.json 是两个进程各自的运行态，
	// 互相搬运会把对端的冷却/熔断状态带进来。
	IncludeData []string `json:"include_data"`
	// RelayURL 中转网关地址（如 https://cli.3216gemini.dpdns.org），relay 写入通道用。
	RelayURL string `json:"relay_url"`
	// RelayKey 中转网关的 api_key（Bearer）。为空且配置了桶时可自动从桶的
	// config.json 读取。
	RelayKey string `json:"relay_key"`
	// Writer 写入通道：auto（默认）/ hf / relay。
	Writer string `json:"writer"`
	// HFBin hf 可执行文件名/路径，默认 "hf"。
	HFBin string `json:"hf_bin"`
	// RestartSpace 形如 "a3216/gcli2api"。配置后，每当本轮向 auths/ 推了新凭证
	// 就请求重启该 Space——Space 上的网关只在启动时扫凭证目录，不重启就加载不到新账号。
	// 空 = 从不重启（省下 30~60s 冷启动，代价是新账号要等下次 Space 重启才生效）。
	RestartSpace string `json:"restart_space"`
	// CABundle 额外的 CA 证书包（PEM）路径。
	//
	// 本机跑抓包代理（DevSidecar / Reqable / ProxyPin 之类）时会出现一种很别扭的
	// 不对称：Go 在 Windows 上吃系统信任库，所以**读**同步一切正常；但写通道用的
	// hf CLI 底层是 httpx，只认 certifi，会把代理的中间人证书判成不可信而失败。
	// 把代理根证书导出成 PEM 指到这里，就会同时喂给 Go 客户端（追加信任，不替换
	// 系统库）与 hf 子进程（SSL_CERT_FILE / REQUESTS_CA_BUNDLE / CURL_CA_BUNDLE）。
	CABundle string `json:"ca_bundle"`
	// Timeout 单次网络操作超时，默认 60s。
	Timeout string `json:"timeout"`
	// InsecureSkipVerify 关闭 TLS 校验（仅本地抓包代理排障用，默认 false）。
	InsecureSkipVerify bool `json:"insecure_skip_verify"`
}

// Normalize 补默认值。
func (c *Config) Normalize() {
	if strings.TrimSpace(c.Endpoint) == "" {
		c.Endpoint = DefaultEndpoint
	}
	if strings.TrimSpace(c.Writer) == "" {
		c.Writer = "auto"
	}
	if strings.TrimSpace(c.HFBin) == "" {
		c.HFBin = "hf"
	}
	for i, d := range c.IncludeData {
		c.IncludeData[i] = strings.TrimSpace(d)
	}
	c.Bucket = strings.Trim(strings.TrimSpace(c.Bucket), "/")
	c.Endpoint = strings.TrimRight(strings.TrimSpace(c.Endpoint), "/")
	c.RelayURL = strings.TrimRight(strings.TrimSpace(c.RelayURL), "/")
}

// timeout 解析 Timeout 字段。
func (c *Config) timeout() time.Duration {
	if strings.TrimSpace(c.Timeout) == "" {
		return 60 * time.Second
	}
	d, err := time.ParseDuration(c.Timeout)
	if err != nil || d <= 0 {
		return 60 * time.Second
	}
	return d
}

// Engine 把配置、桶客户端与写入通道绑在一起。
type Engine struct {
	cfg    Config
	bucket *BucketClient
	writer Writer
}

// New 构造 Engine。桶只读客户端必建；写入通道按 writer 配置选，选不出来也不
// 立刻失败——只读（pull/status）依然可用，真正需要写时 Apply 会报错。
func New(cfg Config) (*Engine, error) {
	cfg.Normalize()
	if cfg.Bucket == "" {
		return nil, fmt.Errorf("sync.bucket 未配置")
	}
	token, err := cfg.resolveToken()
	if err != nil {
		return nil, err
	}
	cfg.Token = token

	e := &Engine{
		cfg:    cfg,
		bucket: NewBucketClient(cfg.Endpoint, cfg.Bucket, token, cfg.timeout(), cfg.InsecureSkipVerify, cfg.CABundle),
	}
	if w, werr := NewWriter(cfg); werr == nil {
		e.writer = w
	}
	return e, nil
}

// Bucket 暴露只读客户端（cmd/sync 读远端 api_key 时用）。
func (e *Engine) Bucket() *BucketClient { return e.bucket }

// WriterName 返回当前写入通道名（未选出时返回空串）。
func (e *Engine) WriterName() string {
	if e.writer == nil {
		return ""
	}
	return e.writer.Name()
}

// Config 返回归一化后的配置副本。
func (e *Engine) Config() Config { return e.cfg }

// resolveToken 取 token：字段优先，其次 token_file，其次 HF_TOKEN 环境变量。
func (c *Config) resolveToken() (string, error) {
	if t := strings.TrimSpace(c.Token); t != "" {
		return t, nil
	}
	if fp := strings.TrimSpace(c.TokenFile); fp != "" {
		raw, err := readTrimmedFile(fp)
		if err != nil {
			return "", err
		}
		return raw, nil
	}
	return "", nil // 允许匿名读公开桶；写的时候才会因 401 报错
}

// FetchRemoteAPIKey 从桶里的 config.json 读取中转网关的 api_key。
//
// 存在的意义：relay 写入通道要带 Bearer，而那个 key 就躺在（私有的）桶里。
// 桶已经要凭 token 才能读了，再去手工抄一遍 key 没有意义。
func (e *Engine) FetchRemoteAPIKey(ctx context.Context) (string, error) {
	raw, err := e.bucket.Get(ctx, "config.json")
	if err != nil {
		return "", err
	}
	var cfg struct {
		APIKey string `json:"api_key"`
	}
	if err := json.Unmarshal(raw, &cfg); err != nil {
		return "", fmt.Errorf("解析远端 config.json 失败: %w", err)
	}
	if strings.TrimSpace(cfg.APIKey) == "" {
		return "", fmt.Errorf("远端 config.json 里 api_key 为空")
	}
	return cfg.APIKey, nil
}

// ---------------------------------------------------------------------------
// 对账
// ---------------------------------------------------------------------------

// Plan 计算这一轮要做什么，不产生任何副作用。
func (e *Engine) Plan(ctx context.Context, mode Mode) (*Plan, error) {
	local, err := e.scanLocal()
	if err != nil {
		return nil, fmt.Errorf("扫描本地失败: %w", err)
	}
	remote, err := e.collectRemote(ctx)
	if err != nil {
		return nil, err
	}
	p := buildPlan(mode, local, remote)
	p.Bucket = e.cfg.Bucket
	p.Writer = e.WriterName()
	return p, nil
}

// buildPlan 是纯函数形式的对账逻辑，便于单测覆盖各种组合。
func buildPlan(mode Mode, local, remote map[string]*FileState) *Plan {
	p := &Plan{Mode: mode, LocalFiles: len(local), RemoteFiles: len(remote)}

	paths := make([]string, 0, len(local)+len(remote))
	seen := map[string]bool{}
	for k := range local {
		paths = append(paths, k)
		seen[k] = true
	}
	for k := range remote {
		if !seen[k] {
			paths = append(paths, k)
		}
	}
	sort.Strings(paths)

	for _, path := range paths {
		l, r := local[path], remote[path]
		p.Actions = append(p.Actions, decideAction(mode, path, l, r))
	}
	return p
}

// decideAction 判定单个路径的动作。
func decideAction(mode Mode, path string, l, r *FileState) Action {
	category := catAuth
	if strings.HasPrefix(path, "data/") {
		category = catData
	}
	a := Action{Path: path, Category: category, Local: l, Remote: r}

	switch {
	case l == nil && r == nil:
		a.Op = OpNoop
		a.Reason = "两端均不存在（不应发生）"

	case r == nil:
		if mode.allowsUpload() {
			a.Op, a.Reason = OpUpload, "远端缺失，本端独有"
		} else {
			a.Op, a.Reason = OpNoop, "本端独有；当前模式不允许写远端"
		}

	case l == nil:
		if mode.allowsDownload() {
			a.Op, a.Reason = OpDownload, "本地缺失，远端独有"
		} else {
			a.Op, a.Reason = OpNoop, "远端独有；当前模式不允许写本地"
		}

	default:
		if l.SHA256 == r.SHA256 {
			a.Op, a.Reason = OpNoop, "两端内容一致"
			return a
		}
		a.Conflict = true

		var winner, reason string
		if category == catAuth {
			winner, reason = decideAuthWinner(l, r)
		} else {
			winner, reason = decideDataWinner(l, r)
		}
		switch {
		case winner == sideLocal && mode.allowsUpload():
			a.Op, a.Reason = OpUpload, reason
		case winner == sideRemote && mode.allowsDownload():
			a.Op, a.Reason = OpDownload, reason
		case winner == sideLocal:
			a.Op, a.Reason = OpNoop, reason+"；但当前模式不允许写远端"
		default:
			a.Op, a.Reason = OpNoop, reason+"；但当前模式不允许写本地"
		}
	}
	return a
}

// collectRemote 列出桶内候选文件并下载内容。
//
// 为什么要下载才算得出结论：桶 tree 里的 xetHash 是 Xet 的内容哈希，
// **不是**文件的 SHA-256（已实测：同一文件两者不同）。所以"是否一致"只能靠
// 拉下来自己算。候选集很小（凭证各 ~2.3KB、可选数据 ~2.7KB），这个代价可以忽略；
// 反过来，用 xetHash 当 SHA-256 会误判成"永远不一致"，每轮都白传一遍。
func (e *Engine) collectRemote(ctx context.Context) (map[string]*FileState, error) {
	files, err := e.bucket.List(ctx, "")
	if err != nil {
		return nil, fmt.Errorf("列出桶目录失败: %w", err)
	}
	out := map[string]*FileState{}
	for _, rf := range files {
		if !e.isCandidate(rf.Path) {
			continue
		}
		raw, err := e.bucket.Get(ctx, rf.Path)
		if err != nil {
			return nil, fmt.Errorf("下载 %s 失败: %w", rf.Path, err)
		}
		category := catAuth
		if strings.HasPrefix(rf.Path, "data/") {
			category = catData
		}
		out[rf.Path] = newFileState(rf.Path, raw, rf.Mtime, category)
	}
	return out, nil
}

// isCandidate 判定桶内一个路径是否属于同步范围。
//
// 明确排除：
//   - config.json —— 两端的 listen / auth_dir / state_file / api_key 本来就不同，
//     互相覆盖会把对端跑不起来。配置是"每端自己的"，不是要同步的数据。
//   - data/request-logs/** —— 本机请求归档，搬来搬去只撑大桶。
func (e *Engine) isCandidate(relPath string) bool {
	p := strings.ReplaceAll(relPath, `\`, "/")
	if p == "config.json" || strings.HasPrefix(p, "data/request-logs/") {
		return false
	}
	return PathAllowed(p, e.cfg.IncludeData)
}

// ---------------------------------------------------------------------------
// 执行
// ---------------------------------------------------------------------------

// Apply 执行计划。dryRun 为真时只回填"将会做什么"，不落盘也不上传。
func (e *Engine) Apply(ctx context.Context, p *Plan, dryRun bool) (*Result, error) {
	res := &Result{Mode: p.Mode}

	// 批量上传：relay 通道一次 HTTP 调用推完，省掉冷启动后的多次往返。
	var batch []PushFile
	batchable := false
	if !dryRun {
		if _, ok := e.writer.(interface {
			PutBatch(context.Context, []PushFile, bool) error
		}); ok {
			batchable = true
		}
	}

	for i := range p.Actions {
		a := &p.Actions[i]
		switch a.Op {
		case OpNoop:
			res.Skipped++
			continue

		case OpDownload:
			if dryRun {
				a.Applied = true
				res.Downloaded++
				continue
			}
			if err := e.applyDownload(ctx, a, res); err != nil {
				a.Err = err.Error()
				res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err))
				res.Failed++
				continue
			}
			a.Applied = true
			res.Downloaded++

		case OpUpload:
			if e.writer == nil {
				err := fmt.Errorf("没有可用的写入通道（配置 sync.writer / relay_url，或安装 hf CLI）")
				a.Err = err.Error()
				res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err))
				res.Failed++
				continue
			}
			if dryRun {
				a.Applied = true
				res.Uploaded++
				continue
			}
			if batchable {
				batch = append(batch, PushFile{Path: a.Path, Content: a.Local.Content, Mtime: a.Local.Mtime})
				a.Applied = true
				res.Uploaded++
				continue
			}
			if err := e.writer.Put(ctx, a.Path, a.Local.Content); err != nil {
				a.Err = err.Error()
				res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err))
				res.Failed++
				continue
			}
			a.Applied = true
			res.Uploaded++
		}
	}

	if len(batch) > 0 {
		type batchWriter interface {
			PutBatch(context.Context, []PushFile, bool) error
		}
		if bw, ok := e.writer.(batchWriter); ok {
			if err := bw.PutBatch(ctx, batch, true); err != nil {
				// 批量失败：把已计入的动作全部回落为失败，避免"报告成功其实没上去"。
				res.Errors = append(res.Errors, err.Error())
				failed := map[string]bool{}
				for _, bf := range batch {
					failed[bf.Path] = true
				}
				for i := range p.Actions {
					if p.Actions[i].Op == OpUpload && failed[p.Actions[i].Path] {
						p.Actions[i].Applied = false
						p.Actions[i].Err = err.Error()
						res.Uploaded--
						res.Failed++
					}
				}
			}
		}
	}

	res.Actions = p.Actions
	return res, nil
}

// applyDownload 把远端内容落到本地。
func (e *Engine) applyDownload(ctx context.Context, a *Action, res *Result) error {
	content := a.Remote.Content
	if content == nil {
		raw, err := e.bucket.Get(ctx, a.Path)
		if err != nil {
			return err
		}
		content = raw
	}
	fp, err := e.localPath(a.Path)
	if err != nil {
		return err
	}
	if err := writeLocalAtomic(fp, content, a.Remote.Mtime); err != nil {
		return err
	}
	_ = res
	return nil
}

// Sync 是 Plan + Apply 的便捷组合。
func (e *Engine) Sync(ctx context.Context, mode Mode, dryRun bool) (*Plan, *Result, error) {
	p, err := e.Plan(ctx, mode)
	if err != nil {
		return nil, nil, err
	}
	res, err := e.Apply(ctx, p, dryRun)
	if err != nil {
		return p, res, err
	}
	return p, res, nil
}