feat: Add agent sync

.env.example

@@ -8,3 +8,4 @@ APP_SECRET= # Use: openssl rand -base64 42
 HUB_URL=https://... # Public URL of the Hub, should be the same as APP_URL
 
 AUTH_TOKEN= # Create a token in the Hub UI and use it here
+DEPLOYMENTS_DIR=/app/deployments # Directory where the agent stores compose files before running them

backend/internal/agent/agent.go

@@ -31,13 +31,14 @@ type FatalConfigError struct {
 func (e *FatalConfigError) Error() string { return e.Msg }
 
 type Config struct {
-	LogLevel     zerolog.Level
-	LogJSON      bool
-	HubUrl       string
-	AuthToken    string
-	AgentID      string
-	HubPublicKey ed25519.PublicKey
-	HealthPort   string
+	LogLevel       zerolog.Level
+	LogJSON        bool
+	HubUrl         string
+	AuthToken      string
+	AgentID        string
+	HubPublicKey   ed25519.PublicKey
+	HealthPort     string
+	DeploymentsDir string
 }
 
 func DefaultConfig() (Config, error) {
@@ -75,14 +76,20 @@ func DefaultConfig() (Config, error) {
 		healthPort = "8090"
 	}
 
+	deploymentsDir := os.Getenv("DEPLOYMENTS_DIR")
+	if deploymentsDir == "" {
+		deploymentsDir = "/app/deployments"
+	}
+
 	return Config{
-		LogLevel:     logLevel,
-		LogJSON:      logJSON,
-		HubUrl:       hubUrl,
-		AuthToken:    authToken,
-		AgentID:      agentID,
-		HubPublicKey: hubPublicKey,
-		HealthPort:   healthPort,
+		LogLevel:       logLevel,
+		LogJSON:        logJSON,
+		HubUrl:         hubUrl,
+		AuthToken:      authToken,
+		AgentID:        agentID,
+		HubPublicKey:   hubPublicKey,
+		HealthPort:     healthPort,
+		DeploymentsDir: deploymentsDir,
 	}, nil
 }
 
@@ -158,7 +165,7 @@ func Run(cfg Config) error {
 
 	Log.Info().Str("version", version.Version).Msg("agent started")
 
-	dockerClient, err := docker.New(Log)
+	dockerClient, err := docker.New(Log, cfg.DeploymentsDir)
 	if err != nil {
 		return fmt.Errorf("docker init: %w", err)
 	}
@@ -183,6 +190,7 @@ func Run(cfg Config) error {
 		return err
 	}
 	wsConnected.Store(true)
+	sender := newMessageSender(conn, session)
 
 	for {
 		_, data, readErr := conn.ReadMessage()
@@ -202,13 +210,14 @@ func Run(cfg Config) error {
 				return err
 			}
 			wsConnected.Store(true)
+			sender = newMessageSender(conn, session)
 			continue
 		}
 		msg := &messages.ServerMessage{}
 		if err := proto.Unmarshal(data, msg); err != nil {
 			Log.Error().Err(err).Msg("unmarshal error")
 			continue
 		}
-		handleServerMessage(msg, conn, session)
+		handleServerMessage(ctx, msg, session, sender, dockerClient)
 	}
 }

backend/internal/agent/agent_config_test.go

