package indexer import ( "bufio" "context" "fmt" "os" "path/filepath" "strings" "sync" "time" "github.com/fsnotify/fsnotify" "github.com/zzet/gortex/internal/gitcmd" "go.uber.org/zap" ) // GitWatcher observes `.git/HEAD` (and the target ref) for a single // tracked repository. When the resolved commit moves, it computes the // diff between the old and new commits and dispatches EvictFile / // IndexFileNoResolve per changed path, running the resolver once at // the end. This is the branch-switch / rebase / reset path: the // regular fsnotify per-file watcher sees 500 Remove+Create events for // a checkout and walks each through per-file resolve + search, which // is measurably slow and incorrect for renames (git sees them as // atomic; fsnotify sees unrelated remove+create pairs). // // Gitwatcher complements the file watcher — it does not replace it. // File edits outside git operations (save buffer, `git stash`, // `git reset --mixed` leaving the working tree dirty) still fire // normal fsnotify events and flow through the per-file path. type GitWatcher struct { repoPath string indexer *Indexer logger *zap.Logger fsw *fsnotify.Watcher debounce time.Duration done chan struct{} stopped chan struct{} mu sync.Mutex lastSHA string fireTimer *time.Timer loopStarted bool stopCalled bool // reconciling single-flights reconcile: a ref event landing while // one is in flight sets rerun instead of spawning a second // concurrent reconcile from the same stale base (each AfterFunc // fires on its own goroutine, and lastSHA only advances at the // END of a reconcile — overlapping runs would diff and apply the // same range twice and race applyChanges/ResolveAll). reconciling bool rerun bool // drained is a test hook that fires after a reconcile completes // with the number of files patched. nil in production. drained func(int) } // NewGitWatcher creates a watcher for repoPath/.git/HEAD. repoPath is // the absolute path to the worktree root; the .git dir is discovered // by looking at the HEAD file (handles worktrees, submodules via the // `gitdir:` indirection). func NewGitWatcher(repoPath string, idx *Indexer, logger *zap.Logger) (*GitWatcher, error) { absRepo, err := filepath.Abs(repoPath) if err != nil { return nil, err } fsw, err := fsnotify.NewWatcher() if err != nil { return nil, err } return &GitWatcher{ repoPath: absRepo, indexer: idx, logger: logger, fsw: fsw, debounce: 300 * time.Millisecond, done: make(chan struct{}), stopped: make(chan struct{}), }, nil } // Start sets up fsnotify watches on the repo's git control files and // launches the event-processing goroutine. Safe to call once per // GitWatcher instance. Returns an error (and does not launch the loop) // when the repo has no .git directory — Stop remains safe to call. func (gw *GitWatcher) Start() error { gitDir, err := resolveGitDir(gw.repoPath) if err != nil { return fmt.Errorf("resolve .git dir for %s: %w", gw.repoPath, err) } // HEAD + refs/heads/ cover branch switches and same-branch // commits; packed-refs covers the gc case where loose refs get // packed and moved out of refs/heads. Missing files are not fatal // — a fresh repo may not have packed-refs yet. for _, rel := range []string{"HEAD", "packed-refs", "refs/heads"} { path := filepath.Join(gitDir, rel) if _, err := os.Stat(path); err != nil { continue } if err := gw.fsw.Add(path); err != nil { gw.logger.Warn("git-watcher: failed to watch", zap.String("path", path), zap.Error(err)) } } gw.lastSHA, _ = gw.currentSHA(context.Background()) gw.mu.Lock() gw.loopStarted = true gw.mu.Unlock() go gw.loop() return nil } // Stop halts the watcher. Idempotent — safe whether Start succeeded, // failed, or was never called. We only block on `stopped` when the // loop goroutine is actually running; otherwise Stop would deadlock // on a channel nobody's going to close. func (gw *GitWatcher) Stop() error { gw.mu.Lock() started := gw.loopStarted already := gw.stopCalled gw.stopCalled = true gw.mu.Unlock() if already { return nil } close(gw.done) _ = gw.fsw.Close() if started { <-gw.stopped } return nil } func (gw *GitWatcher) loop() { defer close(gw.stopped) for { select { case <-gw.done: return case event, ok := <-gw.fsw.Events: if !ok { return } gw.scheduleReconcile(event.Name) case err, ok := <-gw.fsw.Errors: if !ok { return } gw.logger.Warn("git-watcher: fsnotify error", zap.Error(err)) } } } // scheduleReconcile coalesces bursts of ref-file events (a branch // switch touches HEAD, refs/heads/, and often packed-refs in // rapid succession) into a single reconcile. Resets the debounce // timer on every event and lets the last one win. func (gw *GitWatcher) scheduleReconcile(trigger string) { gw.mu.Lock() defer gw.mu.Unlock() if gw.fireTimer != nil { gw.fireTimer.Stop() } gw.fireTimer = time.AfterFunc(gw.debounce, func() { gw.reconcile(trigger) }) } // gitWatcherScopedResolveMaxFiles caps how many changed files a ref // reconcile resolves through the scoped incremental path before it // falls back to a single whole-graph ResolveAll. Below the cap a // commit-sized change resolves only its own files (and their incoming // refs) -- the same path edit_file uses; above it (branch switches, // large merges) one amortised whole-graph pass beats many per-file // resolves. A package var so tests can lower it to exercise the // fallback. var gitWatcherScopedResolveMaxFiles = 100 // changedAbsPaths collects the absolute on-disk paths a diff touched -- // the new path for every status plus the old path of a rename -- // deduplicated. Disk existence is resolved later by // IncrementalReindexPaths: a path present on disk is reindexed, one // absent is evicted, so a delete needs no special-casing here. func (gw *GitWatcher) changedAbsPaths(changes []gitChange) []string { seen := make(map[string]struct{}, len(changes)+4) out := make([]string, 0, len(changes)+4) add := func(rel string) { if rel == "" { return } abs := filepath.Join(gw.repoPath, rel) if _, ok := seen[abs]; ok { return } seen[abs] = struct{}{} out = append(out, abs) } for _, c := range changes { add(c.Path) add(c.OldPath) } return out } // reconcile reads the current HEAD SHA, diffs against the previously // seen SHA via `git diff --name-status`, and dispatches // the changed paths to a scoped reindex + resolve (whole-graph only // above gitWatcherScopedResolveMaxFiles). Silently no-ops when HEAD hasn't moved — branches can // touch packed-refs without the resolved commit actually changing. func (gw *GitWatcher) reconcile(trigger string) { // Single-flight: one reconcile at a time. A ref change observed // mid-flight is coalesced into exactly one follow-up run that // diffs the (now advanced) lastSHA against wherever HEAD ended // up — so we always converge on the latest state without ever // running two whole-graph passes concurrently. gw.mu.Lock() if gw.reconciling { gw.rerun = true gw.mu.Unlock() return } gw.reconciling = true gw.mu.Unlock() defer func() { gw.mu.Lock() gw.reconciling = false again := gw.rerun && !gw.stopCalled gw.rerun = false gw.mu.Unlock() if again { go gw.reconcile("coalesced") } }() if gw.rebaseInProgress() { // Defer until the rebase lands. The final rebase state // either updates HEAD (triggering another fsnotify fire) or // leaves the branch where it was — in both cases we'll pick // up the right end state on the next event. gw.logger.Debug("git-watcher: rebase in progress, deferring", zap.String("trigger", trigger)) return } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() newSHA, err := gw.currentSHA(ctx) if err != nil || newSHA == "" { return } gw.mu.Lock() oldSHA := gw.lastSHA gw.mu.Unlock() if oldSHA == newSHA { return } // First observation (no prior SHA) — just record it without // diffing. The caller warmed up with a full index already. if oldSHA == "" { gw.mu.Lock() gw.lastSHA = newSHA gw.mu.Unlock() return } start := time.Now() changes, err := gw.diffNameStatus(ctx, oldSHA, newSHA) if err != nil { gw.logger.Warn("git-watcher: diff failed", zap.String("from", oldSHA), zap.String("to", newSHA), zap.Error(err)) return } // Resolve the batch. A commit-sized change set resolves only the // files it touched (and their incoming refs) through the scoped // incremental path -- the same one edit_file / fsnotify saves use -- // instead of a whole-graph ResolveAll that costs O(graph) no matter // how few files moved. A large change set (branch switch, big merge) // still takes one amortised whole-graph pass, where skipping the // per-file resolver in applyChanges is the better trade. changedPaths := gw.changedAbsPaths(changes) var patched int switch { case len(changedPaths) > gitWatcherScopedResolveMaxFiles: patched = gw.applyChanges(changes) gw.indexer.ResolveAll() default: res, rerr := gw.indexer.IncrementalReindexPaths(gw.repoPath, changedPaths) if rerr != nil { gw.logger.Warn("git-watcher: scoped reindex failed, falling back to whole-graph resolve", zap.String("trigger", trigger), zap.Error(rerr)) patched = gw.applyChanges(changes) gw.indexer.ResolveAll() } else { patched = res.StaleFileCount + res.DeletedFileCount } } gw.mu.Lock() gw.lastSHA = newSHA drained := gw.drained gw.mu.Unlock() // Re-stamp the on-disk freshness row at the new HEAD. The graph now // reflects this commit, but the persisted repo_index_state row is // otherwise written only by a full (re)index — leave it and // `gortex repos` keeps reporting the repo stale after every commit. gw.indexer.reconcileRepoIndexState(gw.repoPath) gw.logger.Info("git-watcher: reconciled ref change", zap.String("from", oldSHA[:min(len(oldSHA), 12)]), zap.String("to", newSHA[:min(len(newSHA), 12)]), zap.Int("paths", patched), zap.Duration("elapsed", time.Since(start))) if drained != nil { drained(patched) } } // gitChange describes one entry in `git diff --name-status` output. // Status is a single char (A/M/D/T) or R/C with a similarity score. type gitChange struct { Status byte Path string OldPath string // only populated for R/C } // applyChanges dispatches the decoded diff to the indexer. Returns // the count of files touched (index + evict). func (gw *GitWatcher) applyChanges(changes []gitChange) int { n := 0 for _, c := range changes { switch c.Status { case 'A', 'M', 'T': abs := filepath.Join(gw.repoPath, c.Path) if _, err := os.Stat(abs); err != nil { // Git says it's here, working tree disagrees — // partial / sparse checkout. Skip without error. continue } if err := gw.indexer.IndexFileNoResolve(abs); err != nil { gw.logger.Warn("git-watcher: index failed", zap.String("path", c.Path), zap.Error(err)) continue } n++ case 'D': // A 'D' in the diff means the file left git tracking, not // necessarily the disk. A file un-tracked (git rm --cached) // or absent from the new branch but kept on disk is // "untracked but visible" and must stay indexed. Only a // file genuinely gone from disk is evicted; one still on // disk is re-indexed unless it is now excluded. abs := filepath.Join(gw.repoPath, c.Path) _, statErr := os.Stat(abs) switch { case statErr != nil: gw.indexer.EvictFile(abs) case gw.indexer.shouldExclude(abs, gw.repoPath, false): gw.indexer.EvictFile(abs) default: if err := gw.indexer.IndexFileNoResolve(abs); err != nil { gw.logger.Warn("git-watcher: re-index of now-untracked file failed", zap.String("path", c.Path), zap.Error(err)) } } n++ case 'R': // Rename: evict the old path, index the new one. Git // gives us the pair atomically — fsnotify would have // surfaced this as a remove + create with no linkage. if c.OldPath != "" { gw.indexer.EvictFile(filepath.Join(gw.repoPath, c.OldPath)) } abs := filepath.Join(gw.repoPath, c.Path) if _, err := os.Stat(abs); err == nil { if err := gw.indexer.IndexFileNoResolve(abs); err != nil { gw.logger.Warn("git-watcher: index failed (rename)", zap.String("path", c.Path), zap.Error(err)) continue } } n++ case 'C': // Copy: the source is unchanged; we only need to index // the new file. abs := filepath.Join(gw.repoPath, c.Path) if _, err := os.Stat(abs); err == nil { if err := gw.indexer.IndexFileNoResolve(abs); err != nil { gw.logger.Warn("git-watcher: index failed (copy)", zap.String("path", c.Path), zap.Error(err)) continue } } n++ } } return n } // currentSHA returns the resolved commit SHA of HEAD. Shells out to // git rather than parsing .git/HEAD directly so symbolic refs, // packed-refs, and worktree indirection all work without us // reimplementing git's ref resolution. func (gw *GitWatcher) currentSHA(ctx context.Context) (string, error) { return gitcmd.Output(ctx, gw.repoPath, "rev-parse", "HEAD") } // diffNameStatus shells out to `git diff --name-status -M -C oldSHA..newSHA` // and decodes the output into gitChange records. -M enables rename // detection, -C enables copy detection. func (gw *GitWatcher) diffNameStatus(ctx context.Context, oldSHA, newSHA string) ([]gitChange, error) { out, err := gitcmd.Run(ctx, gw.repoPath, "diff", "--name-status", "-M", "-C", "-z", oldSHA, newSHA) if err != nil { return nil, err } return parseDiffNameStatus(out), nil } // parseDiffNameStatus decodes the `-z` NUL-delimited output of // `git diff --name-status`. Each entry is: STATUS\0path[\0newpath]. // Rename (R) and copy (C) statuses carry a similarity score appended // to the letter (e.g., "R100") and come with two paths separated by // a NUL. Everything else is a single path. func parseDiffNameStatus(out []byte) []gitChange { var changes []gitChange // bufio.Scanner with a NUL split function gives us one token per // field; we consume them in pairs (or triples for R/C). scanner := bufio.NewScanner(strings.NewReader(string(out))) scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) scanner.Split(scanNul) for scanner.Scan() { status := scanner.Text() if status == "" { continue } letter := status[0] if !scanner.Scan() { break } path := scanner.Text() c := gitChange{Status: letter, Path: path} if letter == 'R' || letter == 'C' { if !scanner.Scan() { break } c.OldPath = path c.Path = scanner.Text() } changes = append(changes, c) } return changes } // scanNul is a bufio.SplitFunc that tokenises on NUL bytes. Used for // `git diff -z` output where paths may contain whitespace. func scanNul(data []byte, atEOF bool) (advance int, token []byte, err error) { if atEOF && len(data) == 0 { return 0, nil, nil } for i, b := range data { if b == 0 { return i + 1, data[:i], nil } } if atEOF { return len(data), data, nil } return 0, nil, nil } // rebaseInProgress reports whether a rebase, merge, or cherry-pick is // currently in flight — any of which touch HEAD/refs rapidly and // produce intermediate states we don't want to reconcile against. // Detection is by sentinel file presence in .git (the canonical way // other git tooling does it). func (gw *GitWatcher) rebaseInProgress() bool { gitDir, err := resolveGitDir(gw.repoPath) if err != nil { return false } for _, sentinel := range []string{"rebase-merge", "rebase-apply", "MERGE_HEAD", "CHERRY_PICK_HEAD", "BISECT_LOG"} { if _, err := os.Stat(filepath.Join(gitDir, sentinel)); err == nil { return true } } return false } // resolveGitDir returns the absolute path to the .git directory for a // worktree. Handles the worktree / submodule case where .git is a // file containing `gitdir: ` instead of a directory. func resolveGitDir(repoPath string) (string, error) { candidate := filepath.Join(repoPath, ".git") info, err := os.Stat(candidate) if err != nil { return "", err } if info.IsDir() { return candidate, nil } // .git is a file pointing at the real gitdir — common for // worktrees (under modules/) and submodules. content, err := os.ReadFile(candidate) if err != nil { return "", err } line := strings.TrimSpace(string(content)) const prefix = "gitdir:" if !strings.HasPrefix(line, prefix) { return "", fmt.Errorf("malformed .git file: %s", candidate) } dir := strings.TrimSpace(strings.TrimPrefix(line, prefix)) if !filepath.IsAbs(dir) { dir = filepath.Join(repoPath, dir) } return dir, nil }