kenn-io--agentsview/internal/sync/watcher.go

package sync

import (
	"errors"
	"fmt"
	"io/fs"
	"log"
	"math"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/fsnotify/fsnotify"
)

type RecursiveWatchResult struct {
	Watched             int
	Unwatched           int
	Err                 error
	BudgetExhausted     bool
	ResourceExhausted   bool
	ResourceExhaustedAt string
}

const (
	// A callback can hold one batch while the event loop accumulates the next.
	// Bounding each batch keeps watcher backpressure below two bounded path
	// sets instead of allowing a slow sync to retain an unbounded event storm.
	defaultWatchBatchMaxEntries   = 8192
	defaultWatchBatchMaxPathBytes = 2 << 20
)

// WatchBatch describes one serialized watcher callback. FullSync is an
// explicit overflow signal: Paths is empty and the consumer must rescan all
// configured sources so coalescing never silently drops a changed path.
type WatchBatch struct {
	Paths    []string
	FullSync bool
}

// pendingWatchBatch bounds retained strings by both unique path count and the
// sum of their byte lengths. Map overhead is bounded separately by the entry
// limit. Once either limit would be exceeded, individual paths are discarded
// in favor of one full-sync marker until Take resets the accumulator.
type pendingWatchBatch struct {
	paths        map[string]struct{}
	pathBytes    int
	maxEntries   int
	maxPathBytes int
	fullSync     bool
}

func newPendingWatchBatch(maxEntries, maxPathBytes int) *pendingWatchBatch {
	return &pendingWatchBatch{
		paths:        make(map[string]struct{}),
		maxEntries:   maxEntries,
		maxPathBytes: maxPathBytes,
	}
}

func (p *pendingWatchBatch) Empty() bool {
	return !p.fullSync && len(p.paths) == 0
}

func (p *pendingWatchBatch) Add(path string) {
	if p.fullSync {
		return
	}
	if _, exists := p.paths[path]; exists {
		return
	}
	if len(p.paths)+1 > p.maxEntries ||
		p.pathBytes+len(path) > p.maxPathBytes {
		clear(p.paths)
		p.pathBytes = 0
		p.fullSync = true
		return
	}
	p.paths[path] = struct{}{}
	p.pathBytes += len(path)
}

func (p *pendingWatchBatch) Take() (WatchBatch, bool) {
	if p.Empty() {
		return WatchBatch{}, false
	}
	if p.fullSync {
		p.fullSync = false
		return WatchBatch{FullSync: true}, true
	}

	paths := make([]string, 0, len(p.paths))
	for path := range p.paths {
		paths = append(paths, path)
	}
	slices.Sort(paths)
	clear(p.paths)
	p.pathBytes = 0
	return WatchBatch{Paths: paths}, true
}

// Watcher uses fsnotify to watch session directories for changes and triggers
// serialized callbacks with short-burst batching and a dispatch floor.
type Watcher struct {
	onChange     func(batch WatchBatch)
	watcher      *fsnotify.Watcher
	batchDelay   time.Duration
	minInterval  time.Duration
	maxEntries   int
	maxPathBytes int
	excludes     []string
	roots        []string
	shallow      []string
	rootsMu      sync.RWMutex
	dispatchMu   sync.Mutex
	stopping     bool
	stop         chan struct{}
	done         chan struct{}
	stopOnce     sync.Once
}

// NewWatcher creates a file watcher that uses delay for both batching and the
// minimum interval between callbacks.
func NewWatcher(delay time.Duration, onChange func(batch WatchBatch), excludes []string) (*Watcher, error) {
	return NewWatcherWithInterval(delay, delay, onChange, excludes)
}

// NewWatcherWithInterval creates a file watcher with separate batching and
// minimum callback intervals.
func NewWatcherWithInterval(
	batchDelay, minInterval time.Duration,
	onChange func(batch WatchBatch), excludes []string,
) (*Watcher, error) {
	return newWatcherWithLimits(
		batchDelay,
		minInterval,
		onChange,
		excludes,
		defaultWatchBatchMaxEntries,
		defaultWatchBatchMaxPathBytes,
	)
}

func newWatcherWithLimits(
	batchDelay, minInterval time.Duration,
	onChange func(batch WatchBatch), excludes []string,
	maxEntries, maxPathBytes int,
) (*Watcher, error) {
	if onChange == nil {
		return nil, fmt.Errorf("onChange callback is nil: %w", os.ErrInvalid)
	}

	fsw, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}

	w := &Watcher{
		onChange:     onChange,
		watcher:      fsw,
		batchDelay:   batchDelay,
		minInterval:  minInterval,
		maxEntries:   maxEntries,
		maxPathBytes: maxPathBytes,
		excludes:     normalizeExcludePatterns(excludes),
		stop:         make(chan struct{}),
		done:         make(chan struct{}),
	}
	return w, nil
}

