panel/agent/internal/runtime/docker.go

package runtime

import (
	"archive/tar"
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/build"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/image"
	"github.com/docker/docker/api/types/mount"
	"github.com/docker/docker/api/types/network"
	"github.com/docker/docker/api/types/volume"
	"github.com/docker/docker/client"
	"github.com/docker/docker/pkg/stdcopy"
	"github.com/moby/go-archive"
	"github.com/docker/go-connections/nat"
)

// DockerRuntime launches instances as Docker containers on the local daemon.
// Container IDs (the string returned by Create) are Docker's full container IDs.
type DockerRuntime struct {
	cli *client.Client
	// hostNetUnsupported is true when the daemon is Docker Desktop
	// (Windows/macOS), where NetworkMode=host binds inside the hidden
	// Linux VM and is NOT reachable from the host LAN. On such daemons we
	// fall back to bridge networking + published ports so the panel's
	// allocated ports actually reach players. Detected once at startup;
	// overridable via PANEL_FORCE_BRIDGE=1 / PANEL_FORCE_HOST_NET=1.
	hostNetUnsupported bool
}

// NewDocker connects to the local Docker daemon using environment-configured
// settings (DOCKER_HOST, DOCKER_CERT_PATH, …) and negotiates an API version.
func NewDocker() (*DockerRuntime, error) {
	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
	if err != nil {
		return nil, fmt.Errorf("docker client: %w", err)
	}
	r := &DockerRuntime{cli: cli}
	r.hostNetUnsupported = detectHostNetUnsupported(cli)
	return r, nil
}

// applyHostNetPolicy decides the effective network mode + port plumbing for a
// container. Native Linux keeps host mode and drops the (meaningless) port
// bindings; a Docker-Desktop daemon converts host mode to bridge and KEEPS the
// published bindings so the panel-allocated ports reach the host LAN. Non-host
// modes pass through unchanged. Extracted for testability.
func applyHostNetPolicy(r *DockerRuntime, requested string, exposed nat.PortSet, bindings nat.PortMap) (string, nat.PortSet, nat.PortMap) {
	if requested != "host" {
		return requested, exposed, bindings
	}
	if r.hostNetUnsupported {
		return "", exposed, bindings // bridge + published ports
	}
	return "host", nil, nil // native host mode, no bindings
}

