21959106240

Committed 12 Feb 2026 06:24PM UTC coverage: 80.884% (-0.04%) from 80.928%

Build # 21959106240

Build Type

push

github

Committed by

umputun

Commit Message

docs: update changelog for v0.10.5

Run Details

5471 of 6764 relevant lines covered (80.88%)

171.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.72

/pkg/web/watcher.go

package web

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

        "github.com/fsnotify/fsnotify"
)

// skipDirs is the set of directory names to skip during recursive watching.
// these are known high-volume or non-relevant directories that won't contain progress files.
var skipDirs = map[string]bool{
        ".git":         true,
        ".idea":        true,
        ".vscode":      true,
        ".cache":       true,
        ".npm":         true,
        ".yarn":        true,
        "node_modules": true,
        "vendor":       true,
        "__pycache__":  true,
        "target":       true,
        "build":        true,
        "dist":         true,
}

// Watcher monitors directories for progress file changes.
// it uses fsnotify for efficient file system event detection
// and notifies the SessionManager when new progress files appear.
type Watcher struct {
        dirs    []string
        sm      *SessionManager
        watcher *fsnotify.Watcher

        mu      sync.Mutex
        started bool
}

// NewWatcher creates a watcher for the specified directories.
// directories are watched recursively for progress-*.txt files.
func NewWatcher(dirs []string, sm *SessionManager) (*Watcher, error) {
        w, err := fsnotify.NewWatcher()
        if err != nil {
                return nil, fmt.Errorf("create fsnotify watcher: %w", err)
        }

        return &Watcher{
                dirs:    dirs,
                sm:      sm,
                watcher: w,
        }, nil
}

// Start begins watching directories for progress file changes.
// runs until the context is canceled.
// performs initial discovery before starting the watch loop.
func (w *Watcher) Start(ctx context.Context) error {
        w.mu.Lock()
        if w.started {
                w.mu.Unlock()
                return nil
        }
        w.started = true
        w.mu.Unlock()

        // add all directories to watcher (including subdirectories)
        for _, dir := range w.dirs {
                if err := w.addRecursive(dir); err != nil {
                        return err
                }
        }

        // initial discovery (recursive to find existing progress files in subdirectories)
        for _, dir := range w.dirs {
                if _, err := w.sm.DiscoverRecursive(dir); err != nil {
                        log.Printf("[WARN] initial discovery failed for %s: %v", dir, err)
                }
        }

        // start tailing for active sessions
        w.sm.StartTailingActive()

        // start periodic state refresh to detect completed sessions
        go w.refreshLoop(ctx)

        // run the watch loop
        return w.run(ctx)
}

// addRecursive adds a directory and all its subdirectories to the watcher.
func (w *Watcher) addRecursive(dir string) error {
        walkErr := filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error {
                if err != nil {
                        // skip directories that can't be accessed
                        if d != nil && d.IsDir() {
                                return filepath.SkipDir
                        }
                        return err
                }

                if d.IsDir() {
                        name := d.Name()
                        // skip directories that typically contain many subdirs and no progress files
                        if skipDirs[name] && path != dir {
                                return filepath.SkipDir
                        }
                        // best-effort: continue walking even if we can't watch a specific directory
                        if err := w.watcher.Add(path); err != nil {
                                log.Printf("[WARN] failed to watch directory %s: %v", path, err)
                        }
                }
                return nil
        })
        if walkErr != nil {
                return fmt.Errorf("walk directory %s: %w", dir, walkErr)
        }
        return nil
}

// run is the main watch loop processing fsnotify events.
func (w *Watcher) run(ctx context.Context) error {
        for {
                select {
                case <-ctx.Done():
                        return w.Close()

                case event, ok := <-w.watcher.Events:
                        if !ok {
                                return nil
                        }
                        w.handleEvent(event)

                case err, ok := <-w.watcher.Errors:
                        if !ok {
                                return nil
                        }
                        // log error but continue watching
                        log.Printf("[WARN] fsnotify error: %v", err)
                }
        }
}

