package client import ( "bufio" "context" "encoding/json" "errors" "io/fs" "log" "net/http" "os" "path/filepath" "slices" "strings" "sync" "time" "lodge/internal/gitrepo" "lodge/internal/protocol" "github.com/fsnotify/fsnotify" ) const ( debounce = 2 * time.Second fullSyncEvery = 5 * time.Minute minBackoff = 5 * time.Second maxBackoff = time.Minute headFile = "HEAD" ) type tree struct { key string gitDir string rules Rules } type daemon struct { client *Client watcher *fsnotify.Watcher queue chan string mu sync.Mutex owners map[string]string trees map[string]tree pending map[string]*time.Timer } func (c *Client) Daemon(ctx context.Context) error { watcher, err := fsnotify.NewWatcher() if err != nil { return err } defer watcher.Close() if err := os.MkdirAll(c.DataDir, 0o755); err != nil { return err } if err := watcher.Add(c.DataDir); err != nil { return err } d := &daemon{ client: c, watcher: watcher, queue: make(chan string, 256), owners: map[string]string{}, trees: map[string]tree{}, pending: map[string]*time.Timer{}, } go d.listen(ctx) go d.work(ctx) d.syncAll() ticker := time.NewTicker(fullSyncEvery) defer ticker.Stop() for { select { case <-ctx.Done(): return nil case <-ticker.C: d.syncAll() case err := <-watcher.Errors: log.Printf("watcher: %v", err) case event := <-watcher.Events: d.handle(event) } } } func (d *daemon) handle(event fsnotify.Event) { if event.Name == d.client.registryPath() { d.syncAll() return } d.mu.Lock() if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) { delete(d.owners, event.Name) } root, watched := d.owners[filepath.Dir(event.Name)] relevant := watched && d.trees[root].relevant(root, event.Name) d.mu.Unlock() if relevant { d.schedule(root) } } func (t tree) relevant(root, name string) bool { if filepath.Dir(name) == t.gitDir { return filepath.Base(name) == headFile } rel, err := filepath.Rel(root, name) if err != nil || strings.HasSuffix(name, tempSuffix) { return false } rel = filepath.ToSlash(rel) return slices.ContainsFunc(t.rules, func(rule Rule) bool { return within(rel, rule.Path) || within(rule.Path, rel) }) } func (d *daemon) schedule(root string) { d.mu.Lock() defer d.mu.Unlock() if timer, waiting := d.pending[root]; waiting { timer.Reset(debounce) return } d.pending[root] = time.AfterFunc(debounce, func() { d.mu.Lock() delete(d.pending, root) d.mu.Unlock() d.queue <- root }) } func (d *daemon) syncAll() { for _, root := range d.client.Worktrees() { if _, err := os.Stat(root); errors.Is(err, fs.ErrNotExist) { d.forget(root) } else { d.schedule(root) } } } func (d *daemon) forget(root string) { if err := d.client.Forget(root); err != nil { log.Printf("%s: %v", root, err) return } log.Printf("%s: gone, forgotten", root) d.mu.Lock() defer d.mu.Unlock() delete(d.trees, root) d.release(root, nil) } func (d *daemon) work(ctx context.Context) { for { select { case <-ctx.Done(): return case root := <-d.queue: res, err := d.client.Sync(ctx, root, true) if err != nil { log.Printf("%s: %v", root, err) } else if res.Pushed+res.Pulled > 0 { log.Printf("%s: pushed %d, pulled %d", root, res.Pushed, res.Pulled) } d.watch(root) } } } func (d *daemon) watch(root string) { repo, err := gitrepo.Open(root) if err != nil { return } rules := LoadRules(repo.Root) dirs := map[string]bool{repo.GitDir: true} for _, rule := range rules { target := filepath.Join(repo.Root, filepath.FromSlash(rule.Path)) dirs[nearestDir(filepath.Dir(target), repo.Root)] = true _ = filepath.WalkDir(target, func(path string, entry fs.DirEntry, err error) error { if err != nil || !entry.IsDir() { return nil } if entry.Name() == ".git" { return filepath.SkipDir } dirs[path] = true return nil }) } d.mu.Lock() defer d.mu.Unlock() d.trees[repo.Root] = tree{key: repo.Key, gitDir: repo.GitDir, rules: rules} d.release(repo.Root, dirs) for dir := range dirs { if owner, taken := d.owners[dir]; taken && owner != repo.Root { continue } if d.watcher.Add(dir) == nil { d.owners[dir] = repo.Root } } } func (d *daemon) release(root string, keep map[string]bool) { for dir, owner := range d.owners { if owner == root && !keep[dir] { _ = d.watcher.Remove(dir) delete(d.owners, dir) } } } func nearestDir(path, root string) string { for ; path != root; path = filepath.Dir(path) { if info, err := os.Stat(path); err == nil && info.IsDir() { return path } } return root } func (d *daemon) listen(ctx context.Context) { backoff := minBackoff for ctx.Err() == nil { started := time.Now() err := d.stream(ctx) if time.Since(started) > maxBackoff { backoff = minBackoff } log.Printf("event stream closed, retrying in %s: %v", backoff, err) select { case <-ctx.Done(): case <-time.After(backoff): } backoff = min(backoff*2, maxBackoff) } } func (d *daemon) stream(ctx context.Context) error { res, err := d.client.request(ctx, http.MethodGet, "/api/events", nil) if err != nil { return err } defer res.Body.Close() d.syncAll() scanner := bufio.NewScanner(res.Body) for scanner.Scan() { data, isEvent := strings.CutPrefix(scanner.Text(), "data: ") var event protocol.Event if !isEvent || json.Unmarshal([]byte(data), &event) != nil { continue } d.mu.Lock() var roots []string for root, tree := range d.trees { if tree.key == event.Repo { roots = append(roots, root) } } d.mu.Unlock() for _, root := range roots { d.schedule(root) } } return scanner.Err() }