// detectHostNetUnsupported returns true when NetworkMode=host won't reach the
// host LAN — i.e. Docker Desktop's Linux VM on Windows/macOS. Explicit env
// overrides win over autodetection so operators can force either behavior.
func detectHostNetUnsupported(cli *client.Client) bool {
	if os.Getenv("PANEL_FORCE_BRIDGE") == "1" {
		return true
	}
	if os.Getenv("PANEL_FORCE_HOST_NET") == "1" {
		return false
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	info, err := cli.Info(ctx)
	if err != nil {
		return false // can't tell → assume native Linux (host works)
	}
	// Docker Desktop reports OperatingSystem "Docker Desktop" and, on
	// Windows hosts, OSType "linux" while the actual client OS is windows.
	// The tell-tale is the "Docker Desktop" marker or a non-Linux daemon.
	if strings.Contains(info.OperatingSystem, "Docker Desktop") {
		return true
	}
	if info.OSType != "" && info.OSType != "linux" {
		return true
	}
	return false
}

// Close releases daemon resources.
func (r *DockerRuntime) Close() error {
	if r.cli == nil {
		return nil
	}
	return r.cli.Close()
}

// Name identifies the runtime kind.
func (r *DockerRuntime) Name() string { return "docker" }

// Ping verifies we can reach the Docker daemon.
func (r *DockerRuntime) Ping(ctx context.Context) error {
	_, err := r.cli.Ping(ctx)
	return err
}

// Create pulls the image (if missing locally) and creates the container.
// The container is NOT started — call Start afterward.
func (r *DockerRuntime) Create(ctx context.Context, spec InstanceSpec) (string, error) {
	if spec.InstanceID == "" {
		return "", errors.New("spec.InstanceID is required")
	}
	if spec.Image == "" {
		return "", errors.New("spec.Image is required")
	}

	if err := r.acquireImage(ctx, spec.Image, spec.BuildContext, spec.LogSink); err != nil {
		return "", err
	}

	exposed, bindings, err := buildPortBindings(spec.Ports)
	if err != nil {
		return "", err
	}
	// Docker rejects PortBindings + ExposedPorts when NetworkMode=host —
	// the container shares the host's network namespace so publish-port
	// plumbing is meaningless there.
	var netMode string
	netMode, exposed, bindings = applyHostNetPolicy(r, spec.NetworkMode, exposed, bindings)
	if spec.NetworkMode == "host" && netMode != "host" && len(bindings) > 0 && spec.LogSink != nil {
		spec.LogSink("Docker Desktop detected: publishing ports via bridge instead of host networking (host mode is not LAN-reachable on Docker Desktop)")
	}

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

	mounts := make([]mount.Mount, 0, len(spec.Volumes))
	for _, v := range spec.Volumes {
		m := mount.Mount{
			Target:   v.ContainerPath,
			ReadOnly: v.ReadOnly,
		}
		switch v.Type {
		case "volume":
			if v.VolumeName == "" {
				return "", fmt.Errorf("volume mount %q: Name required", v.ContainerPath)
			}
			if err := r.ensureVolume(ctx, v.VolumeName, spec.InstanceID); err != nil {
				return "", err
			}
			m.Type = mount.TypeVolume
			m.Source = v.VolumeName
		default: // "" or "bind"
			if v.HostPath == "" {
				return "", fmt.Errorf("bind mount %q: HostPath required", v.ContainerPath)
			}
			// Docker CLI auto-creates missing bind sources, but the Go API
			// does not — it errors with "bind source path does not exist"
			// and refuses to start the container. Modules declaring subdir
			// binds like `$DATA_PATH/logs` rely on the source existing.
			// Two cases:
			//   - Path exists as a file (panel-rendered config like
			//     serverconfig.xml.rendered): leave it alone, it's a
			//     file bind.
			//   - Path missing or already a directory: mkdir -p so docker
			//     bind-mount has something to mount.
			if st, err := os.Stat(v.HostPath); err == nil && !st.IsDir() {
				// existing file → file bind, no mkdir
			} else if err := os.MkdirAll(v.HostPath, 0o755); err != nil {
				return "", fmt.Errorf("bind mount %q: mkdir source %q: %w", v.ContainerPath, v.HostPath, err)
			}
			m.Type = mount.TypeBind
			m.Source = v.HostPath
		}
		mounts = append(mounts, m)
	}

	restart := spec.RestartPolicy
	if restart == "" {
		restart = "unless-stopped"
	}

	hostCfg := &container.HostConfig{
		Mounts:       mounts,
		PortBindings: bindings,
		NetworkMode:  container.NetworkMode(netMode),
		SecurityOpt:  spec.SecurityOpts,
		RestartPolicy: container.RestartPolicy{
			Name: container.RestartPolicyMode(restart),
		},
		Resources: container.Resources{
			CPUShares: int64(spec.Resources.CPUShares),
			Memory:    int64(spec.Resources.MemoryBytes),
			NanoCPUs:  spec.Resources.NanoCPUs,
			PidsLimit: pidsPtr(spec.Resources.PidsLimit),
		},
	}

	cfg := &container.Config{
		Image:        spec.Image,
		Env:          envSlice,
		Entrypoint:   spec.Entrypoint,
		Cmd:          spec.Command,
		User:         spec.User,
		WorkingDir:   spec.WorkingDir,
		ExposedPorts: exposed,
		Tty:          false,
		// OpenStdin + StdinOnce=false = stdin stays open across reconnects.
		// The stdio RCON adapter attaches to stdin later via ContainerAttach.
		OpenStdin: spec.OpenStdin,
		StdinOnce: false,
		Labels: map[string]string{
			"panel.instance_id": spec.InstanceID,
			"panel.role":        roleLabel(spec.IsSidecar),
		},
	}

	name := "panel-" + spec.InstanceID
	resp, err := r.cli.ContainerCreate(ctx, cfg, hostCfg, &network.NetworkingConfig{}, nil, name)
	if err != nil {
		return "", fmt.Errorf("container create: %w", err)
	}
	return resp.ID, nil
}

// Start starts a previously-created container.
func (r *DockerRuntime) Start(ctx context.Context, id string) error {
	if err := r.cli.ContainerStart(ctx, id, container.StartOptions{}); err != nil {
		return fmt.Errorf("container start: %w", err)
	}
	return nil
}

// Stop stops the container, waiting up to grace before SIGKILL.
func (r *DockerRuntime) Stop(ctx context.Context, id string, grace time.Duration) error {
	secs := int(grace.Seconds())
	if secs < 0 {
		secs = 0
	}
	opts := container.StopOptions{Timeout: &secs}
	if err := r.cli.ContainerStop(ctx, id, opts); err != nil {
		return fmt.Errorf("container stop: %w", err)
	}
	return nil
}

// Restart restarts the container. Same StopOptions semantics as Stop —
// engine waits up to grace seconds before SIGKILL, then the container is
// started fresh. The container's RestartPolicy is irrelevant here;
// ContainerRestart always brings it back regardless of policy.
func (r *DockerRuntime) Restart(ctx context.Context, id string, grace time.Duration) error {
	secs := int(grace.Seconds())
	if secs < 0 {
		secs = 0
	}
	opts := container.StopOptions{Timeout: &secs}
	if err := r.cli.ContainerRestart(ctx, id, opts); err != nil {
		return fmt.Errorf("container restart: %w", err)
	}
	return nil
}

// Remove removes the container. Force=true kills if still running.
func (r *DockerRuntime) Remove(ctx context.Context, id string) error {
	err := r.cli.ContainerRemove(ctx, id, container.RemoveOptions{
		Force:         true,
		RemoveVolumes: false,
	})
	if err != nil && !client.IsErrNotFound(err) {
		return fmt.Errorf("container remove: %w", err)
	}
	return nil
}

// StreamLogs follows stdout+stderr and invokes handler for each line.
// Blocks until ctx is cancelled, the container exits, or the connection errors.
func (r *DockerRuntime) StreamLogs(ctx context.Context, id string, handler LogHandler) error {
	rc, err := r.cli.ContainerLogs(ctx, id, container.LogsOptions{
		ShowStdout: true,
		ShowStderr: true,
		Follow:     true,
		// Timestamps:true prefixes each line with docker's own RFC3339Nano
		// time. We parse that prefix below and pass it as the line's `at`
		// instead of time.Now(). Critical for the readiness gate: the
		// docker log file persists across container restarts, so a Tail:400
		// replay can include the PREVIOUS run's "Server has completed
		// startup" line. Without docker's real timestamp on each line, the
		// frontend can't tell stale replays from current-run lines and
		// the pill flips to "running" the instant the buffer is fetched.
		Timestamps: true,
		// Start with a little backlog so the console isn't blank if someone
		// opens the modal mid-run. 400 lines is ~5s of chatter at the busiest
		// and is trivial to stream.
		Tail: "400",
	})
	if err != nil {
		return fmt.Errorf("container logs: %w", err)
	}
	defer rc.Close()

	outPr, outPw := io.Pipe()
	errPr, errPw := io.Pipe()
	defer outPw.Close()
	defer errPw.Close()

	done := make(chan error, 3)

	// Demux multiplexed docker log stream into separate stdout/stderr pipes.
	go func() {
		_, err := stdcopy.StdCopy(outPw, errPw, rc)
		_ = outPw.Close()
		_ = errPw.Close()
		done <- err
	}()

	scan := func(r io.Reader, stream LogStream) {
		sc := bufio.NewScanner(r)
		sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
		// Split on BOTH \n and \r so terminal-style progress updates
		// (SteamCMD "Update state ... progress: X%" uses \r to overwrite
		// the same line) show up as they happen instead of buffering
		// until the next real newline minutes later.
		sc.Split(scanLogLine)
		for sc.Scan() {
			raw := stripANSI(sc.Text())
			if raw == "" {
				continue
			}
			// Strip docker's RFC3339Nano timestamp prefix and use it as
			// the line's authoritative time. Format is exact:
			//   "2026-04-29T17:46:53.123456789Z <line content>"
			// On parse failure we fall back to time.Now() so a malformed
			// line still gets streamed (better wrong timestamp than
			// dropped log line).
			line, ts := splitDockerTimestamp(raw)
			handler(stream, line, ts)
		}
		done <- sc.Err()
	}
	go scan(outPr, LogStreamStdout)
	go scan(errPr, LogStreamStderr)

	for i := 0; i < 3; i++ {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case err := <-done:
			if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
				return err
			}
		}
	}
	return nil
}

