lodge/backend/internal/client/daemon.go

package client

import (
	"bufio"
	"context"
	"encoding/json"
	"io/fs"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"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 daemon struct {
	client  *Client
	watcher *fsnotify.Watcher
	queue   chan string

	mu      sync.Mutex
	owners  map[string]string
	gitDirs map[string]bool
	keys    map[string]string
	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{},
		gitDirs: map[string]bool{},
		keys:    map[string]string{},
		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
	}
	dir := filepath.Dir(event.Name)
	d.mu.Lock()
	root, watched := d.owners[dir]
	irrelevant := d.gitDirs[dir] && filepath.Base(event.Name) != headFile
	d.mu.Unlock()
	if watched && !irrelevant && !strings.HasSuffix(event.Name, tempSuffix) {
		d.schedule(root)
	}
}

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() {
		d.schedule(root)
	}
}

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
	}
	dirs := []string{repo.GitDir}
	for _, rule := range LoadRules(repo.Root) {
		target := filepath.Join(repo.Root, filepath.FromSlash(rule.Path))
		info, err := os.Stat(target)
		for err != nil && target != repo.Root {
			target = filepath.Dir(target)
			info, err = os.Stat(target)
		}
		if err != nil {
			continue
		}
		if !info.IsDir() {
			dirs = append(dirs, filepath.Dir(target))
			continue
		}
		_ = 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 = append(dirs, path)
			return nil
		})
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	d.keys[repo.Root] = repo.Key
	d.gitDirs[repo.GitDir] = true
	for _, dir := range dirs {
		if _, watching := d.owners[dir]; !watching && d.watcher.Add(dir) == nil {
			d.owners[dir] = repo.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, key := range d.keys {
			if key == event.Repo {
				roots = append(roots, root)
			}
		}
		d.mu.Unlock()
		for _, root := range roots {
			d.schedule(root)
		}
	}
	return scanner.Err()
}