// handleEvent processes a single fsnotify event.
func (w *Watcher) handleEvent(event fsnotify.Event) {
        // filter for progress-*.txt files only
        if !isProgressFile(event.Name) {
                w.handleNonProgressEvent(event)
                return
        }

        // handle create or write events
        if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) {
                w.handleProgressFileChange(event.Name)
        }

        // handle remove events
        if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) {
                id := sessionIDFromPath(event.Name)
                w.sm.Remove(id)
        }
}

// handleNonProgressEvent handles events for non-progress files (e.g., new directories).
func (w *Watcher) handleNonProgressEvent(event fsnotify.Event) {
        if !event.Has(fsnotify.Create) {
                return
        }
        info, err := os.Stat(event.Name)
        if err != nil || !info.IsDir() {
                return
        }
        if err := w.addRecursive(event.Name); err != nil {
                log.Printf("[WARN] failed to watch new directory %s: %v", event.Name, err)
        }
}

// handleProgressFileChange handles create/write events for progress files.
func (w *Watcher) handleProgressFileChange(path string) {
        dir := filepath.Dir(path)
        ids, err := w.sm.Discover(dir)
        if err != nil {
                log.Printf("[WARN] discovery failed for %s: %v", dir, err)
                return
        }

        // start tailing for any newly active sessions
        for _, id := range ids {
                w.startTailingIfNeeded(id)
        }
}

// startTailingIfNeeded starts tailing for a session if it's active and not already tailing.
func (w *Watcher) startTailingIfNeeded(id string) {
        session := w.sm.Get(id)
        if session == nil {
                return
        }
        if session.GetState() != SessionStateActive || session.IsTailing() {
                return
        }
        if err := session.StartTailing(true); err != nil {
                log.Printf("[WARN] failed to start tailing for session %s: %v", id, err)
        }
}

// refreshLoop periodically checks for session state changes (active->completed).
// runs until context is canceled.
func (w *Watcher) refreshLoop(ctx context.Context) {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()

        for {
                select {
                case <-ctx.Done():
                        return
                case <-ticker.C:
                        w.sm.RefreshStates()
                }
        }
}

// Close stops the watcher and releases resources.
func (w *Watcher) Close() error {
        if err := w.watcher.Close(); err != nil {
                return fmt.Errorf("close fsnotify watcher: %w", err)
        }
        return nil
}

// isProgressFile returns true if the path matches progress-*.txt pattern.
func isProgressFile(path string) bool {
        name := filepath.Base(path)
        return strings.HasPrefix(name, "progress-") && strings.HasSuffix(name, ".txt")
}

// ResolveWatchDirs determines the directories to watch based on precedence:
// CLI flags > config file > current directory (default).
// returns at least one directory (current directory if nothing else specified).
func ResolveWatchDirs(cliDirs, configDirs []string) []string {
        // CLI flags take highest precedence
        if len(cliDirs) > 0 {
                return normalizeDirs(cliDirs)
        }

        // config file is second
        if len(configDirs) > 0 {
                return normalizeDirs(configDirs)
        }

        // default to current directory
        cwd, err := os.Getwd()
        if err != nil {
                return []string{"."}
        }
        return []string{cwd}
}

// normalizeDirs converts relative paths to absolute and removes duplicates.
// logs warnings for invalid directories to help users debug configuration issues.
func normalizeDirs(dirs []string) []string {
        seen := make(map[string]bool)
        result := make([]string, 0, len(dirs))

        for _, dir := range dirs {
                // convert to absolute path
                abs, err := filepath.Abs(dir)
                if err != nil {
                        log.Printf("[WARN] failed to resolve path %q: %v", dir, err)
                        abs = dir
                }

                // resolve symlinks for consistent deduplication (macOS has /var -> /private/var)
                if resolved, evalErr := filepath.EvalSymlinks(abs); evalErr == nil {
                        abs = resolved
                }

                // skip duplicates
                if seen[abs] {
                        continue
                }
                seen[abs] = true

                // verify directory exists
                info, err := os.Stat(abs)
                if err != nil {
                        log.Printf("[WARN] watch directory %q does not exist: %v", abs, err)
                        continue
                }
                if !info.IsDir() {
                        log.Printf("[WARN] watch path %q is not a directory", abs)
                        continue
                }
                result = append(result, abs)
        }

        // fallback to current directory if all specified dirs are invalid
        if len(result) == 0 {
                log.Printf("[WARN] all watch directories invalid, falling back to current directory")
                cwd, err := os.Getwd()
                if err != nil {
                        return []string{"."}
                }
                return []string{cwd}
        }

        return result
}