// ListVolumesByInstance returns every volume labeled panel.instance_id=id.
// Used by the delete-with-purge path to clean up the instance's storage.
func (r *DockerRuntime) ListVolumesByInstance(ctx context.Context, instanceID string) ([]string, error) {
	lst, err := r.cli.VolumeList(ctx, volume.ListOptions{})
	if err != nil {
		return nil, fmt.Errorf("volume list: %w", err)
	}
	var names []string
	for _, v := range lst.Volumes {
		if v == nil {
			continue
		}
		if v.Labels["panel.instance_id"] == instanceID {
			names = append(names, v.Name)
		}
	}
	return names, nil
}

// RemoveVolume removes a Docker volume by name. force=true removes even
// if in-use references exist (shouldn't happen for us since we remove
// containers first).
func (r *DockerRuntime) RemoveVolume(ctx context.Context, name string, force bool) error {
	if err := r.cli.VolumeRemove(ctx, name, force); err != nil {
		return fmt.Errorf("volume remove %q: %w", name, err)
	}
	return nil
}

// ListContainersHoldingInstanceVolumes returns container IDs for any
// container (running OR exited) whose mount list references one of
// the panel volumes labeled for the given instance.
//
// Used by the delete path to sweep stranded sidecars — exited steamcmd
// or backup containers that the agent crashed before removing.
// They keep volumes "in use" for `docker volume rm` purposes even when
// they're not running, blocking instance delete-with-purge forever.
//
// We list ALL containers (including exited) and filter by mount name
// rather than relying on the container's own panel label, because
// updater sidecars don't carry the instance label — they're spawned
// by the alpine helper which doesn't know about panel.* labels.
func (r *DockerRuntime) ListContainersHoldingInstanceVolumes(ctx context.Context, instanceID string) ([]string, error) {
	// Build the set of volume names this instance owns.
	vols, err := r.ListVolumesByInstance(ctx, instanceID)
	if err != nil {
		return nil, err
	}
	if len(vols) == 0 {
		return nil, nil
	}
	wanted := make(map[string]bool, len(vols))
	for _, v := range vols {
		wanted[v] = true
	}
	// Cheap heuristic: panel-<id>- prefix on the container name catches
	// most sidecars (panel-X-steamcmd-N, panel-X-fs, panel-X-backup).
	// We still cross-check via mounts so renamed/old containers don't
	// slip through.
	prefix := "panel-" + instanceID + "-"

	cs, err := r.cli.ContainerList(ctx, container.ListOptions{All: true})
	if err != nil {
		return nil, fmt.Errorf("container list: %w", err)
	}
	out := make([]string, 0, len(cs))
	for _, c := range cs {
		// Match by mount.
		matched := false
		for _, m := range c.Mounts {
			if m.Type == mount.TypeVolume && wanted[m.Name] {
				matched = true
				break
			}
		}
		// Or by name prefix.
		if !matched {
			for _, n := range c.Names {
				name := strings.TrimPrefix(n, "/")
				if strings.HasPrefix(name, prefix) {
					matched = true
					break
				}
			}
		}
		if matched {
			out = append(out, c.ID)
		}
	}
	return out, nil
}