@@ -68,6 +68,9 @@ func TestDefaultConfig_Valid(t *testing.T) {
 	if len(cfg.HubPublicKey) != ed25519.PublicKeySize {
 		t.Errorf("HubPublicKey length = %d, want %d", len(cfg.HubPublicKey), ed25519.PublicKeySize)
 	}
+	if cfg.DeploymentsDir != "/app/deployments" {
+		t.Errorf("DeploymentsDir = %q, want %q", cfg.DeploymentsDir, "/app/deployments")
+	}
 }
 
 func TestDefaultConfig_Defaults(t *testing.T) {
@@ -88,6 +91,9 @@ func TestDefaultConfig_Defaults(t *testing.T) {
 	if cfg.LogJSON {
 		t.Error("LogJSON = true, want false by default")
 	}
+	if cfg.DeploymentsDir != "/app/deployments" {
+		t.Errorf("DeploymentsDir = %q, want %q", cfg.DeploymentsDir, "/app/deployments")
+	}
 }
 
 func TestDefaultConfig_LogLevels(t *testing.T) {
@@ -151,6 +157,22 @@ func TestDefaultConfig_LogJSON(t *testing.T) {
 	}
 }
 
+func TestDefaultConfig_DeploymentsDirOverride(t *testing.T) {
+	token, _ := makeTestToken(t, "test-agent-id")
+	t.Setenv("HUB_URL", "https://hub.example.com")
+	t.Setenv("AUTH_TOKEN", token)
+	t.Setenv("DEPLOYMENTS_DIR", "/srv/orca/deployments")
+
+	cfg, err := DefaultConfig()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if cfg.DeploymentsDir != "/srv/orca/deployments" {
+		t.Errorf("DeploymentsDir = %q, want %q", cfg.DeploymentsDir, "/srv/orca/deployments")
+	}
+}
+
 func TestParseTokenClaims_MissingHubPublicKey(t *testing.T) {
 	// JWT without hub_pubkey claim — must be rejected.
 	pub, priv, err := ed25519.GenerateKey(rand.Reader)

backend/internal/agent/docker/deploy.go

@@ -0,0 +1,90 @@
+package docker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	composetypes "github.com/compose-spec/compose-go/v2/types"
+	"github.com/docker/compose/v5/pkg/api"
+)
+
+const (
+	composeFileName   = "compose.yaml"
+	deployWaitTimeout = 2 * time.Minute
+)
+
+type DeployRequest struct {
+	ApplicationID   string
+	ApplicationName string
+	ComposeFile     string
+}
+
+var loadProject = func(ctx context.Context, composeService api.Compose, options api.ProjectLoadOptions) (*composetypes.Project, error) {
+	return composeService.LoadProject(ctx, options)
+}
+
+var upProject = func(ctx context.Context, composeService api.Compose, project *composetypes.Project, options api.UpOptions) error {
+	return composeService.Up(ctx, project, options)
+}
+
+func (c *Client) Deploy(ctx context.Context, req DeployRequest) error {
+	if c.compose == nil {
+		return errors.New("docker compose service is not initialized")
+	}
+	if c.deploymentsDir == "" {
+		return errors.New("deployments directory is not configured")
+	}
+	if !c.Ready() {
+		return errors.New("docker daemon is not ready")
+	}
+
+	if req.ComposeFile == "" {
+		return errors.New("compose file is empty")
+	}
+
+	applicationDir := filepath.Join(c.deploymentsDir, req.ApplicationName)
+	composePath := filepath.Join(applicationDir, composeFileName)
+
+	if err := os.MkdirAll(applicationDir, 0o755); err != nil {
+		return fmt.Errorf("create deployment directory: %w", err)
+	}
+	if err := os.WriteFile(composePath, []byte(req.ComposeFile), 0o600); err != nil {
+		return fmt.Errorf("write compose file: %w", err)
+	}
+
+	project, err := loadProject(ctx, c.compose, api.ProjectLoadOptions{
+		ProjectName: "orca-" + strings.ToLower(req.ApplicationName),
+		ConfigPaths: []string{composePath},
+		WorkingDir:  applicationDir,
+	})
+	if err != nil {
+		return fmt.Errorf("load compose project: %w", err)
+	}
+
+	if err := upProject(ctx, c.compose, project, api.UpOptions{
+		Create: api.CreateOptions{
+			RemoveOrphans:        true,
+			Recreate:             api.RecreateDiverged,
+			RecreateDependencies: api.RecreateDiverged,
+		},
+		Start: api.StartOptions{
+			Wait:        true,
+			WaitTimeout: deployWaitTimeout,
+		},
+	}); err != nil {
+		return fmt.Errorf("compose up: %w", err)
+	}
+
+	c.log.Info().
+		Str("application_id", req.ApplicationID).
+		Str("application_name", req.ApplicationName).
+		Str("compose_path", composePath).
+		Msg("deployment completed")
+
+	return nil
+}

backend/internal/agent/docker/deploy_test.go

@@ -0,0 +1,93 @@
+package docker
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	composetypes "github.com/compose-spec/compose-go/v2/types"
+	"github.com/docker/compose/v5/pkg/api"
+)
+
+func TestDeploy_WritesComposeFileAndRunsComposeUp(t *testing.T) {
+	c := newTestClient(t)
+	c.deploymentsDir = t.TempDir()
+
+	originalLoadProject := loadProject
+	originalUpProject := upProject
+	t.Cleanup(func() {
+		loadProject = originalLoadProject
+		upProject = originalUpProject
+	})
+
+	var gotLoadOptions api.ProjectLoadOptions
+	loadProject = func(_ context.Context, _ api.Compose, options api.ProjectLoadOptions) (*composetypes.Project, error) {
+		gotLoadOptions = options
+		return &composetypes.Project{Name: options.ProjectName}, nil
+	}
+
+	var gotUpOptions api.UpOptions
+	upProject = func(_ context.Context, _ api.Compose, project *composetypes.Project, options api.UpOptions) error {
+		if project.Name != "orca-billing" {
+			t.Fatalf("expected project name %q, got %q", "orca-billing", project.Name)
+		}
+		gotUpOptions = options
+		return nil
+	}
+
+	req := DeployRequest{
+		ApplicationID:   "app-123",
+		ApplicationName: "billing",
+		ComposeFile:     "services:\n  app:\n    image: ghcr.io/orcacd/app:1.0.0\n",
+	}
+
+	if err := c.Deploy(t.Context(), req); err != nil {
+		t.Fatalf("Deploy: %v", err)
+	}
+
+	composePath := filepath.Join(c.deploymentsDir, req.ApplicationName, composeFileName)
+	content, err := os.ReadFile(composePath)
+	if err != nil {
+		t.Fatalf("ReadFile: %v", err)
+	}
+	if string(content) != req.ComposeFile {
+		t.Fatalf("expected compose file to be written to deployment volume")
+	}
+
+	if gotLoadOptions.ProjectName != "orca-billing" {
+		t.Fatalf("expected load project name %q, got %q", "orca-billing", gotLoadOptions.ProjectName)
+	}
+	if len(gotLoadOptions.ConfigPaths) != 1 || gotLoadOptions.ConfigPaths[0] != composePath {
+		t.Fatalf("unexpected config paths: %#v", gotLoadOptions.ConfigPaths)
+	}
+	if gotLoadOptions.WorkingDir != filepath.Join(c.deploymentsDir, req.ApplicationName) {
+		t.Fatalf("expected working dir %q, got %q", filepath.Join(c.deploymentsDir, req.ApplicationName), gotLoadOptions.WorkingDir)
+	}
+
+	if !gotUpOptions.Start.Wait {
+		t.Fatal("expected compose up to wait for services to become ready")
+	}
+	if gotUpOptions.Start.WaitTimeout != deployWaitTimeout {
+		t.Fatalf("expected wait timeout %s, got %s", deployWaitTimeout, gotUpOptions.Start.WaitTimeout)
+	}
+	if !gotUpOptions.Create.RemoveOrphans {
+		t.Fatal("expected compose up to remove orphaned containers")
+	}
+	if gotUpOptions.Create.Recreate != api.RecreateDiverged {
+		t.Fatalf("expected recreate strategy %q, got %q", api.RecreateDiverged, gotUpOptions.Create.Recreate)
+	}
+}
+
+func TestDeploy_RejectsUnsafeApplicationID(t *testing.T) {
+	c := newTestClient(t)
+	c.deploymentsDir = t.TempDir()
+
+	err := c.Deploy(t.Context(), DeployRequest{
+		ApplicationID: "bad/id",
+		ComposeFile:   "services: {}\n",
+	})
+	if err == nil {
+		t.Fatal("expected deploy to reject unsafe application ids")
+	}
+}

