AdGuardHome/internal/aghos/fswatcher.go

package aghos

import (
	"context"
	"fmt"
	"io/fs"
	"log/slog"
	"path/filepath"

	"github.com/AdguardTeam/golibs/container"
	"github.com/AdguardTeam/golibs/errors"
	"github.com/AdguardTeam/golibs/logutil/slogutil"
	"github.com/AdguardTeam/golibs/osutil"
	"github.com/AdguardTeam/golibs/service"
	"github.com/fsnotify/fsnotify"
)

// event is a convenient alias for an empty struct to signal that watching
// event happened.
type event = struct{}

// FSWatcher tracks all the fyle system events and notifies about those.
//
// TODO(e.burkov, a.garipov): Move into another package like aghfs.
//
// TODO(e.burkov):  Add tests.
type FSWatcher interface {
	service.Interface

	// Events returns the channel to notify about the file system events.
	Events() (e <-chan event)

	// Add starts tracking the file.  It returns an error if the file can't be
	// tracked.  It must not be called after Start.
	Add(name string) (err error)
}

// osWatcher tracks the file system provided by the OS.
type osWatcher struct {
	// logger is used for logging the operations of the osWatcher.
	logger *slog.Logger

	// watcher is the actual notifier that is handled by osWatcher.
	watcher *fsnotify.Watcher

	// events is the channel to notify.
	events chan event

	// files is the set of tracked files.
	files *container.MapSet[string]
}

// osWatcherPref is a prefix for logging and wrapping errors in osWathcer's
// methods.
const osWatcherPref = "os watcher"

// NewOSWritesWatcher creates FSWatcher that tracks the real file system of the
// OS and notifies only about writing events.  l must not be nil.
func NewOSWritesWatcher(l *slog.Logger) (w FSWatcher, err error) {
	defer func() { err = errors.Annotate(err, "%s: %w", osWatcherPref) }()

	var watcher *fsnotify.Watcher
	watcher, err = fsnotify.NewWatcher()
	if err != nil {
		return nil, fmt.Errorf("creating watcher: %w", err)
	}

	return &osWatcher{
		logger:  l,
		watcher: watcher,
		events:  make(chan event, 1),
		files:   container.NewMapSet[string](),
	}, nil
}

// type check
var _ FSWatcher = (*osWatcher)(nil)

// Start implements the [FSWatcher] interface for *osWatcher.
func (w *osWatcher) Start(ctx context.Context) (err error) {
	go w.handleErrors(ctx)
	go w.handleEvents(ctx)

	return nil
}

// Shutdown implements the [FSWatcher] interface for *osWatcher.
func (w *osWatcher) Shutdown(_ context.Context) (err error) {
	return w.watcher.Close()
}

// Events implements the FSWatcher interface for *osWatcher.
func (w *osWatcher) Events() (e <-chan event) {
	return w.events
}

// Add implements the [FSWatcher] interface for *osWatcher.
//
// TODO(e.burkov):  Make it accept non-existing files to detect it's creating.
func (w *osWatcher) Add(name string) (err error) {
	defer func() { err = errors.Annotate(err, "%s: %w", osWatcherPref) }()

	fi, err := fs.Stat(osutil.RootDirFS(), name)
	if err != nil {
		return fmt.Errorf("checking file %q: %w", name, err)
	}

	name = filepath.Join("/", name)
	w.files.Add(name)

	// Watch the directory and filter the events by the file name, since the
	// common recomendation to the fsnotify package is to watch the directory
	// instead of the file itself.
	//
	// See https://pkg.go.dev/github.com/fsnotify/fsnotify@v1.7.0#readme-watching-a-file-doesn-t-work-well.
	if !fi.IsDir() {
		name = filepath.Dir(name)
	}

	return w.watcher.Add(name)
}

// handleEvents notifies about the received file system's event if needed.  It
// is intended to be used as a goroutine.
func (w *osWatcher) handleEvents(ctx context.Context) {
	defer slogutil.RecoverAndLog(ctx, w.logger)

	defer close(w.events)

	ch := w.watcher.Events
	for e := range ch {
		if e.Op&fsnotify.Write == 0 || !w.files.Has(e.Name) {
			continue
		}

		skipDuplicates(ch)

		select {
		case w.events <- event{}:
			// Go on.
		default:
			w.logger.DebugContext(ctx, "events buffer is full")
		}
	}
}

// skipDuplicates drains the given channel of events, assuming that some events
// might occur multiple times.
func skipDuplicates(ch <-chan fsnotify.Event) {
	for {
		select {
		case <-ch:
			// Go on.
		default:
			return
		}
	}
}

// handleErrors handles accompanying errors.  It used to be called in a separate
// goroutine.
func (w *osWatcher) handleErrors(ctx context.Context) {
	defer slogutil.RecoverAndLog(ctx, w.logger)

	for err := range w.watcher.Errors {
		w.logger.ErrorContext(ctx, "handling error", slogutil.KeyError, err)
	}
}

// EmptyFSWatcher is a no-op implementation of the [FSWatcher] interface.  It
// may be used on systems not supporting filesystem events.
type EmptyFSWatcher struct{}

// type check
var _ FSWatcher = EmptyFSWatcher{}

// Start implements the [FSWatcher] interface for EmptyFSWatcher.  It always
// returns nil error.
func (EmptyFSWatcher) Start(_ context.Context) (err error) {
	return nil
}

// Shutdown implements the [FSWatcher] interface for EmptyFSWatcher.  It always
// returns nil error.
func (EmptyFSWatcher) Shutdown(_ context.Context) (err error) {
	return nil
}

// Events implements the [FSWatcher] interface for EmptyFSWatcher.  It always
// returns nil channel.
func (EmptyFSWatcher) Events() (e <-chan event) {
	return nil
}

// Add implements the [FSWatcher] interface for EmptyFSWatcher.  It always
// returns nil error.
func (EmptyFSWatcher) Add(_ string) (err error) {
	return nil
}