// roleLabel maps the sidecar flag to the "panel.role" label value.
func roleLabel(isSidecar bool) string {
	if isSidecar {
		return "sidecar"
	}
	return "instance"
}

// ListMainInstanceNames returns the instance IDs of every main
// game-server container (panel.role=instance), excluding the transient
// fs / steamcmd / backup / etc. helper sidecars. Used by Rehydrate to
// detect orphans: a main container with no matching sidecar metadata,
// which would otherwise be silently invisible to the agent.
//
// Filtering is by the "panel.role" label, set at create time, not by
// name-suffix heuristics — so new sidecar kinds can't be misclassified
// as instances. Containers predating the label (no panel.role) fall
// back to a name check: anything matching a known sidecar suffix is
// excluded, everything else under the panel- prefix is treated as a
// main instance (the conservative choice — better to surface a stray
// helper as a candidate orphan than to hide a real instance).
func (r *DockerRuntime) ListMainInstanceNames(ctx context.Context) ([]string, error) {
	cs, err := r.cli.ContainerList(ctx, container.ListOptions{All: true})
	if err != nil {
		return nil, fmt.Errorf("container list: %w", err)
	}
	out := make([]string, 0, len(cs))
	for _, c := range cs {
		// Prefer the explicit label.
		if role, ok := c.Labels["panel.role"]; ok {
			if role != "instance" {
				continue
			}
			if id := c.Labels["panel.instance_id"]; id != "" {
				out = append(out, id)
			}
			continue
		}
		// Legacy fallback for containers created before panel.role existed.
		for _, n := range c.Names {
			name := strings.TrimPrefix(n, "/")
			id, ok := strings.CutPrefix(name, "panel-")
			if !ok || legacyIsSidecarName(id) {
				continue
			}
			out = append(out, id)
		}
	}
	return out, nil
}

// legacyIsSidecarName classifies pre-"panel.role" containers by their
// reserved trailing role. Instance ids themselves can contain hyphens
// (e.g. "scorched-earth"), so we match known suffixes rather than
// splitting on "-".
func legacyIsSidecarName(afterPrefix string) bool {
	for _, suf := range []string{"-fs", "-backup", "-restore", "-wipe", "-basemods", "-warmseed"} {
		if strings.HasSuffix(afterPrefix, suf) {
			return true
		}
	}
	// steamcmd sidecars are panel-<id>-steamcmd-<n>.
	if strings.Contains(afterPrefix, "-steamcmd-") || strings.HasSuffix(afterPrefix, "-steamcmd") {
		return true
	}
	return false
}

// ForceRemoveContainer is `docker rm -f` — kills a running container
// and removes it. Used by the orphan-sidecar sweep on delete; the
// sidecar's exit-code state doesn't matter since we're nuking it.
func (r *DockerRuntime) ForceRemoveContainer(ctx context.Context, containerID string) error {
	return r.cli.ContainerRemove(ctx, containerID, container.RemoveOptions{
		Force:         true,
		RemoveVolumes: false, // we'll remove volumes separately
	})
}

// ensureVolume creates the named volume if it doesn't exist. Labels it so
// operators can identify panel-owned volumes (`docker volume ls -f
// label=panel.instance_id=ark-test`).
func (r *DockerRuntime) ensureVolume(ctx context.Context, name, instanceID string) error {
	_, err := r.cli.VolumeInspect(ctx, name)
	if err == nil {
		return nil
	}
	if !client.IsErrNotFound(err) {
		return fmt.Errorf("inspect volume %q: %w", name, err)
	}
	_, err = r.cli.VolumeCreate(ctx, volume.CreateOptions{
		Name:   name,
		Driver: "local",
		Labels: map[string]string{
			"panel.instance_id": instanceID,
		},
	})
	if err != nil {
		return fmt.Errorf("create volume %q: %w", name, err)
	}
	return nil
}