backend/internal/agent/docker/docker.go

@@ -17,16 +17,17 @@ import (
 const daemonCheckInterval = 5 * time.Second
 
 type Client struct {
-	mu      sync.RWMutex // protects ready
-	log     zerolog.Logger
-	cli     command.Cli
-	compose api.Compose
-	ready   bool
-	ctx     context.Context
-	cancel  context.CancelFunc
+	mu             sync.RWMutex // protects ready
+	log            zerolog.Logger
+	cli            command.Cli
+	compose        api.Compose
+	deploymentsDir string
+	ready          bool
+	ctx            context.Context
+	cancel         context.CancelFunc
 }
 
-func New(log zerolog.Logger) (*Client, error) {
+func New(log zerolog.Logger, deploymentsDir string) (*Client, error) {
 	dockerCLI, err := command.NewDockerCli()
 	if err != nil {
 		return nil, err
@@ -43,11 +44,12 @@ func New(log zerolog.Logger) (*Client, error) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	c := &Client{
-		log:     log,
-		cli:     dockerCLI,
-		compose: composeSvc,
-		ctx:     ctx,
-		cancel:  cancel,
+		log:            log,
+		cli:            dockerCLI,
+		compose:        composeSvc,
+		deploymentsDir: deploymentsDir,
+		ctx:            ctx,
+		cancel:         cancel,
 	}
 
 	if c.pingDaemon() {

backend/internal/agent/docker/docker_test.go

@@ -8,7 +8,7 @@ import (
 
 func newTestClient(t *testing.T) *Client {
 	t.Helper()
-	c, err := New(zerolog.Nop())
+	c, err := New(zerolog.Nop(), t.TempDir())
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -50,7 +50,7 @@ func TestPingDaemon(t *testing.T) {
 func TestNew_DaemonUnreachable(t *testing.T) {
 	t.Setenv("DOCKER_HOST", "tcp://localhost:1")
 
-	c, err := New(zerolog.Nop())
+	c, err := New(zerolog.Nop(), t.TempDir())
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -62,7 +62,7 @@ func TestNew_DaemonUnreachable(t *testing.T) {
 func TestPingDaemon_Unreachable(t *testing.T) {
 	t.Setenv("DOCKER_HOST", "tcp://localhost:1")
 
-	c, err := New(zerolog.Nop())
+	c, err := New(zerolog.Nop(), t.TempDir())
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}