lodge/backend/internal/client/daemon.go

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 !t.rules.ignored(rel) && slices.ContainsFunc(t.rules, func(rule Rule) bool {
		return !rule.Ignore && (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.synced() {
		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
			}
			rel, _ := filepath.Rel(repo.Root, path)
			if entry.Name() == ".git" || rules.ignored(filepath.ToSlash(rel)) {
				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()
}