// InspectByName looks up a container by its Docker name (leading "/"
// stripped by the API) and returns its state. Used by the Agent on cold
// start to check whether a sidecar-metadata file still points at a live
// container.
func (r *DockerRuntime) InspectByName(ctx context.Context, name string) (RuntimeInstanceState, error) {
	c, err := r.cli.ContainerInspect(ctx, name)
	if err != nil {
		return RuntimeInstanceState{}, err
	}
	st := RuntimeInstanceState{ContainerID: c.ID}
	if c.State != nil {
		st.Status = c.State.Status
		st.ExitCode = int32(c.State.ExitCode)
	}
	return st, nil
}

// ContainerMounts returns the mount points of a container by name. Works
// on stopped containers. Used to resolve a 7DTD instance's /cluster bind
// to its host path for the cluster player-save snapshotter.
func (r *DockerRuntime) ContainerMounts(ctx context.Context, name string) ([]MountInfo, error) {
	c, err := r.cli.ContainerInspect(ctx, name)
	if err != nil {
		return nil, err
	}
	out := make([]MountInfo, 0, len(c.Mounts))
	for _, m := range c.Mounts {
		out = append(out, MountInfo{
			Type:        string(m.Type),
			Source:      m.Source,
			Destination: m.Destination,
			Name:        m.Name,
		})
	}
	return out, nil
}

// ContainerExists reports definitive existence vs. an indeterminate
// query failure. A not-found error from the daemon means the container
// is gone (exists=false, err=nil); any other error (daemon unreachable,
// connection dropped mid-call during agent shutdown, ctx cancelled)
// leaves existence UNKNOWN and is returned verbatim so callers don't
// mistake "couldn't ask" for "doesn't exist".
func (r *DockerRuntime) ContainerExists(ctx context.Context, name string) (bool, error) {
	_, err := r.cli.ContainerInspect(ctx, name)
	if err == nil {
		return true, nil
	}
	if client.IsErrNotFound(err) {
		return false, nil
	}
	return false, err
}

// StatsStream opens Docker's ContainerStats (streaming) and returns the
// raw JSON reader. Each line is a stats frame — caller parses + throttles.
func (r *DockerRuntime) StatsStream(ctx context.Context, id string) (io.ReadCloser, error) {
	resp, err := r.cli.ContainerStats(ctx, id, true)
	if err != nil {
		return nil, fmt.Errorf("container stats: %w", err)
	}
	return resp.Body, nil
}

// Wait blocks until the container exits and returns its exit code.
func (r *DockerRuntime) Wait(ctx context.Context, id string) (int, error) {
	statusCh, errCh := r.cli.ContainerWait(ctx, id, container.WaitConditionNotRunning)
	select {
	case <-ctx.Done():
		return -1, ctx.Err()
	case err := <-errCh:
		if err != nil {
			return -1, fmt.Errorf("container wait: %w", err)
		}
		return -1, nil
	case s := <-statusCh:
		if s.Error != nil {
			return int(s.StatusCode), fmt.Errorf("wait: %s", s.Error.Message)
		}
		return int(s.StatusCode), nil
	}
}

// ---- Container-path file ops (used by the panel's file manager) ----

// ExecCapture runs cmd inside the given running container and returns its
// captured stdout + stderr + exit code.
func (r *DockerRuntime) ExecCapture(ctx context.Context, id string, cmd []string) ([]byte, []byte, int, error) {
	resp, err := r.cli.ContainerExecCreate(ctx, id, container.ExecOptions{
		Cmd:          cmd,
		AttachStdout: true,
		AttachStderr: true,
	})
	if err != nil {
		return nil, nil, -1, fmt.Errorf("exec create: %w", err)
	}
	att, err := r.cli.ContainerExecAttach(ctx, resp.ID, container.ExecStartOptions{})
	if err != nil {
		return nil, nil, -1, fmt.Errorf("exec attach: %w", err)
	}
	defer att.Close()

	// StdCopy reads the hijacked exec stream, which is NOT bound by ctx — a
	// container caught mid-restart can leave this Read blocked forever, leaking
	// the underlying docker connection. Unbounded, that once cascaded into every
	// runtime op hanging (config file-reads 504'd; a log stream stopped
	// re-attaching). Run StdCopy off a goroutine and enforce ctx ourselves: on
	// timeout, Close() the attach to unblock the Read, then drain the goroutine
	// so the buffers are race-free to read.
	var stdout, stderr bytes.Buffer
	copyDone := make(chan error, 1)
	go func() { _, e := stdcopy.StdCopy(&stdout, &stderr, att.Reader); copyDone <- e }()
	select {
	case <-ctx.Done():
		att.Close() // unblock the goroutine's Read (defer Close is a safe no-op double-close)
		<-copyDone
		return stdout.Bytes(), stderr.Bytes(), -1, fmt.Errorf("exec copy: %w", ctx.Err())
	case err := <-copyDone:
		if err != nil && !errors.Is(err, io.EOF) {
			return stdout.Bytes(), stderr.Bytes(), -1, fmt.Errorf("exec copy: %w", err)
		}
	}
	insp, err := r.cli.ContainerExecInspect(ctx, resp.ID)
	if err != nil {
		return stdout.Bytes(), stderr.Bytes(), -1, fmt.Errorf("exec inspect: %w", err)
	}
	return stdout.Bytes(), stderr.Bytes(), insp.ExitCode, nil
}

