bri-sandbox-development-platform/agent-micro/internal/deployer/deployer.go

// Package deployer is the per-deployment state machine. One Deployer is
// created per DeployRequest; it runs the stages sequentially and emits
// events on its output channel. The caller (the agent) reads events and
// fans them out to the control plane.
//
// Runtime model (per REQUIREMENTS.md):
//   1. git fetch / checkout / pull on the host
//   2. go build -o <bin> on the host (the VM has the Go toolchain)
//   3. docker run alpine:3.20 with -v <repo>:/src and command /src/<bin>
//      — NO Dockerfile, NO image build step. The alpine image must be
//      loaded on the host ahead of time (manual docker load).
//   4. environment variables from the Environment config are passed via
//      -e flags into the container.
package deployer

import (
	"context"
	"fmt"
	"os/exec"
	"path/filepath"
	"strings"
	"time"

	"github.com/docker/docker/api/types/container"
	dockerclient "github.com/docker/docker/client"
	"github.com/docker/go-connections/nat"

	"github.com/sdp/agent-micro/internal/gitutil"
	"github.com/sdp/protocol"
)

type Deployer struct {
	ID         string
	Repository string
	RepoPath   string
	Branch     string
	Env        map[string]string
	Creds      gitutil.Creds
	cli        *dockerclient.Client
	binName    string
	imageTag   string
	containerID string
}

func New(cli *dockerclient.Client, id, repo, path, branch string, env map[string]string, c gitutil.Creds) *Deployer {
	return &Deployer{
		ID:         id,
		Repository: repo,
		RepoPath:   path,
		Branch:     branch,
		Env:        env,
		Creds:      c,
		cli:        cli,
		binName:    "app-" + repo,
		imageTag:   "alpine:3.20", // ponytail: pre-loaded on the VM. Swap to a configurable tag.
	}
}

// Run executes the full pipeline. It pushes events to out and returns the
// final state (RUNNING on success, FAILED otherwise).
func (d *Deployer) Run(ctx context.Context, out chan<- protocol.Event) string {
	emit := func(kind, state, stage, line string) {
		out <- protocol.Event{
			DeploymentID: d.ID,
			Kind:         kind,
			State:        state,
			Stage:        stage,
			Line:         line,
			At:           time.Now().UnixMilli(),
		}
	}

	stages := []struct {
		name string
		fn   func() error
	}{
		{"git fetch", func() error { _, err := gitutil.Fetch(ctx, d.RepoPath, d.Creds); return err }},
		{"git checkout", func() error { _, err := gitutil.Checkout(ctx, d.RepoPath, d.Branch, d.Creds); return err }},
		{"git pull", func() error { _, err := gitutil.Pull(ctx, d.RepoPath, d.Creds); return err }},
		{"go build", d.build},
		{"start container", d.startContainer},
	}
	for _, s := range stages {
		emit("progress", "", s.name, "starting "+s.name)
		if err := s.fn(); err != nil {
			emit("progress", "FAILED", s.name, err.Error())
			emit("status", "FAILED", "", "")
			return "FAILED"
		}
		emit("progress", "", s.name, "ok")
	}
	emit("status", "RUNNING", "", "")
	return "RUNNING"
}

func (d *Deployer) build(ctx context.Context) error {
	binPath := filepath.Join(d.RepoPath, d.binName)
	cmd := exec.CommandContext(ctx, "go", "build", "-o", binPath, "./...")
	cmd.Dir = d.RepoPath
	out, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("go build: %w: %s", err, out)
	}
	return nil
}

// startContainer runs `alpine:3.20` with the host repo dir bind-mounted at
// /src and the binary executed as the container command. The container's
// working dir is set to the bind mount so the binary finds its config files.
func (d *Deployer) startContainer(ctx context.Context) error {

	envList := make([]string, 0, len(d.Env))
	for k, v := range d.Env {
		envList = append(envList, k+"="+v)
	}

	cfg := &container.Config{
		Image: d.imageTag,
		Cmd:   []string{"/src/" + d.binName},
		Env:   envList,
		WorkingDir: "/src",
		ExposedPorts: nat.PortSet{
			"8080/tcp": struct{}{},
		},
	}
	hostCfg := &container.HostConfig{
		Binds: []string{d.RepoPath + ":/src"},
		RestartPolicy: container.RestartPolicy{Name: "unless-stopped"},
	}

	resp, err := d.cli.ContainerCreate(ctx, cfg, hostCfg, nil, nil, d.containerName())
	if err != nil {
		return fmt.Errorf("container create: %w", err)
	}
	d.containerID = resp.ID
	if err := d.cli.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
		return fmt.Errorf("container start: %w", err)
	}
	return nil
}

func (d *Deployer) containerName() string {
	return "sdp-" + d.Repository + "-" + d.ID
}

func (d *Deployer) Stop(ctx context.Context) error {
	if d.containerID == "" {
		return nil
	}
	return d.cli.ContainerStop(ctx, d.containerID, container.StopOptions{})
}

// StreamLogs tails the container's logs into the events channel until ctx
// is cancelled.
func (d *Deployer) StreamLogs(ctx context.Context, out chan<- protocol.Event) {
	if d.containerID == "" {
		return
	}
	rc, err := d.cli.ContainerLogs(ctx, d.containerID, container.LogsOptions{
		ShowStdout: true,
		ShowStderr: true,
		Follow:     true,
		Timestamps: false,
	})
	if err != nil {
		return
	}
	defer rc.Close()
	// ponytail: Docker multiplexes stdout/stderr; the leading 8 bytes per
	// frame are the stream header. For MVP we just split on newlines and
	// don't try to demux.
	buf := make([]byte, 4096)
	var carry strings.Builder
	for {
		n, err := rc.Read(buf)
		if n > 0 {
			carry.Write(buf[:n])
			for {
				s := carry.String()
				i := strings.IndexByte(s, '\n')
				if i < 0 {
					break
				}
				line := s[:i]
				carry.Reset()
				carry.WriteString(s[i+1:])
				if len(line) > 8 && line[0] <= 2 {
					line = line[8:]
				}
				if line != "" {
					out <- protocol.Event{
						DeploymentID: d.ID,
						Kind:         "log",
						Line:         line,
						At:           time.Now().UnixMilli(),
					}
				}
			}
		}
		if err != nil {
			return
		}
	}
}