1	package web
2
3	import (
4	"context"
5	"fmt"
6	"log"
7	"os"
8	"path/filepath"
9	"strings"
10	"sync"
11	"time"
12
13	"github.com/fsnotify/fsnotify"
14	)
15
16	// skipDirs is the set of directory names to skip during recursive watching.
17	// these are known high-volume or non-relevant directories that won't contain progress files.
18	var skipDirs = map[string]bool{
19	".git": true,
20	".idea": true,
21	".vscode": true,
22	".cache": true,
23	".npm": true,
24	".yarn": true,
25	"node_modules": true,
26	"vendor": true,
27	"__pycache__": true,
28	"target": true,
29	"build": true,
30	"dist": true,
31	}
32
33	// Watcher monitors directories for progress file changes.
34	// it uses fsnotify for efficient file system event detection
35	// and notifies the SessionManager when new progress files appear.
36	type Watcher struct {
37	dirs []string
38	sm *SessionManager
39	watcher *fsnotify.Watcher
40
41	mu sync.Mutex
42	started bool
43	}
44
45	// NewWatcher creates a watcher for the specified directories.
46	// directories are watched recursively for progress-*.txt files.
47	func NewWatcher(dirs []string, sm SessionManager) (Watcher, error) {	20✔
48	w, err := fsnotify.NewWatcher()	20✔
49	if err != nil {	20✔
50	return nil, fmt.Errorf("create fsnotify watcher: %w", err)	×
51	}	×
52
53	return &Watcher{	20✔
54	dirs: dirs,	20✔
55	sm: sm,	20✔
56	watcher: w,	20✔
57	}, nil	20✔
58	}
59
60	// Start begins watching directories for progress file changes.
61	// runs until the context is canceled.
62	// performs initial discovery before starting the watch loop.
63	func (w *Watcher) Start(ctx context.Context) error {	19✔
64	w.mu.Lock()	19✔
65	if w.started {	20✔
66	w.mu.Unlock()	1✔
67	return nil	1✔
68	}	1✔
69	w.started = true	18✔
70	w.mu.Unlock()	18✔
71		18✔
72	// add all directories to watcher (including subdirectories)	18✔
73	for _, dir := range w.dirs {	36✔
74	if err := w.addRecursive(dir); err != nil {	18✔
75	return err	×
76	}	×
77	}
78
79	// initial discovery (recursive to find existing progress files in subdirectories)
80	for _, dir := range w.dirs {	36✔
81	if _, err := w.sm.DiscoverRecursive(dir); err != nil {	18✔
82	log.Printf("[WARN] initial discovery failed for %s: %v", dir, err)	×
83	}	×
84	}
85
86	// start tailing for active sessions
87	w.sm.StartTailingActive()	18✔
88		18✔
89	// start periodic state refresh to detect completed sessions	18✔
90	go w.refreshLoop(ctx)	18✔
91		18✔
92	// run the watch loop	18✔
93	return w.run(ctx)	18✔
94	}
95
96	// addRecursive adds a directory and all its subdirectories to the watcher.
97	func (w *Watcher) addRecursive(dir string) error {	19✔
98	walkErr := filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error {	47✔
99	if err != nil {	28✔
100	// skip directories that can't be accessed	×
101	if d != nil && d.IsDir() {	×
102	return filepath.SkipDir	×
103	}	×
104	return err	×
105	}
106
107	if d.IsDir() {	55✔
108	name := d.Name()	27✔
109	// skip directories that typically contain many subdirs and no progress files	27✔
110	if skipDirs[name] && path != dir {	33✔
111	return filepath.SkipDir	6✔
112	}	6✔
113	// best-effort: continue walking even if we can't watch a specific directory
114	if err := w.watcher.Add(path); err != nil {	21✔
115	log.Printf("[WARN] failed to watch directory %s: %v", path, err)	×
116	}	×
117	}
118	return nil	22✔
119	})
120	if walkErr != nil {	19✔
121	return fmt.Errorf("walk directory %s: %w", dir, walkErr)	×
122	}	×
123	return nil	19✔
124	}
125
126	// run is the main watch loop processing fsnotify events.
127	func (w *Watcher) run(ctx context.Context) error {	18✔
128	for {	48✔
129	select {	30✔
130	case <-ctx.Done():	17✔
131	return w.Close()	17✔
132
133	case event, ok := <-w.watcher.Events:	12✔
134	if !ok {	12✔
135	return nil	×
136	}	×
137	w.handleEvent(event)	12✔
138
139	case err, ok := <-w.watcher.Errors:	1✔
140	if !ok {	2✔
141	return nil	1✔
142	}	1✔
143	// log error but continue watching
144	log.Printf("[WARN] fsnotify error: %v", err)	×
145	}
146	}
147	}
148
149	// handleEvent processes a single fsnotify event.
150	func (w *Watcher) handleEvent(event fsnotify.Event) {	12✔
151	// filter for progress-*.txt files only	12✔
152	if !isProgressFile(event.Name) {	15✔
153	w.handleNonProgressEvent(event)	3✔
154	return	3✔
155	}	3✔
156
157	// handle create or write events
158	if event.Has(fsnotify.Create) \|\| event.Has(fsnotify.Write) {	17✔
159	w.handleProgressFileChange(event.Name)	8✔
160	}	8✔
161
162	// handle remove events
163	if event.Has(fsnotify.Remove) \|\| event.Has(fsnotify.Rename) {	10✔
164	id := sessionIDFromPath(event.Name)	1✔
165	w.sm.Remove(id)	1✔
166	}	1✔
167	}
168
169	// handleNonProgressEvent handles events for non-progress files (e.g., new directories).
170	func (w *Watcher) handleNonProgressEvent(event fsnotify.Event) {	3✔
171	if !event.Has(fsnotify.Create) {	4✔
172	return	1✔
173	}	1✔
174	info, err := os.Stat(event.Name)	2✔
175	if err != nil \|\| !info.IsDir() {	3✔
176	return	1✔
177	}	1✔
178	if err := w.addRecursive(event.Name); err != nil {	1✔
179	log.Printf("[WARN] failed to watch new directory %s: %v", event.Name, err)	×
180	}	×
181	}
182
183	// handleProgressFileChange handles create/write events for progress files.
184	func (w *Watcher) handleProgressFileChange(path string) {	8✔
185	dir := filepath.Dir(path)	8✔
186	ids, err := w.sm.Discover(dir)	8✔
187	if err != nil {	8✔
188	log.Printf("[WARN] discovery failed for %s: %v", dir, err)	×
189	return	×
190	}	×
191
192	// start tailing for any newly active sessions
193	for _, id := range ids {	16✔
194	w.startTailingIfNeeded(id)	8✔
195	}	8✔
196	}
197
198	// startTailingIfNeeded starts tailing for a session if it's active and not already tailing.
199	func (w *Watcher) startTailingIfNeeded(id string) {	8✔
200	session := w.sm.Get(id)	8✔
201	if session == nil {	8✔
202	return	×
203	}	×
204	if session.GetState() != SessionStateActive \|\| session.IsTailing() {	16✔
205	return	8✔
206	}	8✔
207	if err := session.StartTailing(true); err != nil {	×
208	log.Printf("[WARN] failed to start tailing for session %s: %v", id, err)	×
209	}	×
210	}
211
212	// refreshLoop periodically checks for session state changes (active->completed).
213	// runs until context is canceled.
214	func (w *Watcher) refreshLoop(ctx context.Context) {	18✔
215	ticker := time.NewTicker(5 * time.Second)	18✔
216	defer ticker.Stop()	18✔
217		18✔
218	for {	36✔
219	select {	18✔
220	case <-ctx.Done():	18✔
221	return	18✔
222	case <-ticker.C:	×
223	w.sm.RefreshStates()	×
224	}
225	}
226	}
227
228	// Close stops the watcher and releases resources.
229	func (w *Watcher) Close() error {	20✔
230	if err := w.watcher.Close(); err != nil {	20✔
231	return fmt.Errorf("close fsnotify watcher: %w", err)	×
232	}	×
233	return nil	20✔
234	}
235
236	// isProgressFile returns true if the path matches progress-*.txt pattern.
237	func isProgressFile(path string) bool {	23✔
238	name := filepath.Base(path)	23✔
239	return strings.HasPrefix(name, "progress-") && strings.HasSuffix(name, ".txt")	23✔
240	}	23✔
241
242	// ResolveWatchDirs determines the directories to watch based on precedence:
243	// CLI flags > config file > current directory (default).
244	// returns at least one directory (current directory if nothing else specified).
245	func ResolveWatchDirs(cliDirs, configDirs []string) []string {	7✔
246	// CLI flags take highest precedence	7✔
247	if len(cliDirs) > 0 {	12✔
248	return normalizeDirs(cliDirs)	5✔
249	}	5✔
250
251	// config file is second
252	if len(configDirs) > 0 {	3✔
253	return normalizeDirs(configDirs)	1✔
254	}	1✔
255
256	// default to current directory
257	cwd, err := os.Getwd()	1✔
258	if err != nil {	1✔
259	return []string{"."}	×
260	}	×
261	return []string{cwd}	1✔
262	}
263
264	// normalizeDirs converts relative paths to absolute and removes duplicates.
265	// logs warnings for invalid directories to help users debug configuration issues.
266	func normalizeDirs(dirs []string) []string {	7✔
267	seen := make(map[string]bool)	7✔
268	result := make([]string, 0, len(dirs))	7✔
269		7✔
270	for _, dir := range dirs {	17✔
271	// convert to absolute path	10✔
272	abs, err := filepath.Abs(dir)	10✔
273	if err != nil {	10✔
274	log.Printf("[WARN] failed to resolve path %q: %v", dir, err)	×
275	abs = dir	×
276	}	×
277
278	// resolve symlinks for consistent deduplication (macOS has /var -> /private/var)
279	if resolved, evalErr := filepath.EvalSymlinks(abs); evalErr == nil {	18✔
280	abs = resolved	8✔
281	}	8✔
282
283	// skip duplicates
284	if seen[abs] {	12✔
285	continue	2✔
286	}
287	seen[abs] = true	8✔
288		8✔
289	// verify directory exists	8✔
290	info, err := os.Stat(abs)	8✔
291	if err != nil {	10✔
292	log.Printf("[WARN] watch directory %q does not exist: %v", abs, err)	2✔
293	continue	2✔
294	}
295	if !info.IsDir() {	6✔
296	log.Printf("[WARN] watch path %q is not a directory", abs)	×
297	continue	×
298	}
299	result = append(result, abs)	6✔
300	}
301
302	// fallback to current directory if all specified dirs are invalid
303	if len(result) == 0 {	8✔
304	log.Printf("[WARN] all watch directories invalid, falling back to current directory")	1✔
305	cwd, err := os.Getwd()	1✔
306	if err != nil {	1✔
307	return []string{"."}	×
308	}	×
309	return []string{cwd}	1✔
310	}
311
312	return result	6✔
313	}

umputun / ralphex / 21959106240

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous