iop/packages/go/agentconfig/watcher.go

package agentconfig

import (
	"context"
	"fmt"
	"sync"
	"time"
)

const DefaultRuntimeConfigPollInterval = 250 * time.Millisecond

// RuntimeConfigWatcher polls both source documents and atomically publishes
// only valid changed snapshots. Invalid edits are reported while the last
// valid snapshot remains current.
type RuntimeConfigWatcher struct {
	repoGlobalPath string
	userLocalPath  string
	pollInterval   time.Duration

	mu        sync.RWMutex
	current   RuntimeSnapshot
	updates   chan RuntimeSnapshot
	errors    chan error
	done      chan struct{}
	cancel    context.CancelFunc
	closeOnce sync.Once
}

// WatchRuntimeConfig starts a watcher using the default poll interval.
func WatchRuntimeConfig(
	ctx context.Context,
	repoGlobalPath string,
	userLocalPath string,
) (*RuntimeConfigWatcher, error) {
	return NewRuntimeConfigWatcher(ctx, repoGlobalPath, userLocalPath, DefaultRuntimeConfigPollInterval)
}

// NewRuntimeConfigWatcher loads the initial snapshot before starting its
// polling goroutine. A non-positive interval is rejected.
func NewRuntimeConfigWatcher(
	ctx context.Context,
	repoGlobalPath string,
	userLocalPath string,
	pollInterval time.Duration,
) (*RuntimeConfigWatcher, error) {
	if ctx == nil {
		return nil, fmt.Errorf("agentconfig: watcher context is required")
	}
	if pollInterval <= 0 {
		return nil, fmt.Errorf("agentconfig: watcher poll interval must be positive")
	}
	initial, err := LoadRuntimeConfig(repoGlobalPath, userLocalPath)
	if err != nil {
		return nil, fmt.Errorf("agentconfig: watcher initial load: %w", err)
	}
	watchContext, cancel := context.WithCancel(ctx)
	watcher := &RuntimeConfigWatcher{
		repoGlobalPath: repoGlobalPath,
		userLocalPath:  userLocalPath,
		pollInterval:   pollInterval,
		current:        initial,
		updates:        make(chan RuntimeSnapshot, 1),
		errors:         make(chan error, 1),
		done:           make(chan struct{}),
		cancel:         cancel,
	}
	go watcher.run(watchContext)
	return watcher, nil
}

// Snapshot returns the immutable snapshot pinned for a new invocation.
func (w *RuntimeConfigWatcher) Snapshot() RuntimeSnapshot {
	w.mu.RLock()
	defer w.mu.RUnlock()
	return w.current
}

// Updates reports valid revisions after the initial snapshot. The channel is
// bounded and coalesces unread values to the latest valid revision.
func (w *RuntimeConfigWatcher) Updates() <-chan RuntimeSnapshot {
	return w.updates
}

// Errors reports invalid reloads without changing Snapshot.
func (w *RuntimeConfigWatcher) Errors() <-chan error {
	return w.errors
}

// Done closes after the watcher stops.
func (w *RuntimeConfigWatcher) Done() <-chan struct{} {
	return w.done
}

// Close stops the watcher and waits for its channels to close.
func (w *RuntimeConfigWatcher) Close() {
	w.closeOnce.Do(w.cancel)
	<-w.done
}

func (w *RuntimeConfigWatcher) run(ctx context.Context) {
	defer close(w.done)
	defer close(w.errors)
	defer close(w.updates)

	ticker := time.NewTicker(w.pollInterval)
	defer ticker.Stop()

	lastError := ""
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			next, err := LoadRuntimeConfig(w.repoGlobalPath, w.userLocalPath)
			if err != nil {
				message := err.Error()
				if message != lastError {
					w.publishError(err)
					lastError = message
				}
				continue
			}
			lastError = ""

			w.mu.Lock()
			if next.Revision() == w.current.Revision() {
				w.mu.Unlock()
				continue
			}
			w.current = next
			w.mu.Unlock()
			w.publishUpdate(next)
		}
	}
}

func (w *RuntimeConfigWatcher) publishUpdate(snapshot RuntimeSnapshot) {
	select {
	case w.updates <- snapshot:
		return
	default:
	}
	select {
	case <-w.updates:
	default:
	}
	select {
	case w.updates <- snapshot:
	default:
	}
}

func (w *RuntimeConfigWatcher) publishError(err error) {
	select {
	case w.errors <- err:
		return
	default:
	}
	select {
	case <-w.errors:
	default:
	}
	select {
	case w.errors <- err:
	default:
	}
}