// WatchRecursive walks a directory tree and adds all
// subdirectories to the watch list. Returns the number
// of directories watched and unwatched (failed to add).
func (w *Watcher) WatchRecursive(root string) (watched int, unwatched int, err error) {
	result := w.WatchRecursiveBudgeted(root, math.MaxInt)
	return result.Watched, result.Unwatched, result.Err
}

// WatchRecursiveBudgeted walks a directory tree and adds at most
// budget subdirectories to the watch list. The walk stops as soon
// as the budget is exhausted or fsnotify reports resource
// exhaustion, so the caller can degrade the rest of the tree to
// polling without continuing to traverse it.
func (w *Watcher) WatchRecursiveBudgeted(root string, budget int) RecursiveWatchResult {
	var result RecursiveWatchResult
	root = filepath.Clean(root)
	w.addRoot(root)

	remaining := budget
	result.Err = filepath.WalkDir(root,
		func(path string, d fs.DirEntry, err error) error {
			if err != nil {
				return nil // skip inaccessible dirs
			}
			if !d.IsDir() {
				return nil
			}
			// Skip entire excluded subtrees, but always keep the root.
			if path != root && w.shouldExcludeForRoot(path, root) {
				return filepath.SkipDir
			}
			if remaining <= 0 {
				result.BudgetExhausted = true
				return filepath.SkipAll
			}
			if addErr := w.watcher.Add(path); addErr != nil {
				result.Unwatched++
				if isWatchResourceExhaustion(addErr) {
					result.ResourceExhausted = true
					result.ResourceExhaustedAt = path
					return filepath.SkipAll
				}
				return nil
			}
			remaining--
			result.Watched++
			return nil
		})
	if errors.Is(result.Err, filepath.SkipAll) {
		result.Err = nil
	}
	return result
}

func isWatchResourceExhaustion(err error) bool {
	return errors.Is(err, syscall.EMFILE) || errors.Is(err, syscall.ENOSPC)
}

// WatchShallow adds only the root directory to the watch list,
// without recursing into subdirectories. Use this for directories
// with many subdirectories where periodic sync handles changes.
// Returns true if the directory was successfully watched.
func (w *Watcher) WatchShallow(root string) bool {
	root = filepath.Clean(root)
	w.addRoot(root)
	w.addShallowRoot(root)
	return w.watcher.Add(root) == nil
}

// Start begins processing file events in a goroutine.
func (w *Watcher) Start() {
	go w.loop()
}

// Stop stops the watcher and waits for it to finish.
func (w *Watcher) Stop() {
	w.stopOnce.Do(func() {
		w.dispatchMu.Lock()
		w.stopping = true
		close(w.stop)
		w.dispatchMu.Unlock()
		<-w.done
		w.watcher.Close()
	})
}

func (w *Watcher) loop() {
	batches := make(chan WatchBatch)
	callbackDone := make(chan time.Time, 1)
	var worker sync.WaitGroup
	worker.Go(func() {
		for batch := range batches {
			if batch.FullSync {
				log.Printf("watcher: pending path limit exceeded, triggering full sync")
			} else {
				log.Printf("watcher: %d file(s) changed, triggering sync", len(batch.Paths))
			}
			startedAt := time.Now()
			w.onChange(batch)
			callbackDone <- startedAt
		}
	})

	pending := newPendingWatchBatch(w.maxEntries, w.maxPathBytes)
	var firstPendingAt time.Time
	var lastDispatch time.Time
	var timer *time.Timer
	var timerC <-chan time.Time
	callbackBusy := false

	stopTimer := func() {
		if timer != nil {
			timer.Stop()
		}
		timer = nil
		timerC = nil
	}
	defer func() {
		stopTimer()
		close(batches)
		worker.Wait()
		close(w.done)
	}()

	schedule := func() {
		if callbackBusy || pending.Empty() || timerC != nil {
			return
		}
		deadline := firstPendingAt.Add(w.batchDelay)
		if !lastDispatch.IsZero() {
			floor := lastDispatch.Add(w.minInterval)
			if floor.After(deadline) {
				deadline = floor
			}
		}
		timer = time.NewTimer(time.Until(deadline))
		timerC = timer.C
	}

	dispatch := func() bool {
		if callbackBusy || pending.Empty() {
			return true
		}
		w.dispatchMu.Lock()
		defer w.dispatchMu.Unlock()
		if w.stopping {
			return false
		}
		batch, ok := pending.Take()
		if !ok {
			return true
		}
		firstPendingAt = time.Time{}
		callbackBusy = true
		batches <- batch
		return true
	}

	for {
		// Give shutdown priority over ready event and timer channels. Pending
		// changes are recovered by the next startup sync.
		select {
		case <-w.stop:
			return
		default:
		}
		select {
		case <-w.stop:
			return

		case event, ok := <-w.watcher.Events:
			if !ok {
				return
			}
			path, relevant := w.handleEvent(event)
			if !relevant {
				continue
			}
			if pending.Empty() {
				firstPendingAt = time.Now()
			}
			pending.Add(path)
			schedule()

		case err, ok := <-w.watcher.Errors:
			if !ok {
				return
			}
			log.Printf("watcher error: %v", err)

		case <-timerC:
			timer = nil
			timerC = nil
			select {
			case <-w.stop:
				return
			default:
			}
			if !dispatch() {
				return
			}

		case startedAt := <-callbackDone:
			lastDispatch = startedAt
			callbackBusy = false
			schedule()
		}
	}
}