// CopyFileFromContainer returns the raw bytes of the file at containerPath.
// Works on both running and stopped containers. Follows symlinks (Docker's
// archive API tars a symlink as a symlink without dereferencing — needed
// for cases like the 7DTD module's /game/serverconfig.xml -> /game-saves/…).
func (r *DockerRuntime) CopyFileFromContainer(ctx context.Context, id, containerPath string) ([]byte, error) {
	return r.copyFileFromContainerHop(ctx, id, containerPath, 0)
}

// maxCopyFileSymlinkHops bounds symlink resolution depth so a malicious or
// accidental symlink loop can't pin an agent goroutine. 8 is way more than
// any legitimate chain we've seen; bump it if a real use-case demands.
const maxCopyFileSymlinkHops = 8

func (r *DockerRuntime) copyFileFromContainerHop(ctx context.Context, id, containerPath string, depth int) ([]byte, error) {
	if depth >= maxCopyFileSymlinkHops {
		return nil, fmt.Errorf("symlink chain too deep (>%d hops) at %s", maxCopyFileSymlinkHops, containerPath)
	}
	rc, _, err := r.cli.CopyFromContainer(ctx, id, containerPath)
	if err != nil {
		return nil, fmt.Errorf("copy from container: %w", err)
	}
	defer rc.Close()
	tr := tar.NewReader(rc)
	for {
		hdr, err := tr.Next()
		if err == io.EOF {
			return nil, fmt.Errorf("file not found in archive")
		}
		if err != nil {
			return nil, fmt.Errorf("tar next: %w", err)
		}
		switch hdr.Typeflag {
		case tar.TypeReg, tar.TypeRegA:
			return io.ReadAll(tr)
		case tar.TypeDir:
			return nil, fmt.Errorf("path is a directory")
		case tar.TypeSymlink, tar.TypeLink:
			// Resolve the target. Docker hands back the symlink itself —
			// it never dereferences. Linkname can be absolute or relative;
			// relative is resolved against the link's containing dir, not
			// our cwd, mirroring how the kernel resolves it inside the FS.
			target := hdr.Linkname
			if !path.IsAbs(target) {
				target = path.Join(path.Dir(containerPath), target)
			}
			return r.copyFileFromContainerHop(ctx, id, target, depth+1)
		}
	}
}

// AttachStdio attaches to the container's multiplexed stdin/stdout/stderr
// stream and returns a single ReadWriteCloser. Writes go to stdin; reads
// are the raw multiplexed frame stream (our stdio RCON adapter drains and
// discards the reader — responses reach the panel via docker logs instead,
// so we don't need to demux here).
func (r *DockerRuntime) AttachStdio(ctx context.Context, id string) (io.ReadWriteCloser, error) {
	resp, err := r.cli.ContainerAttach(ctx, id, container.AttachOptions{
		Stream: true,
		Stdin:  true,
		Stdout: true,
		Stderr: true,
	})
	if err != nil {
		return nil, fmt.Errorf("container attach: %w", err)
	}
	return &hijackedStream{resp: resp}, nil
}

// hijackedStream adapts Docker's HijackedResponse (a bidirectional net.Conn
// plus a buffered reader) to io.ReadWriteCloser so the stdio adapter can
// treat it uniformly.
type hijackedStream struct {
	resp types.HijackedResponse
	mu   sync.Mutex
}

func (h *hijackedStream) Write(p []byte) (int, error) {
	return h.resp.Conn.Write(p)
}
func (h *hijackedStream) Read(p []byte) (int, error) {
	return h.resp.Reader.Read(p)
}
func (h *hijackedStream) Close() error {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.resp.Close()
	return nil
}

// CopyFileToContainer writes content to containerPath inside the container.
// Parent dir must exist (Docker's archive API doesn't create them).
func (r *DockerRuntime) CopyFileToContainer(ctx context.Context, id, containerPath string, content []byte) error {
	dir := path.Dir(containerPath)
	base := path.Base(containerPath)

	var buf bytes.Buffer
	tw := tar.NewWriter(&buf)
	if err := tw.WriteHeader(&tar.Header{
		Name: base,
		Mode: 0o644,
		Size: int64(len(content)),
	}); err != nil {
		return fmt.Errorf("tar header: %w", err)
	}
	if _, err := tw.Write(content); err != nil {
		return fmt.Errorf("tar body: %w", err)
	}
	if err := tw.Close(); err != nil {
		return fmt.Errorf("tar close: %w", err)
	}

	return r.cli.CopyToContainer(ctx, id, dir, &buf, container.CopyToContainerOptions{AllowOverwriteDirWithFile: false})
}

