subscribe to container changes in Caddyfile controller

internal/machine/caddyfile/controller.go

@@ -3,9 +3,11 @@ package caddyfile
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"os"
 	"path/filepath"
 	"sync"
+	"uncloud/internal/api"
 	"uncloud/internal/machine/store"
 )
 
@@ -30,6 +32,62 @@ func NewController(store *store.Store, path string) (*Controller, error) {
 	}, nil
 }
 
-func (cc *Controller) Run(ctx context.Context) error {
+func (c *Controller) Run(ctx context.Context) error {
+	containerRecords, changes, err := c.store.SubscribeContainers(ctx)
+	if err != nil {
+		return fmt.Errorf("subscribe to container changes: %w", err)
+	}
+	slog.Info("Subscribed to container changes in the cluster to generate Caddy configuration.")
+
+	containers, err := c.filterAvailableContainers(containerRecords)
+	if err != nil {
+		return fmt.Errorf("filter available containers: %w", err)
+	}
+	if err = c.generateConfig(containers); err != nil {
+		return fmt.Errorf("generate Caddy configuration: %w", err)
+	}
+
+	for {
+		select {
+		case _, ok := <-changes:
+			if !ok {
+				return fmt.Errorf("containers subscription failed")
+			}
+			slog.Debug("Cluster containers changed, updating Caddy configuration.")
+
+			containerRecords, err = c.store.ListContainers(ctx, store.ListOptions{})
+			if err != nil {
+				slog.Error("Failed to list containers.", "err", err)
+				continue
+			}
+			containers, err = c.filterAvailableContainers(containerRecords)
+			if err != nil {
+				slog.Error("Failed to filter available containers.", "err", err)
+				continue
+			}
+			if err = c.generateConfig(containers); err != nil {
+				slog.Error("Failed to generate Caddy configuration.", "err", err)
+			}
+
+			slog.Debug("Updated Caddy configuration.", "path", c.path)
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// filterAvailableContainers filters out containers that are likely unavailable from this machine. The availability
+// is determined by the cluster membership state of the machine that the container is running on.
+// TODO: implement machine membership check using Corrossion Admin client.
+func (c *Controller) filterAvailableContainers(containerRecords []*store.ContainerRecord) ([]*api.Container, error) {
+	containers := make([]*api.Container, len(containerRecords))
+	for i, cr := range containerRecords {
+		containers[i] = cr.Container
+	}
+	return containers, nil
+}
+
+func (c *Controller) generateConfig(containers []*api.Container) error {
+	// TODO
 	return nil
 }

internal/machine/docker/manager.go

@@ -184,6 +184,7 @@ func (m *Manager) syncContainersToStore(ctx context.Context) error {
 	}
 
 	// Create or update the current Docker containers in the store.
+	// TODO: update only the changed containers to reduce unnecessary gossip traffic and not trigger other controllers.
 	for _, dc := range containers {
 		c := &api.Container{Container: dc}
 		if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil {

internal/machine/store/container.go

@@ -76,7 +76,8 @@ func (s *Store) CreateOrUpdateContainer(ctx context.Context, c *api.Container, m
 
 // ListContainers returns a list of container records from the store database that match the given options.
 func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]*ContainerRecord, error) {
-	q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers")
+	q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers").
+		Where(sq.Eq{"sync_status": SyncStatusSynced})
 
 	if len(opts.MachineIDs) > 0 {
 		q = q.Where(sq.Eq{"machine_id": opts.MachineIDs})
@@ -154,3 +155,66 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error
 
 	return nil
 }
+
+// SubscribeContainers returns a list of containers and a channel that signals changes to the list. The channel doesn't
+// receive any values, it just signals when a container(s) has been added, updated, or deleted in the database.
+func (s *Store) SubscribeContainers(ctx context.Context) ([]*ContainerRecord, <-chan struct{}, error) {
+	// TODO: figure out whether we need sync_status at all.
+	q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers").
+		Where(sq.Eq{"sync_status": SyncStatusSynced})
+	query, args, err := q.ToSql()
+	if err != nil {
+		return nil, nil, fmt.Errorf("build query: %w", err)
+	}
+
+	sub, err := s.corro.SubscribeContext(ctx, query, args, false)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var containers []*ContainerRecord
+	var cJSON, updatedAtStr string
+
+	rows := sub.Rows()
+	for rows.Next() {
+		var cr ContainerRecord
+		if err = rows.Scan(&cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil {
+			return nil, nil, err
+		}
+
+		if err = json.Unmarshal([]byte(cJSON), &cr.Container); err != nil {
+			return nil, nil, fmt.Errorf("unmarshal container: %w", err)
+		}
+		if cr.UpdatedAt, err = time.Parse(time.DateTime, updatedAtStr); err != nil {
+			return nil, nil, fmt.Errorf("parse updated_at: %w", err)
+		}
+		containers = append(containers, &cr)
+	}
+	events, err := sub.Changes()
+	if err != nil {
+		return nil, nil, fmt.Errorf("get subscription changes: %w", err)
+	}
+
+	changes := make(chan struct{})
+	go func() {
+		defer close(changes)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case _, ok := <-events:
+				if !ok {
+					// events channel has been closed.
+					if sub.Err() != nil {
+						slog.Error("Containers subscription failed.", "id", sub.ID(), "err", sub.Err())
+					}
+					return
+				}
+				// Just signal that there is a change in the containers list.
+				changes <- struct{}{}
+			}
+		}
+	}()
+
+	return containers, changes, nil
+}

internal/machine/store/schema.sql

@@ -19,7 +19,7 @@ CREATE TABLE machines
 CREATE TABLE containers
 (
     id           TEXT NOT NULL PRIMARY KEY,
-    -- container is a JSON-serialized Docker container.Summary struct.
+    -- container is a JSON-serialized api.Container struct.
     container    TEXT NOT NULL DEFAULT '{}' CHECK (json_valid(container)),
     machine_id   TEXT NOT NULL DEFAULT '',
     service_id   TEXT AS (json_extract(container, '$.Labels."uncloud.service.id"')),

internal/machine/store/store.go

@@ -120,7 +120,14 @@ func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-cha
 			select {
 			case <-ctx.Done():
 				return
-			case <-events:
+			case _, ok := <-events:
+				if !ok {
+					// events channel has been closed.
+					if sub.Err() != nil {
+						slog.Error("Machines subscription failed.", "id", sub.ID(), "err", sub.Err())
+					}
+					return
+				}
 				// Just signal that there is a change in the machines list.
 				changes <- struct{}{}
 			}