watch.go RawHistoryFolder

1package config
2
3import (
4	"context"
5	"errors"
6	"io/fs"
7	"time"
8
9	"github.com/bnema/kvconf"
10	"github.com/bnema/neferafk/internal/ports"
11)
12
13// fallbackPoll is the polling interval used when native watching fails.
14var fallbackPoll = 2 * time.Second
15
16// kvWatch is kvconf.Watch; tests replace it to make native watching fail.
17var kvWatch = kvconf.Watch
18
19// Watch retains last-valid state across missing files/directories and follows
20// config symlink targets (kvconf.Watch). It always emits the first valid
21// candidate, including a candidate equal to defaults: this closes an app
22// Load -> Watch startup gap. Ready closes after initial watch installation,
23// before candidate delivery. No invalid/deleted candidate emits a defaults
24// reset. Errors never echo file content or paths, which may themselves
25// contain sensitive source metadata. When the OS refuses to watch (inotify
26// limits, permissions), Watch reports it on failures and polls instead: a
27// watch error must never stop the daemon and with it idle locking.
28func Watch(ctx context.Context, path string, out chan<- ports.ConfigChanged, failures chan<- error, ready chan<- struct{}) error {
29	ctx, cancel := context.WithCancel(ctx)
30	changed := make(chan struct{}, 1)
31	installed := make(chan struct{})
32	done := make(chan struct{})
33	var watchErr error
34	go func() {
35		defer close(done)
36		err := kvWatch(ctx, path, kvconf.WatchOptions{Ready: installed}, changed)
37		if err == nil || ctx.Err() != nil {
38			return
39		}
40		if failures != nil {
41			select {
42			case failures <- errors.New("config watch failed; polling"):
43			default:
44			}
45		}
46		opts := kvconf.WatchOptions{Poll: fallbackPoll}
47		select {
48		case <-installed: // closed by the native watch before it failed
49		default:
50			opts.Ready = installed
51		}
52		// Changes made while nothing watched are seen by this reload.
53		select {
54		case changed <- struct{}{}:
55		default:
56		}
57		watchErr = kvWatch(ctx, path, opts, changed)
58	}()
59	defer func() { cancel(); <-done }()
60
61	select {
62	case <-installed:
63	case <-done:
64		if watchErr != nil {
65			return errors.New("initialize config watch")
66		}
67		return nil
68	}
69	if ready != nil {
70		close(ready)
71	}
72	var last ports.Config
73	haveLast := false
74	publish := func() bool {
75		cfg, err := Load(path)
76		if err != nil {
77			if !errors.Is(err, fs.ErrNotExist) && failures != nil {
78				select {
79				case failures <- errors.New("config reload rejected"):
80				default:
81				}
82			}
83			return true
84		}
85		if haveLast && cfg == last {
86			return true
87		}
88		select {
89		case out <- ports.ConfigChanged{Config: cfg}:
90			last, haveLast = cfg, true
91			return true
92		case <-ctx.Done():
93			return false
94		}
95	}
96	for ok := publish(); ok; {
97		select {
98		case <-ctx.Done():
99			return nil
100		case <-done:
101			if watchErr != nil {
102				return errors.New("config watch")
103			}
104			return nil
105		case <-changed:
106			ok = publish()
107		}
108	}
109	return nil
110}