// CopyTarToContainer writes a raw tar stream onto dstDir. Caller is
// responsible for the tar formatting (one Header+body per entry, then
// Close()). Docker extracts in place. Used by the file manager's
// extract endpoint so we can ship a whole archive's contents in one
// API call instead of N per-file CopyFileToContainer round-trips.
func (r *DockerRuntime) CopyTarToContainer(ctx context.Context, id, dstDir string, tarStream io.Reader) error {
	return r.cli.CopyToContainer(ctx, id, dstDir, tarStream, container.CopyToContainerOptions{AllowOverwriteDirWithFile: true})
}

// CopyDirFromContainer returns a tar stream rooted at containerPath.
// Caller must Close() the reader. Used by the file manager's compress
// endpoint to walk a directory's contents without per-file API calls.
func (r *DockerRuntime) CopyDirFromContainer(ctx context.Context, id, containerPath string) (io.ReadCloser, error) {
	rc, _, err := r.cli.CopyFromContainer(ctx, id, containerPath)
	if err != nil {
		return nil, fmt.Errorf("copy from container: %w", err)
	}
	return rc, nil
}

// acquireImage ensures the requested image is present locally. Fast path:
// already on disk → return. Otherwise:
//
//   - `panel-*` images are NEVER published to a registry. We auto-build
//     from buildContext/Dockerfile when buildContext is set. This is the
//     "drop in a new agent and any module just works" behavior — first
//     instance create on a fresh agent triggers the local build of the
//     module's runtime image. Subsequent creates hit the inspect-success
//     fast path.
//
//   - Non-panel images (alpine, acekorneya/asa-linux-server, etc.) are
//     pulled from the registry as usual.
//
// If the image is `panel-*` and buildContext is empty (no Dockerfile in
// the module dir), we surface the same "build it on the agent host"
// message as before — the operator's module is broken, not the agent.
func (r *DockerRuntime) acquireImage(ctx context.Context, ref, buildContext string, sink func(string)) error {
	emit := func(line string) {
		if sink != nil {
			sink(line)
		}
	}
	if _, err := r.cli.ImageInspect(ctx, ref); err == nil {
		return nil
	} else if !client.IsErrNotFound(err) {
		return fmt.Errorf("image inspect: %w", err)
	}

	isPanel := strings.HasPrefix(ref, "panel-") && !strings.Contains(ref, "/")
	if isPanel {
		if buildContext == "" {
			return fmt.Errorf("module image %q not built on this agent and no build context provided", ref)
		}
		dockerfile := filepath.Join(buildContext, "Dockerfile")
		if _, err := os.Stat(dockerfile); err != nil {
			return fmt.Errorf("module image %q missing on this agent and no Dockerfile at %s", ref, dockerfile)
		}
		emit(fmt.Sprintf("[image] %s missing on this agent — building from %s/Dockerfile (first-run; takes 30s–3min)", ref, buildContext))
		return r.buildLocalImage(ctx, ref, buildContext, emit)
	}

	// Public image — pull from registry.
	emit(fmt.Sprintf("[image] pulling %s from registry…", ref))
	rc, err := r.cli.ImagePull(ctx, ref, image.PullOptions{})
	if err != nil {
		return fmt.Errorf("image pull: %w", err)
	}
	defer rc.Close()
	streamDockerJSON(rc, emit)
	emit(fmt.Sprintf("[image] %s ready", ref))
	return nil
}

// buildLocalImage runs `docker build -t <ref> <ctxDir>` via the Docker
// daemon's API. Build output is parsed (NDJSON) and forwarded to `emit`
// so the controller's Console tab gets live progress while the operator
// waits for first-time builds.
func (r *DockerRuntime) buildLocalImage(ctx context.Context, ref, ctxDir string, emit func(string)) error {
	tarball, err := archive.TarWithOptions(ctxDir, &archive.TarOptions{})
	if err != nil {
		return fmt.Errorf("tar build context: %w", err)
	}
	defer tarball.Close()
	resp, err := r.cli.ImageBuild(ctx, tarball, build.ImageBuildOptions{
		Tags:        []string{ref},
		Dockerfile:  "Dockerfile",
		Remove:      true,
		ForceRemove: true,
		PullParent:  false,
	})
	if err != nil {
		return fmt.Errorf("image build start: %w", err)
	}
	defer resp.Body.Close()
	if buildErr := streamDockerJSON(resp.Body, emit); buildErr != nil {
		return fmt.Errorf("image build: %w", buildErr)
	}
	if _, err := r.cli.ImageInspect(ctx, ref); err != nil {
		return fmt.Errorf("image build %q produced no tag: %w", ref, err)
	}
	emit(fmt.Sprintf("[image] %s built and tagged", ref))
	return nil
}

