fn_registry/functions/infra/watch_dir_fsnotify.go

package infra

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

	"github.com/fsnotify/fsnotify"
)

// WatchDirFsnotify crea un watcher recursivo sobre root y todos sus subdirectorios.
// Emite FsEvent al canal devuelto con debounce de 200ms por path (si llegan multiples
// eventos del mismo archivo en la ventana, se emite solo el ultimo).
// Cierra el canal cuando ctx.Done() se dispara.
func WatchDirFsnotify(ctx context.Context, root string) (<-chan FsEvent, error) {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, fmt.Errorf("watch_dir_fsnotify: new watcher: %w", err)
	}

	// Anadir root y todos los subdirectorios recursivamente.
	if err := addDirsRecursive(watcher, root); err != nil {
		watcher.Close()
		return nil, fmt.Errorf("watch_dir_fsnotify: add dirs: %w", err)
	}

	ch := make(chan FsEvent, 64)

	go func() {
		defer watcher.Close()
		defer close(ch)

		// Mapa de debounce: path -> (timer, ultimo op)
		type pending struct {
			timer *time.Timer
			op    string
		}
		debounce := make(map[string]*pending)
		const debounceDelay = 200 * time.Millisecond

		for {
			select {
			case <-ctx.Done():
				// Cancelar todos los timers pendientes antes de salir.
				for _, p := range debounce {
					p.timer.Stop()
				}
				return

			case event, ok := <-watcher.Events:
				if !ok {
					return
				}

				op := fsnotifyOpToString(event.Op)
				if op == "" {
					continue
				}

				path := event.Name

				// Si el directorio nuevo fue creado, anadirlo al watcher.
				if event.Op&fsnotify.Create != 0 {
					if info, err := os.Stat(path); err == nil && info.IsDir() {
						if err := watcher.Add(path); err != nil {
							log.Printf("watch_dir_fsnotify: add new dir %s: %v", path, err)
						}
					}
				}

				// Debounce: resetear el timer si ya habia uno para este path.
				if p, exists := debounce[path]; exists {
					p.timer.Stop()
					p.op = op
					p.timer.Reset(debounceDelay)
				} else {
					p = &pending{op: op}
					p.timer = time.AfterFunc(debounceDelay, func() {
						select {
						case ch <- FsEvent{Path: path, Op: p.op}:
						case <-ctx.Done():
						}
						delete(debounce, path)
					})
					debounce[path] = p
				}

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

	return ch, nil
}

// addDirsRecursive anade root y todos sus subdirectorios al watcher.
// Retorna error si root no existe o no es accesible.
func addDirsRecursive(watcher *fsnotify.Watcher, root string) error {
	if _, err := os.Stat(root); err != nil {
		return fmt.Errorf("root dir %s: %w", root, err)
	}
	return filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return nil // ignora errores de acceso en subdirs
		}
		if info.IsDir() {
			return watcher.Add(path)
		}
		return nil
	})
}

// fsnotifyOpToString convierte fsnotify.Op al string canonico del registry.
// Retorna "" para operaciones no mapeadas (CHMOD, etc.).
func fsnotifyOpToString(op fsnotify.Op) string {
	switch {
	case op&fsnotify.Create != 0:
		return "create"
	case op&fsnotify.Write != 0:
		return "write"
	case op&fsnotify.Remove != 0:
		return "remove"
	case op&fsnotify.Rename != 0:
		return "rename"
	default:
		return ""
	}
}