// handleEvent processes a single fsnotify event, auto-watching
// newly created directories and returning relevant changed paths.
func (w *Watcher) handleEvent(event fsnotify.Event) (string, bool) {
	if event.Op&(fsnotify.Write|
		fsnotify.Create|
		fsnotify.Remove|
		fsnotify.Rename) == 0 {
		return "", false
	}

	if event.Op&fsnotify.Create != 0 {
		isDir, excluded := w.watchIfDir(event.Name)
		if isDir && excluded {
			return "", false
		}
	}

	return filepath.Clean(event.Name), true
}

// watchIfDir adds a path to the watch list if it is a directory.
// Returns whether path is a directory and whether it was excluded.
func (w *Watcher) watchIfDir(path string) (isDir bool, excluded bool) {
	info, err := os.Stat(path)
	if err != nil || !info.IsDir() {
		return false, false
	}
	if w.isUnderShallowRoot(path) {
		return true, false
	}
	if w.shouldExclude(path) {
		return true, true
	}
	_ = w.watcher.Add(path)
	return true, false
}

func normalizeExcludePatterns(patterns []string) []string {
	if len(patterns) == 0 {
		return nil
	}
	out := make([]string, 0, len(patterns))
	for _, p := range patterns {
		p = strings.TrimSpace(filepath.Clean(p))
		if p == "" || p == "." {
			continue
		}
		if !slices.Contains(out, p) {
			out = append(out, p)
		}
	}
	return out
}

func (w *Watcher) addRoot(root string) {
	w.rootsMu.Lock()
	defer w.rootsMu.Unlock()
	if !slices.Contains(w.roots, root) {
		w.roots = append(w.roots, root)
	}
}

func (w *Watcher) addShallowRoot(root string) {
	w.rootsMu.Lock()
	defer w.rootsMu.Unlock()
	if !slices.Contains(w.shallow, root) {
		w.shallow = append(w.shallow, root)
	}
}

// isUnderShallowRoot reports whether path's most specific containing watch
// root is a shallow root. A path that also sits under a more specific
// recursive root (for example a new sessions/YYYY/MM/DD directory beneath a
// recursive sessions root that itself lives inside a shallow parent root) is
// NOT shadowed, so auto-watching still adds new subdirectories of recursive
// roots.
func (w *Watcher) isUnderShallowRoot(path string) bool {
	root, ok := w.mostSpecificContainingRoot(path)
	if !ok {
		return false
	}
	w.rootsMu.RLock()
	defer w.rootsMu.RUnlock()
	return slices.Contains(w.shallow, root)
}

func (w *Watcher) shouldExclude(path string) bool {
	if len(w.excludes) == 0 {
		return false
	}
	root, ok := w.mostSpecificContainingRoot(path)
	if !ok {
		return false
	}
	return w.shouldExcludeForRoot(path, root)
}

func (w *Watcher) shouldExcludeForRoot(path string, root string) bool {
	if len(w.excludes) == 0 {
		return false
	}
	clean := filepath.Clean(path)
	root = filepath.Clean(root)
	rel, err := filepath.Rel(root, clean)
	if err != nil {
		return false
	}
	if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
		return false
	}
	rel = filepath.Clean(rel)
	if rel == "." {
		return false
	}
	parts := strings.Split(rel, string(filepath.Separator))

	for _, pat := range w.excludes {
		if strings.Contains(pat, string(filepath.Separator)) {
			if ok, _ := filepath.Match(pat, rel); ok {
				return true
			}
			continue
		}
		for _, part := range parts {
			if ok, _ := filepath.Match(pat, part); ok {
				return true
			}
		}
	}
	return false
}

func (w *Watcher) mostSpecificContainingRoot(path string) (string, bool) {
	w.rootsMu.RLock()
	defer w.rootsMu.RUnlock()

	if len(w.roots) == 0 {
		return "", false
	}

	clean := filepath.Clean(path)
	var best string
	for _, root := range w.roots {
		rel, err := filepath.Rel(root, clean)
		if err != nil {
			continue
		}
		if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
			continue
		}
		if best == "" || len(root) > len(best) {
			best = root
		}
	}
	if best == "" {
		return "", false
	}
	return best, true
}