package runtime

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/container"
	dockerimage "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/client"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/stdcopy"
	"github.com/tidbcloud/AgentTeam/internal/automation/dsl"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
)

const (
	defaultDockerNetworkMode = "none"
	defaultDockerCPUs        = 1
	defaultDockerMemoryBytes = int64(1024 * 1024 * 1024)
	defaultDockerPidsLimit   = int64(256)
	dockerWorkspaceDir       = "/workspace"
	dockerHealthTimeout      = 5 * time.Second
	dockerCleanupTimeout     = 10 * time.Second
)

// ErrDockerUnavailable marks startup failures caused by an unreachable Docker
// daemon. Configuration errors remain ordinary errors so callers can fail fast.
var ErrDockerUnavailable = errors.New("Docker is unavailable")

// DockerExecutorConfig configures the Docker-backed automation job executor.
type DockerExecutorConfig struct {
	Images      map[string]string
	NetworkMode string
	CPUs        float64
	MemoryBytes int64
	PidsLimit   int64
	User        string
	Mounts      []DockerBindMount
}

// DockerBindMount is an operator-configured bind mount exposed to all Docker jobs.
type DockerBindMount struct {
	Source   string
	Target   string
	ReadOnly bool
}

// DockerExecutor executes server-managed automation jobs through Docker Engine.
type DockerExecutor struct {
	client dockerClient
	cfg    DockerExecutorConfig
	labels []string
}

type dockerClient interface {
	Ping(ctx context.Context) (types.Ping, error)
	ImageInspectWithRaw(ctx context.Context, imageID string) (dockerimage.InspectResponse, []byte, error)
	ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, containerName string) (container.CreateResponse, error)
	ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error
	ContainerExecCreate(ctx context.Context, containerID string, options container.ExecOptions) (container.ExecCreateResponse, error)
	ContainerExecAttach(ctx context.Context, execID string, config container.ExecAttachOptions) (types.HijackedResponse, error)
	ContainerExecInspect(ctx context.Context, execID string) (container.ExecInspect, error)
	ContainerKill(ctx context.Context, containerID string, signal string) error
	ContainerRemove(ctx context.Context, containerID string, options container.RemoveOptions) error
	Close() error
}

type dockerAPIClient struct {
	*client.Client
}

func (c dockerAPIClient) ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, containerName string) (container.CreateResponse, error) {
	return c.Client.ContainerCreate(ctx, config, hostConfig, networkingConfig, nil, containerName)
}

// NewDockerExecutor constructs a Docker executor and verifies Docker
// connectivity before the server-managed runner is advertised.
func NewDockerExecutor(cfg DockerExecutorConfig) (*DockerExecutor, error) {
	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
	if err != nil {
		return nil, fmt.Errorf("create Docker client: %w", err)
	}
	executor, err := newDockerExecutorWithClient(dockerAPIClient{Client: cli}, cfg)
	if err != nil {
		_ = cli.Close()
		return nil, err
	}
	return executor, nil
}

