feat: lodge server, client and admin

backend/internal/client/daemon.go

@@ -0,0 +1,222 @@
+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()
+}