RawMark/internal/watch/watch.go

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

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"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
	info  os.FileInfo
}

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

// Start 开始监听。onScan 在首次扫描及之后每次有变化的扫描后调用，参数为全部笔记的路径 -> rev（用于更新搜索索引）
func Start(ctx context.Context, v *vault.Vault, hub *events.Hub, onScan func(notes map[string]string)) 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{}, onScan: onScan}
	snap, err := w.scan()
	if err != nil {
		fw.Close()
		return err
	}
	w.snap = snap
	w.onScan(notes(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
	if len(evs) > 0 {
		w.onScan(notes(next))
	}
	w.hub.Broadcast(evs...)
}

func notes(snap map[string]fileState) map[string]string {
	out := make(map[string]string, len(snap))
	for p, st := range snap {
		if !st.dir && vault.IsNote(p) {
			visible := true
			for _, segment := range strings.Split(p, "/") {
				if strings.HasSuffix(segment, vault.AssetsExt) {
					visible = false
					break
				}
			}
			if visible {
				out[p] = st.rev
			}
		}
	}
	return out
}

// scan 读取当前目录树；内容版本只对新出现或大小/修改时间变化的文件重新计算
func (w *Watcher) scan() (map[string]fileState, error) {
	entries, err := w.v.TreeAt("")
	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.Info.ModTime().UnixNano(), info: e.Info}
		abs := filepath.Join(w.v.Root(), filepath.FromSlash(e.Path))
		if e.Dir {
			dirs[abs] = true
		} else if vault.IsNote(e.Path) {
			if old, ok := w.snap[e.Path]; ok && !old.dir && old.size == st.size && old.mtime == st.mtime && os.SameFile(old.info, st.info) {
				st.rev = old.rev
			} else {
				data, err := os.ReadFile(abs)
				if errors.Is(err, fs.ErrNotExist) {
					continue // 扫描期间被删除，下一轮会处理
				}
				if err != nil {
					return nil, fmt.Errorf("读取笔记 %s 失败: %w", e.Path, err)
				}
				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]fileState{}
	for p, old := range prev {
		if _, ok := next[p]; ok {
			continue
		}
		structural = true
		removed[p] = old
	}
	candidates := map[string]string{} // 新路径 -> 唯一的原路径
	destinations := map[string]int{}
	for p, cur := range next {
		old, existed := prev[p]
		switch {
		case !existed:
			structural = true
			from := ""
			matches := 0
			for oldPath, old := range removed {
				if os.SameFile(old.info, cur.info) {
					from = oldPath
					matches++
				}
			}
			if matches == 1 {
				candidates[p] = from
				destinations[from]++
			}
		case old.dir != cur.dir:
			structural = true
		case !cur.dir && cur.rev == "" && (old.size != cur.size || old.mtime != cur.mtime || !os.SameFile(old.info, cur.info)):
			// 图片没有正文版本号；替换后仍需刷新列表元数据与预览。
			if vault.IsImage(p) {
				structural = true
			}
		case !cur.dir && old.rev != cur.rev:
			evs = append(evs, events.Event{Type: "changed", Path: p, Rev: cur.rev})
		}
	}
	var moves []events.Event
	for to, from := range candidates {
		if destinations[from] == 1 {
			delete(removed, from)
			moves = append(moves, events.Event{Type: "moved", From: from, Path: to, Rev: next[to].rev})
		}
	}
	sort.Slice(moves, func(i, j int) bool { return len(moves[i].From) < len(moves[j].From) })
	var emitted []events.Event
	for _, move := range moves {
		for _, parent := range emitted {
			if strings.HasPrefix(move.From, parent.From+"/") {
				move.From = parent.Path + strings.TrimPrefix(move.From, parent.From)
			}
		}
		// 先应用父目录映射；子项另有改名时，从映射后的路径继续移动。
		if move.From != move.Path {
			emitted = append(emitted, move)
		}
	}
	evs = append(emitted, evs...)
	for p := range removed {
		evs = append(evs, events.Event{Type: "deleted", Path: p})
	}
	if structural {
		evs = append(evs, events.Event{Type: "tree"})
	}
	return evs
}