feat(client): lodge remote binds a repository to another server

backend/internal/client/daemon.go

@@ -36,38 +36,46 @@ type tree struct {
 }
 
 type daemon struct {
-	client  *Client
+	config  *Config
 	watcher *fsnotify.Watcher
 	queue   chan string
 
-	mu      sync.Mutex
-	owners  map[string]string
-	trees   map[string]tree
-	pending map[string]*time.Timer
+	mu        sync.Mutex
+	current   *Config
+	listeners map[Client]context.CancelFunc
+	owners    map[string]string
+	trees     map[string]tree
+	pending   map[string]*time.Timer
 }
 
-func (c *Client) Daemon(ctx context.Context) error {
+func (cfg *Config) 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
+	for _, dir := range []string{cfg.DataDir, filepath.Dir(cfg.path)} {
+		if err := os.MkdirAll(dir, 0o755); err != nil {
+			return err
+		}
+		if err := watcher.Add(dir); 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{},
+		config:    cfg,
+		watcher:   watcher,
+		queue:     make(chan string, 256),
+		current:   cfg,
+		listeners: map[Client]context.CancelFunc{},
+		owners:    map[string]string{},
+		trees:     map[string]tree{},
+		pending:   map[string]*time.Timer{},
 	}
-	go d.listen(ctx)
-	go d.work(ctx)
+	d.connect(ctx)
+	var workers sync.WaitGroup
+	defer workers.Wait()
+	workers.Go(func() { d.work(ctx) })
 	d.syncAll()
 
 	ticker := time.NewTicker(fullSyncEvery)
@@ -81,13 +89,43 @@ func (c *Client) Daemon(ctx context.Context) error {
 		case err := <-watcher.Errors:
 			log.Printf("watcher: %v", err)
 		case event := <-watcher.Events:
-			d.handle(event)
+			d.handle(ctx, event)
 		}
 	}
 }
 
-func (d *daemon) handle(event fsnotify.Event) {
-	if event.Name == d.client.registryPath() {
+func (d *daemon) connect(ctx context.Context) {
+	current, err := d.config.reload()
+	if err != nil {
+		log.Printf("config: %v", err)
+		return
+	}
+	clients := current.clients()
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.current = current
+	for c, stop := range d.listeners {
+		if !clients[c] {
+			stop()
+			delete(d.listeners, c)
+		}
+	}
+	for c := range clients {
+		if d.listeners[c] == nil {
+			listenCtx, stop := context.WithCancel(ctx)
+			d.listeners[c] = stop
+			go d.listen(listenCtx, c)
+		}
+	}
+}
+
+func (d *daemon) handle(ctx context.Context, event fsnotify.Event) {
+	switch event.Name {
+	case d.config.registryPath():
+		d.syncAll()
+		return
+	case d.config.path:
+		d.connect(ctx)
 		d.syncAll()
 		return
 	}
@@ -133,7 +171,7 @@ func (d *daemon) schedule(root string) {
 }
 
 func (d *daemon) syncAll() {
-	for _, root := range d.client.Worktrees() {
+	for _, root := range d.config.Worktrees() {
 		if _, err := os.Stat(root); errors.Is(err, fs.ErrNotExist) {
 			d.forget(root)
 		} else {
@@ -143,7 +181,7 @@ func (d *daemon) syncAll() {
 }
 
 func (d *daemon) forget(root string) {
-	if err := d.client.Forget(root); err != nil {
+	if err := d.config.Forget(root); err != nil {
 		log.Printf("%s: %v", root, err)
 		return
 	}
@@ -160,10 +198,13 @@ func (d *daemon) work(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case root := <-d.queue:
-			res, err := d.client.Sync(ctx, root, true)
-			if err != nil {
+			res, err := d.config.Sync(ctx, root, true)
+			switch {
+			case errors.Is(err, ErrNotLoggedIn):
+				continue
+			case err != nil:
 				log.Printf("%s: %v", root, err)
-			} else if res.Pushed+res.Pulled > 0 {
+			case res.Pushed+res.Pulled > 0:
 				log.Printf("%s: pushed %d, pulled %d", root, res.Pushed, res.Pulled)
 			}
 			d.watch(root)
@@ -225,25 +266,29 @@ func nearestDir(path, root string) string {
 	return root
 }
 
-func (d *daemon) listen(ctx context.Context) {
+func (d *daemon) listen(ctx context.Context, c Client) {
 	backoff := minBackoff
-	for ctx.Err() == nil {
+	for {
 		started := time.Now()
-		err := d.stream(ctx)
+		err := d.stream(ctx, c)
+		if ctx.Err() != nil {
+			return
+		}
 		if time.Since(started) > maxBackoff {
 			backoff = minBackoff
 		}
-		log.Printf("event stream closed, retrying in %s: %v", backoff, err)
+		log.Printf("%s: event stream closed, retrying in %s: %v", c.URL, backoff, err)
 		select {
 		case <-ctx.Done():
+			return
 		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)
+func (d *daemon) stream(ctx context.Context, c Client) error {
+	res, err := c.request(ctx, http.MethodGet, "/api/events", nil)
 	if err != nil {
 		return err
 	}
@@ -258,9 +303,11 @@ func (d *daemon) stream(ctx context.Context) error {
 		}
 		d.mu.Lock()
 		var roots []string
-		for root, tree := range d.trees {
-			if tree.key == event.Repo {
-				roots = append(roots, root)
+		if d.current.serverOf(event.Repo) == c.URL {
+			for root, tree := range d.trees {
+				if tree.key == event.Repo {
+					roots = append(roots, root)
+				}
 			}
 		}
 		d.mu.Unlock()