func newDockerExecutorWithClient(cli dockerClient, cfg DockerExecutorConfig) (*DockerExecutor, error) {
	if cli == nil {
		return nil, fmt.Errorf("Docker client is nil")
	}
	normalized, labels, err := normalizeDockerExecutorConfig(cfg)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), dockerHealthTimeout)
	defer cancel()
	if _, err := cli.Ping(ctx); err != nil {
		return nil, fmt.Errorf("%w: connect to Docker: %w", ErrDockerUnavailable, dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	if err := verifyDockerImagesReady(ctx, cli, normalized.Images, labels); err != nil {
		return nil, err
	}
	return &DockerExecutor{client: cli, cfg: normalized, labels: labels}, nil
}

// Labels returns the runner labels advertised by this Docker executor.
func (e *DockerExecutor) Labels() []string {
	if e == nil {
		return nil
	}
	labels := make([]string, len(e.labels))
	copy(labels, e.labels)
	return labels
}

// Close releases the Docker client resources owned by the executor.
func (e *DockerExecutor) Close() error {
	if e == nil || e.client == nil {
		return nil
	}
	return e.client.Close()
}

// ExecuteJob runs a validated shell-only automation job in one Docker container.
func (e *DockerExecutor) ExecuteJob(ctx context.Context, req ExecuteJobRequest) (ExecuteJobResult, error) {
	var result ExecuteJobResult
	if e == nil || e.client == nil {
		return result, fmt.Errorf("Docker executor is nil")
	}
	image, _, err := e.resolveImage(req.Spec.RunsOn)
	if err != nil {
		return result, err
	}
	jobCtx := ctx
	containerID, err := e.createJobContainer(jobCtx, req, image)
	if err != nil {
		return result, err
	}
	killWatch := e.killContainerWhenDone(jobCtx, containerID)
	finish := func(primary error) (ExecuteJobResult, error) {
		cleanupErr := markExecutorCleanupError(e.cleanupContainer(jobCtx, containerID, killWatch.stop()))
		if primary != nil || cleanupErr != nil {
			return result, errors.Join(primary, cleanupErr)
		}
		return result, nil
	}
	if err := e.client.ContainerStart(jobCtx, containerID, container.StartOptions{}); err != nil {
		return finish(fmt.Errorf("start automation Docker container: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err)))
	}
	result.Steps = make([]JobStepResult, 0, len(req.Spec.Steps))
	for idx, step := range req.Spec.Steps {
		started := executeJobNow(req)
		stepResult := JobStepResult{
			ID:        strings.TrimSpace(step.ID),
			Name:      strings.TrimSpace(step.Name),
			Index:     idx,
			StartedAt: started.Format(time.RFC3339Nano),
		}
		exitCode, err := e.executeJobStep(jobCtx, containerID, req, step, idx)
		stepResult.ExitCode = exitCode
		stepResult.CompletedAt = executeJobNow(req).Format(time.RFC3339Nano)
		result.Steps = append(result.Steps, stepResult)
		if err != nil {
			return finish(err)
		}
	}
	return finish(nil)
}

func (e *DockerExecutor) executeJobStep(ctx context.Context, containerID string, req ExecuteJobRequest, step dsl.StepSpec, stepIndex int) (int, error) {
	if strings.TrimSpace(step.Uses) != "" {
		if req.UsesStepHandler == nil {
			return -1, fmt.Errorf("automation step uses is not supported by the configured executor")
		}
		return req.UsesStepHandler(ctx, UsesStepRequest{Step: step, Index: stepIndex})
	}
	return e.executeStep(ctx, containerID, req, stepIndex, step.Run)
}

func normalizeDockerExecutorConfig(cfg DockerExecutorConfig) (DockerExecutorConfig, []string, error) {
	images, labels, err := normalizeDockerImageMap(cfg.Images)
	if err != nil {
		return DockerExecutorConfig{}, nil, err
	}
	cfg.Images = images
	cfg.NetworkMode = strings.TrimSpace(cfg.NetworkMode)
	if cfg.NetworkMode == "" {
		cfg.NetworkMode = defaultDockerNetworkMode
	}
	if strings.HasPrefix(strings.ToLower(cfg.NetworkMode), "container:") {
		return DockerExecutorConfig{}, nil, fmt.Errorf("Docker automation network mode must not join another container")
	}
	if cfg.CPUs <= 0 {
		cfg.CPUs = defaultDockerCPUs
	}
	if cfg.MemoryBytes <= 0 {
		cfg.MemoryBytes = defaultDockerMemoryBytes
	}
	if cfg.PidsLimit <= 0 {
		cfg.PidsLimit = defaultDockerPidsLimit
	}
	cfg.User = strings.TrimSpace(cfg.User)
	if cfg.User == "" {
		cfg.User = defaultDockerUser()
	}
	cfg.Mounts = normalizeDockerBindMounts(cfg.Mounts)
	return cfg, labels, nil
}

func normalizeDockerBindMounts(mounts []DockerBindMount) []DockerBindMount {
	out := make([]DockerBindMount, 0, len(mounts))
	for _, item := range mounts {
		source := strings.TrimSpace(item.Source)
		target := strings.TrimSpace(item.Target)
		if source == "" || target == "" {
			continue
		}
		out = append(out, DockerBindMount{Source: source, Target: target, ReadOnly: item.ReadOnly})
	}
	return out
}

func defaultDockerUser() string {
	return fmt.Sprintf("%d:%d", os.Geteuid(), os.Getegid())
}

func normalizeDockerImageMap(raw map[string]string) (map[string]string, []string, error) {
	if len(raw) == 0 {
		return nil, nil, fmt.Errorf("MESHX_AUTOMATION_DOCKER_IMAGES must map at least one runner label to a Docker image")
	}
	images := make(map[string]string, len(raw))
	labels := make([]string, 0, len(raw))
	for label, image := range raw {
		normalizedLabels := NormalizeRunnerLabels([]string{label})
		if len(normalizedLabels) != 1 {
			return nil, nil, fmt.Errorf("Docker automation image label is required")
		}
		key := strings.ToLower(normalizedLabels[0])
		image = strings.TrimSpace(image)
		if image == "" {
			return nil, nil, fmt.Errorf("Docker automation image for label %q is required", normalizedLabels[0])
		}
		if _, ok := images[key]; ok {
			return nil, nil, fmt.Errorf("Docker automation image label %q is duplicated", normalizedLabels[0])
		}
		images[key] = image
		labels = append(labels, key)
	}
	sort.Strings(labels)
	return images, labels, nil
}

func verifyDockerImagesReady(ctx context.Context, cli dockerClient, images map[string]string, labels []string) error {
	for _, label := range labels {
		image := strings.TrimSpace(images[label])
		if _, _, err := cli.ImageInspectWithRaw(ctx, image); err != nil {
			return fmt.Errorf("inspect Docker image for automation label %q (%s): %w", label, image, dockerImageAvailabilityError(err))
		}
	}
	return nil
}

func (e *DockerExecutor) resolveImage(runsOn any) (string, string, error) {
	group, labels := runnerTargetFromRunsOn(runsOn)
	if strings.TrimSpace(group) != "" {
		return "", "", fmt.Errorf("Docker automation jobs must not specify a runner group")
	}
	if len(labels) != 1 {
		return "", "", fmt.Errorf("Docker automation jobs must specify exactly one runs-on label")
	}
	label := strings.ToLower(strings.TrimSpace(labels[0]))
	image := e.cfg.Images[label]
	if image == "" {
		return "", "", fmt.Errorf("Docker automation image is not configured for runs-on label %q", labels[0])
	}
	return image, label, nil
}

func (e *DockerExecutor) createJobContainer(ctx context.Context, req ExecuteJobRequest, image string) (string, error) {
	pidsLimit := e.cfg.PidsLimit
	networkMode := container.NetworkMode(e.cfg.NetworkMode)
	disabledNetwork := strings.EqualFold(e.cfg.NetworkMode, defaultDockerNetworkMode)
	created, err := e.client.ContainerCreate(ctx, &container.Config{
		Image:           image,
		User:            e.cfg.User,
		WorkingDir:      dockerWorkspaceDir,
		Env:             append([]string(nil), req.Env...),
		Cmd:             []string{"/bin/sh", "-c", "while :; do sleep 3600; done"},
		NetworkDisabled: disabledNetwork,
		Labels: map[string]string{
			"meshx.automation.runner_id":      strings.TrimSpace(req.RunnerID),
			"meshx.automation.run_id":         req.Job.RunID,
			"meshx.automation.job_run_id":     req.Job.JobRunID,
			"meshx.automation.queue_job_id":   fmt.Sprintf("%d", req.Job.QueueJobID),
			"meshx.automation.workspace_id":   req.Job.WorkspaceID,
			"meshx.automation.executor":       "docker",
			"meshx.automation.cleanup_policy": "force-remove",
		},
	}, &container.HostConfig{
		NetworkMode:     networkMode,
		Privileged:      false,
		CapDrop:         []string{"ALL"},
		SecurityOpt:     []string{"no-new-privileges"},
		AutoRemove:      false,
		PublishAllPorts: false,
		Resources: container.Resources{
			NanoCPUs:  int64(e.cfg.CPUs * 1_000_000_000),
			Memory:    e.cfg.MemoryBytes,
			PidsLimit: &pidsLimit,
		},
		Mounts: e.jobMounts(req.WorkDir),
	}, &network.NetworkingConfig{}, "")
	if err != nil {
		return "", fmt.Errorf("create automation Docker container: %w", dockerImageAvailabilityError(err))
	}
	if strings.TrimSpace(created.ID) == "" {
		return "", fmt.Errorf("create automation Docker container: empty container id")
	}
	return created.ID, nil
}

func (e *DockerExecutor) jobMounts(workDir string) []mount.Mount {
	mounts := []mount.Mount{{
		Type:   mount.TypeBind,
		Source: workDir,
		Target: dockerWorkspaceDir,
		BindOptions: &mount.BindOptions{
			Propagation: mount.PropagationRPrivate,
		},
	}}
	for _, item := range e.cfg.Mounts {
		mounts = append(mounts, mount.Mount{
			Type:     mount.TypeBind,
			Source:   item.Source,
			Target:   item.Target,
			ReadOnly: item.ReadOnly,
			BindOptions: &mount.BindOptions{
				Propagation: mount.PropagationRPrivate,
			},
		})
	}
	return mounts
}

func (e *DockerExecutor) executeStep(ctx context.Context, containerID string, req ExecuteJobRequest, stepIndex int, script string) (int, error) {
	created, err := e.client.ContainerExecCreate(ctx, containerID, container.ExecOptions{
		User:         e.cfg.User,
		AttachStdout: true,
		AttachStderr: true,
		WorkingDir:   dockerWorkspaceDir,
		Env:          executeJobStepEnv(req, stepIndex),
		Cmd:          []string{"/bin/sh", "-e", "-c", script},
	})
	if err != nil {
		return -1, fmt.Errorf("create automation Docker exec: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	if strings.TrimSpace(created.ID) == "" {
		return -1, fmt.Errorf("create automation Docker exec: empty exec id")
	}
	attached, err := e.client.ContainerExecAttach(ctx, created.ID, container.ExecAttachOptions{})
	if err != nil {
		return -1, fmt.Errorf("attach automation Docker exec: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	if copyErr := copyDockerExecOutput(ctx, req.Stdout, req.Stderr, attached); copyErr != nil {
		return -1, fmt.Errorf("write automation job log: %w", copyErr)
	}
	if err := ctx.Err(); err != nil {
		return -1, err
	}
	inspected, err := e.client.ContainerExecInspect(ctx, created.ID)
	if err != nil {
		return -1, fmt.Errorf("inspect automation Docker exec: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	if inspected.Running {
		return -1, fmt.Errorf("automation Docker exec is still running after log stream closed")
	}
	if inspected.ExitCode != 0 {
		return inspected.ExitCode, fmt.Errorf("automation shell exited with code %d", inspected.ExitCode)
	}
	return 0, nil
}

func copyDockerExecOutput(ctx context.Context, stdout io.Writer, stderr io.Writer, attached types.HijackedResponse) error {
	if stdout == nil {
		stdout = io.Discard
	}
	if stderr == nil {
		stderr = io.Discard
	}
	stdoutWriter := &dockerExecOutputWriter{writer: stdout}
	stderrWriter := &dockerExecOutputWriter{writer: stderr}
	done := make(chan struct{})
	var closeOnce sync.Once
	closeAttached := func() {
		closeOnce.Do(func() { attached.Close() })
	}
	if ctx != nil {
		logging.Go(nil, func() {
			select {
			case <-ctx.Done():
				closeAttached()
			case <-done:
			}
		})
	}
	_, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, attached.Reader)
	close(done)
	closeAttached()
	if writerErr := errors.Join(stdoutWriter.err, stderrWriter.err); writerErr != nil {
		return writerErr
	}
	if ctx != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return ctxErr
		}
	}
	return err
}

type dockerExecOutputWriter struct {
	writer io.Writer
	err    error
}

func (w *dockerExecOutputWriter) Write(payload []byte) (int, error) {
	n, err := w.writer.Write(payload)
	if err != nil && w.err == nil {
		w.err = err
	}
	return n, err
}

type dockerKillWatch struct {
	done     chan struct{}
	result   chan dockerKillResult
	stopOnce sync.Once
}

type dockerKillResult struct {
	killed bool
	err    error
}

func (w *dockerKillWatch) stop() dockerKillResult {
	if w == nil {
		return dockerKillResult{}
	}
	w.stopOnce.Do(func() { close(w.done) })
	return <-w.result
}

func (e *DockerExecutor) killContainerWhenDone(ctx context.Context, containerID string) *dockerKillWatch {
	watch := &dockerKillWatch{
		done:   make(chan struct{}),
		result: make(chan dockerKillResult, 1),
	}
	logging.Go(nil, func() {
		select {
		case <-watch.done:
			watch.result <- dockerKillResult{}
		case <-ctx.Done():
			watch.result <- dockerKillResult{killed: true, err: e.killContainer(containerID)}
		}
	})
	return watch
}

func (e *DockerExecutor) cleanupContainer(ctx context.Context, containerID string, killResult dockerKillResult) error {
	var cleanupErr error
	if killResult.err != nil {
		cleanupErr = errors.Join(cleanupErr, killResult.err)
	}
	if ctx.Err() != nil && !killResult.killed {
		cleanupErr = errors.Join(cleanupErr, e.killContainer(containerID))
	}
	return errors.Join(cleanupErr, e.removeContainer(containerID))
}

func (e *DockerExecutor) killContainer(containerID string) error {
	ctx, cancel := context.WithTimeout(context.Background(), dockerCleanupTimeout)
	defer cancel()
	if err := e.client.ContainerKill(ctx, containerID, "KILL"); err != nil {
		return fmt.Errorf("kill automation Docker container: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	return nil
}

func (e *DockerExecutor) removeContainer(containerID string) error {
	ctx, cancel := context.WithTimeout(context.Background(), dockerCleanupTimeout)
	defer cancel()
	if err := e.client.ContainerRemove(ctx, containerID, container.RemoveOptions{RemoveVolumes: true, Force: true}); err != nil {
		return fmt.Errorf("remove automation Docker container: %w", dockerExecutorError(ExecutorFailureDockerUnavailable, err))
	}
	return nil
}

func dockerImageAvailabilityError(err error) error {
	kind := ExecutorFailureDockerUnavailable
	if errdefs.IsNotFound(err) || strings.Contains(strings.ToLower(err.Error()), "no such image") {
		kind = ExecutorFailureImageMissing
	}
	return dockerExecutorError(kind, err)
}

func dockerExecutorError(kind ExecutorFailureKind, err error) error {
	if err == nil {
		return nil
	}
	return ExecutorError{Kind: kind, Err: err}
}

func executeJobNow(req ExecuteJobRequest) time.Time {
	if req.Now != nil {
		return req.Now().UTC()
	}
	return time.Now().UTC()
}

func executeJobStepEnv(req ExecuteJobRequest, stepIndex int) []string {
	if stepIndex >= 0 && stepIndex < len(req.StepEnv) {
		return append([]string(nil), req.StepEnv[stepIndex]...)
	}
	return append([]string(nil), req.Env...)
}