// streamDockerJSON parses Docker's streaming JSON output (one object per
// line, fields like {"stream":"...","status":"...","error":"...","errorDetail":{...}})
// and forwards meaningful lines to emit. Returns a non-nil error iff the
// daemon reported an error in the stream.
func streamDockerJSON(rc io.Reader, emit func(string)) error {
	dec := json.NewDecoder(rc)
	for {
		var msg struct {
			Stream      string          `json:"stream"`
			Status      string          `json:"status"`
			Progress    string          `json:"progress"`
			ID          string          `json:"id"`
			Error       string          `json:"error"`
			ErrorDetail json.RawMessage `json:"errorDetail"`
		}
		if err := dec.Decode(&msg); err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}
			// Non-JSON tail / partial trailer — stop parsing but don't
			// fail the build; the post-build ImageInspect will catch
			// any tag-missing case.
			return nil
		}
		if msg.Error != "" {
			return errors.New(msg.Error)
		}
		// Build → "stream" lines (Step 1/12 : FROM debian:slim, etc).
		if s := strings.TrimRight(msg.Stream, "\r\n"); s != "" {
			emit(s)
		}
		// Pull → "status" lines, sometimes paired with progress for
		// each layer. Skip the high-frequency "Downloading" updates
		// (would spam the console at 10–30 Hz) — keep "Pulling /
		// Pulled" markers and final messages.
		if msg.Status != "" {
			st := msg.Status
			if strings.HasPrefix(st, "Downloading") || strings.HasPrefix(st, "Extracting") || strings.HasPrefix(st, "Verifying") {
				continue
			}
			if msg.ID != "" {
				emit(fmt.Sprintf("%s: %s", msg.ID, st))
			} else {
				emit(st)
			}
		}
	}
}

// buildPortBindings translates PortSpec to Docker's nat.PortSet / PortMap.
func buildPortBindings(ports []PortSpec) (nat.PortSet, nat.PortMap, error) {
	exposed := nat.PortSet{}
	bindings := nat.PortMap{}
	for _, p := range ports {
		proto := strings.ToLower(p.Proto)
		if proto != "tcp" && proto != "udp" {
			return nil, nil, fmt.Errorf("port %d/%s: proto must be tcp or udp", p.ContainerPort, p.Proto)
		}
		natPort, err := nat.NewPort(proto, strconv.Itoa(int(p.ContainerPort)))
		if err != nil {
			return nil, nil, fmt.Errorf("invalid port %d/%s: %w", p.ContainerPort, proto, err)
		}
		exposed[natPort] = struct{}{}
		hostPort := p.HostPort
		if hostPort == 0 {
			hostPort = p.ContainerPort
		}
		bindings[natPort] = []nat.PortBinding{{
			HostIP:   p.HostIP,
			HostPort: strconv.Itoa(int(hostPort)),
		}}
	}
	return exposed, bindings, nil
}

func pidsPtr(v int64) *int64 {
	if v <= 0 {
		return nil
	}
	return &v
}

// splitDockerTimestamp pulls docker's RFC3339Nano timestamp prefix off a
// log line and returns the remainder + parsed time. Docker emits each
// line as "<rfc3339nano-z> <line>". A space immediately follows the Z;
// anything else (including missing/garbled prefix) falls back to
// time.Now() so we never drop the line.
func splitDockerTimestamp(s string) (string, time.Time) {
	// Minimum valid prefix: "2026-01-01T00:00:00Z " — 21 chars. With nano
	// precision it's longer; the space after Z is what we look for.
	sp := strings.IndexByte(s, ' ')
	if sp < 20 || sp > 35 {
		return s, time.Now()
	}
	t, err := time.Parse(time.RFC3339Nano, s[:sp])
	if err != nil {
		return s, time.Now()
	}
	return s[sp+1:], t
}

// scanLogLine is a bufio.SplitFunc that treats \n, \r, and \r\n as line
// terminators. The stdlib default (bufio.ScanLines) only splits on \n and
// will buffer everything between \r-overwrites until it sees a real newline
// — that's how SteamCMD's "Update state downloading progress: X%" bars
// used to sit silent for minutes before dumping in one chunk.
func scanLogLine(data []byte, atEOF bool) (advance int, token []byte, err error) {
	if atEOF && len(data) == 0 {
		return 0, nil, nil
	}
	if i := bytes.IndexAny(data, "\r\n"); i >= 0 {
		// Swallow a trailing \n after \r so "\r\n" yields one line not two.
		advance = i + 1
		if data[i] == '\r' && i+1 < len(data) && data[i+1] == '\n' {
			advance++
		}
		return advance, bytes.TrimRight(data[:i], "\r\n"), nil
	}
	if atEOF {
		return len(data), data, nil
	}
	return 0, nil, nil
}

// ansiRE matches ANSI CSI sequences (colour, cursor movement, erase-line, …).
// vinanrra's LGSM wrapper for 7DTD emits these liberally; they're noise in a
// browser console and garble line-collapsing logic on the UI side.
var ansiRE = regexp.MustCompile(`\x1b\[[0-9;?]*[ -/]*[@-~]`)

func stripANSI(s string) string {
	if !strings.ContainsRune(s, 0x1b) {
		return s
	}
	return ansiRE.ReplaceAllString(s, "")
}