RawMark/internal/watch/watch.go

// Package watch 监听仓库目录的变化（包括 RawMark 自身的写入和外部编辑器/同步工具的修改），
// 与上一次快照比较后把差异作为事件广播。所有变更都经由文件系统这一条路径，客户端据 rev 识别自己的写入。
package watch

import (
	"context"
	"log/slog"
	"os"
	"path/filepath"
	"time"

	"github.com/fsnotify/fsnotify"

	"rawmark/internal/events"
	"rawmark/internal/vault"
)

const (
	debounce = 100 * time.Millisecond
	// inotify 在 Docker Desktop 的绑定挂载、NFS 等环境下不可靠，定期全量扫描兜底
	rescanInterval = 5 * time.Minute
)

type fileState struct {
	dir   bool
	size  int64
	mtime int64
	rev   string
}

type Watcher struct {
	v       *vault.Vault
	hub     *events.Hub
	fw      *fsnotify.Watcher
	snap    map[string]fileState
	watched map[string]bool
}

func Start(ctx context.Context, v *vault.Vault, hub *events.Hub) error {
	fw, err := fsnotify.NewWatcher()
	if err != nil {
		return err
	}
	w := &Watcher{v: v, hub: hub, fw: fw, snap: map[string]fileState{}, watched: map[string]bool{}}
	snap, err := w.scan()
	if err != nil {
		fw.Close()
		return err
	}
	w.snap = snap
	go w.loop(ctx)
	return nil
}

func (w *Watcher) loop(ctx context.Context) {
	defer w.fw.Close()
	timer := time.NewTimer(time.Hour)
	timer.Stop()
	ticker := time.NewTicker(rescanInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case _, ok := <-w.fw.Events:
			if !ok {
				return
			}
			timer.Reset(debounce)
		case err, ok := <-w.fw.Errors:
			if !ok {
				return
			}
			// 事件队列溢出等情况会丢事件，立即全量扫描补齐
			slog.Warn("文件监听出错，执行全量扫描", "err", err)
			timer.Reset(debounce)
		case <-timer.C:
			w.rescan()
		case <-ticker.C:
			w.rescan()
		}
	}
}

func (w *Watcher) rescan() {
	next, err := w.scan()
	if err != nil {
		slog.Error("扫描仓库失败", "err", err)
		return
	}
	evs := diff(w.snap, next)
	w.snap = next
	w.hub.Broadcast(evs...)
}

// scan 读取当前目录树；内容版本只对新出现或大小/修改时间变化的文件重新计算
func (w *Watcher) scan() (map[string]fileState, error) {
	entries, err := w.v.Tree()
	if err != nil {
		return nil, err
	}
	next := make(map[string]fileState, len(entries))
	dirs := map[string]bool{w.v.Root(): true}
	for _, e := range entries {
		st := fileState{dir: e.Dir, size: e.Size, mtime: e.ModTime}
		abs := filepath.Join(w.v.Root(), filepath.FromSlash(e.Path))
		if e.Dir {
			dirs[abs] = true
		} else if old, ok := w.snap[e.Path]; ok && !old.dir && old.size == st.size && old.mtime == st.mtime {
			st.rev = old.rev
		} else {
			data, err := os.ReadFile(abs)
			if err != nil {
				continue // 扫描期间被删除，下一轮会处理
			}
			st.rev = vault.Rev(data)
		}
		next[e.Path] = st
	}
	for d := range dirs {
		if !w.watched[d] {
			if err := w.fw.Add(d); err != nil {
				slog.Warn("添加目录监听失败", "dir", d, "err", err)
				continue
			}
			w.watched[d] = true
		}
	}
	for d := range w.watched {
		if !dirs[d] {
			delete(w.watched, d) // 目录已删除，inotify 会自动移除监听
		}
	}
	return next, nil
}

// diff 比较两次快照：内容相同的"删除 + 新增"视为移动
func diff(prev, next map[string]fileState) []events.Event {
	var evs []events.Event
	structural := false
	removed := map[string][]string{} // rev -> 被删除的路径（内容相同的文件可能有多个）
	for p, old := range prev {
		if _, ok := next[p]; ok {
			continue
		}
		structural = true
		if !old.dir {
			removed[old.rev] = append(removed[old.rev], p)
		}
	}
	for p, cur := range next {
		old, existed := prev[p]
		switch {
		case !existed:
			structural = true
			if froms := removed[cur.rev]; len(froms) > 0 && !cur.dir {
				removed[cur.rev] = froms[1:]
				evs = append(evs, events.Event{Type: "moved", From: froms[0], Path: p, Rev: cur.rev})
			}
		case old.dir != cur.dir:
			structural = true
		case !cur.dir && old.rev != cur.rev:
			evs = append(evs, events.Event{Type: "changed", Path: p, Rev: cur.rev})
		}
	}
	for _, paths := range removed {
		for _, p := range paths {
			evs = append(evs, events.Event{Type: "deleted", Path: p})
		}
	}
	if structural {
		evs = append(evs, events.Event{Type: "tree"})
	}
	return evs
}