package-updater/internal/agent/updater/compose.go

package updater

import (
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"slices"
	"strings"
	"time"

	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/filters"
)

const (
	labelComposeProject = "com.docker.compose.project"
	labelComposeService = "com.docker.compose.service"
	labelWorkingDir     = "com.docker.compose.project.working_dir"
	labelConfigFiles    = "com.docker.compose.project.config_files"
	labelEnvFile        = "com.docker.compose.project.environment_file"
)

// errorOutputTail is how much of a failed command's output goes into the
// job's error message; the full output is in the job log.
const errorOutputTail = 2 << 10

// tailWriter adapts a tailBuffer to io.Writer.
type tailWriter struct{ b *tailBuffer }

func (w tailWriter) Write(p []byte) (int, error) {
	w.b.Write(p)
	return len(p), nil
}

// CommandRunner runs "docker <args>" in dir with stdout and stderr to out.
type CommandRunner func(ctx context.Context, dir string, out io.Writer, args ...string) error

// ExecDocker is the real CommandRunner. When ctx ends the process is killed
// and, after a grace period, its output pipes are closed.
func ExecDocker(ctx context.Context, dir string, out io.Writer, args ...string) error {
	cmd := exec.CommandContext(ctx, "docker", args...)
	cmd.Dir = dir
	cmd.Stdout = out
	cmd.Stderr = out
	cmd.Env = dockerEnv()
	cmd.WaitDelay = 10 * time.Second
	return cmd.Run()
}

// dockerEnvVars are the only variables from the agent's environment that
// reach docker compose. Compose prefers process variables over the stack's
// .env file when it interpolates, so passing the agent's own (HOSTNAME,
// AGENT_TOKEN, ...) would change stack configuration and expose the token.
var dockerEnvVars = []string{"PATH", "HOME", "DOCKER_HOST", "DOCKER_CONFIG", "DOCKER_TLS_VERIFY", "DOCKER_CERT_PATH"}

// dockerEnv returns the allowed variables that are set in the agent's
// environment, as KEY=value pairs.
func dockerEnv() []string {
	var env []string
	for _, k := range dockerEnvVars {
		if v := os.Getenv(k); v != "" {
			env = append(env, k+"="+v)
		}
	}
	return env
}

// ComposeDocker is the subset of the Docker SDK ComposeExecutor needs.
type ComposeDocker interface {
	SelfInspector
	HelperDocker
	ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error)
}

// ComposeExecutor updates a Compose service by writing the agent-owned
// override file and running `docker compose pull` and `up -d`.
type ComposeExecutor struct {
	cli     ComposeDocker
	resolve *Resolver
	selfID  string
	run     CommandRunner
	build   *BuildChecker
}

// NewComposeExecutor returns an executor for Compose jobs. selfID names the
// agent's own container, whose mounts decide which directories may be
// touched; run and runOut are ExecDocker and ExecDockerOutput outside tests.
func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run CommandRunner, runOut OutputRunner) *ComposeExecutor {
	return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run, build: NewBuildChecker(cli, selfID, runOut)}
}

// Execute pins job.Ref to its current digest (a ref that has one is used as
// is), writes it to the stack's override file and recreates the service.
// Working directory and config files come from the service's container
// labels on this machine, never from the job. It returns the pinned ref.
//
// When the service is the agent's own container, the recreate runs in a
// helper container instead (see selfUpdate), because it stops this process.
func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (string, error) {
	project, service := job.ComposeProject, job.ComposeService
	cts, err := e.cli.ContainerList(ctx, container.ListOptions{
		All: true,
		Filters: filters.NewArgs(
			filters.Arg("label", labelComposeProject+"="+project),
			filters.Arg("label", labelComposeService+"="+service),
		),
	})
	if err != nil {
		return "", fmt.Errorf("list containers: %w", err)
	}
	if len(cts) == 0 {
		return "", fmt.Errorf("no container for compose service %s/%s on this machine", project, service)
	}
	self, err := inspectSelf(ctx, e.cli, e.selfID)
	if err != nil {
		return "", err
	}
	isSelf := self.ContainerJSONBase != nil && self.ID != "" &&
		slices.ContainsFunc(cts, func(ct container.Summary) bool { return ct.ID == self.ID })
	// What an update of the agent's own service has printed so far goes into
	// the helper's labels.
	var soFar *tailBuffer
	if isSelf {
		soFar = &tailBuffer{max: selfUpdateOutputMax}
		out = io.MultiWriter(out, tailWriter{soFar})
	}
	workingDir := cts[0].Labels[labelWorkingDir]
	configFiles := cts[0].Labels[labelConfigFiles]
	envFileLabel := cts[0].Labels[labelEnvFile]

	roots := rootsOf(self)
	files, dir, err := composeFiles(roots, configFiles)
	if err != nil {
		return "", err
	}
	if workingDir == "" {
		workingDir = dir
	} else if !insideAny(roots, workingDir) {
		return "", notMounted(workingDir)
	}

	envs, err := envFiles(roots, envFileLabel)
	if err != nil {
		return "", err
	}
	// Fresh check, not the inventory's cache: a build: section may have been
	// added since. Nothing is written for such a service.
	built, err := e.build.builtServices(ctx, project, workingDir, files[:len(files)-1], envs)
	if err != nil {
		return "", err
	}
	isBuilt, known := built[service]
	if !known {
		return "", fmt.Errorf("service %s not found in compose config", service)
	}
	if isBuilt {
		return "", fmt.Errorf("service %s is built from source (build:), not updated", service)
	}

	pinned, err := e.resolve.Resolve(ctx, job.Ref)
	if err != nil {
		return "", err
	}
	fmt.Fprintln(out, resolveNote(job.Ref, pinned))

	// Giving the override the stack's owner is best effort: a failure is only
	// a warning, the update itself is fine.
	_, ownerErr, err := WriteOverride(dir, service, pinned, files[0])
	if err != nil {
		return "", err
	}
	if ownerErr != nil {
		fmt.Fprintf(out, "warning: could not give the override the stack's owner: %v\n", ownerErr)
	}

	base := composeBase(project, workingDir, files, envs)
	run := func(args []string) error {
		fmt.Fprintf(out, "$ docker %s\n", strings.Join(args, " "))
		tail := &tailBuffer{max: errorOutputTail}
		if err := e.run(ctx, workingDir, io.MultiWriter(out, tailWriter{tail}), args...); err != nil {
			return composeError(args[len(base)], err, tail)
		}
		return nil
	}
	if err := run(append(append([]string(nil), base...), "pull", service)); err != nil {
		return "", err
	}
	up := append(append([]string(nil), base...), "up", "-d", service)
	if isSelf {
		return e.selfUpdate(ctx, job, self, workingDir, up, pinned, out, soFar)
	}
	if err := run(up); err != nil {
		return "", err
	}
	return pinned, nil
}

// composeError describes a failed `docker compose <step>` call with the tail
// of its output.
func composeError(step string, err error, tail *tailBuffer) error {
	return fmt.Errorf("docker compose %s: %w (output: %s)", step, err, strings.TrimSpace(tail.String()))
}