package k3d

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

	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/cluster"
	"github.com/tidbcloud/controlplane9/internal/config"
	cp9errors "github.com/tidbcloud/controlplane9/internal/errors"
	"github.com/tidbcloud/controlplane9/internal/host"
	"github.com/tidbcloud/controlplane9/internal/stub"
	"github.com/tidbcloud/controlplane9/internal/verbose"
)

const (
	localImportVerifyTimeout    = 10 * time.Second
	localImportVerifyRetryDelay = 1 * time.Second
	localImportVerifyAttempts   = 3
	legacyPauseMarkerFileName   = "paused"
)

// Options carries local-k3d provisioning knobs from config.
type Options struct {
	Agents     int
	Memory     string
	K3dVersion string
	K3sImage   string
}

// Adapter manages the explicit local-k3d host implementation.
type Adapter struct {
	client    cluster.K3dClient
	available func() bool
	ensure    func(ctx context.Context, version string) error
	verify    func(ctx context.Context, clusterName, image string) error
	attempts  int
	delay     time.Duration
	timeout   time.Duration
}

// NewAdapter returns a local-k3d host adapter backed by the real k3d CLI.
func NewAdapter() *Adapter {
	return &Adapter{
		client:    cluster.NewExecK3dClient(),
		available: cluster.K3dAvailable,
		ensure:    cluster.EnsureK3d,
		verify:    verifyImportedImageInCluster,
	}
}

// NewAdapterWithClient returns a local-k3d host adapter backed by a test client.
func NewAdapterWithClient(client cluster.K3dClient) *Adapter {
	return &Adapter{
		client:    client,
		available: func() bool { return true },
		ensure:    func(_ context.Context, _ string) error { return nil },
	}
}

// SelectedHost resolves the local-k3d selected host contract for a session.
func SelectedHost(cfg *config.Config, local *config.LocalConfig) *host.Session {
	return host.Resolve(cfg, local)
}

// CheckConflict verifies that no orphan local-k3d cluster exists for a fresh session.
func (a *Adapter) CheckConflict(ctx context.Context, selected *host.Session) error {
	if !a.available() {
		return nil
	}
	exists, err := a.client.ClusterExists(ctx, selected.ClusterName)
	if err != nil {
		return nil
	}
	if exists {
		return cp9errors.Newf(cp9errors.CategoryResource,
			"k3d cluster %q already exists — another cp9 environment may be using this session name; "+
				"choose a different name in cp9.local.yaml or run `cp9 destroy` first", selected.ClusterName)
	}
	return nil
}

// Destroy deletes the local-k3d cluster when it still exists.
func (a *Adapter) Destroy(ctx context.Context, selected *host.Session) error {
	exists, err := a.client.ClusterExists(ctx, selected.ClusterName)
	if err != nil {
		return fmt.Errorf("check cluster: %w", err)
	}
	if !exists {
		return nil
	}
	if err := a.client.DeleteCluster(ctx, selected.ClusterName); err != nil {
		return fmt.Errorf("delete local-k3d cluster: %w", err)
	}
	return nil
}

// ImportImage imports a built image into the selected local-k3d cluster.
func (a *Adapter) ImportImage(ctx context.Context, selected *host.Session, image string) error {
	verify := a.verify
	if verify == nil {
		return a.client.ImportImage(ctx, selected.ClusterName, image)
	}
	attempts := a.attempts
	if attempts <= 0 {
		attempts = localImportVerifyAttempts
	}

	var lastErr error
	for attempt := 1; attempt <= attempts; attempt++ {
		if err := a.client.ImportImage(ctx, selected.ClusterName, image); err != nil {
			return err
		}
		err := waitForImportedImage(ctx, selected.ClusterName, image, verify, a.timeout)
		if err == nil {
			return nil
		}
		lastErr = err
		if attempt == attempts {
			break
		}
		if err := sleepUntilNextImportAttempt(ctx, a.delay); err != nil {
			return err
		}
	}

	return fmt.Errorf("verify imported image %s in local-k3d cluster %s: %w", image, selected.ClusterName, lastErr)
}

func waitForImportedImage(
	ctx context.Context,
	clusterName, image string,
	verify func(context.Context, string, string) error,
	timeout time.Duration,
) error {
	if timeout <= 0 {
		timeout = localImportVerifyTimeout
	}
	deadline := time.Now().Add(timeout)
	var lastErr error
	for {
		err := verify(ctx, clusterName, image)
		if err == nil {
			return nil
		}
		lastErr = err
		if !time.Now().Before(deadline) {
			return lastErr
		}
		timer := time.NewTimer(500 * time.Millisecond)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func sleepUntilNextImportAttempt(ctx context.Context, delay time.Duration) error {
	if delay <= 0 {
		delay = localImportVerifyRetryDelay
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

func verifyImportedImageInCluster(ctx context.Context, clusterName, image string) error {
	nodes, err := listK3dNodeContainers(ctx, clusterName)
	if err != nil {
		return err
	}

	var missing []string
	var diagnostics []string
	for _, node := range nodes {
		images, err := listNodeImages(ctx, node)
		if err != nil {
			return err
		}
		if slices.Contains(images, image) {
			continue
		}
		missing = append(missing, node)
		diagnostics = append(diagnostics, fmt.Sprintf("%s images=%s", node, summarizeNodeImages(images)))
	}
	if len(missing) == 0 {
		return nil
	}
	return fmt.Errorf("missing from nodes %v after import; %s", missing, strings.Join(diagnostics, "; "))
}

func listK3dNodeContainers(ctx context.Context, clusterName string) ([]string, error) {
	cmd := exec.CommandContext(ctx, "docker", "ps", "--format", "{{.Names}}")
	output, err := verbose.RunCommand(cmd)
	if err != nil {
		return nil, fmt.Errorf("docker ps for k3d nodes: %w\n%s", err, string(output))
	}

	prefix := "k3d-" + clusterName + "-"
	var nodes []string
	for line := range strings.SplitSeq(strings.TrimSpace(string(output)), "\n") {
		name := strings.TrimSpace(line)
		if name == "" {
			continue
		}
		if !strings.HasPrefix(name, prefix) {
			continue
		}
		if strings.Contains(name, "-server-") || strings.Contains(name, "-agent-") {
			nodes = append(nodes, name)
		}
	}
	if len(nodes) == 0 {
		return nil, fmt.Errorf("no k3d node containers found for cluster %s", clusterName)
	}
	slices.Sort(nodes)
	return nodes, nil
}

func listNodeImages(ctx context.Context, node string) ([]string, error) {
	cmd := exec.CommandContext(
		ctx,
		"docker", "exec", node,
		"sh", "-lc",
		"k3s ctr -n k8s.io images list -q 2>/dev/null || ctr -n k8s.io images list -q",
	)
	output, err := verbose.RunCommand(cmd)
	if err != nil {
		return nil, fmt.Errorf("list node images on %s: %w\n%s", node, err, string(output))
	}
	var images []string
	for line := range strings.SplitSeq(strings.TrimSpace(string(output)), "\n") {
		image := strings.TrimSpace(line)
		if image == "" {
			continue
		}
		images = append(images, image)
	}
	slices.Sort(images)
	return images, nil
}

func summarizeNodeImages(images []string) string {
	if len(images) == 0 {
		return "<none>"
	}
	var local []string
	for _, image := range images {
		if strings.HasPrefix(image, "cp9.local/") {
			local = append(local, image)
		}
	}
	if len(local) > 0 {
		if len(local) > 8 {
			local = append(local[:8], fmt.Sprintf("... (%d total cp9.local images)", len(local)))
		}
		return strings.Join(local, ", ")
	}
	if len(images) > 8 {
		images = append(images[:8], fmt.Sprintf("... (%d total images)", len(images)))
	}
	return strings.Join(images, ", ")
}

// ClusterStep provisions the selected local-k3d host for cluster_create.
type ClusterStep struct {
	Selected *host.Session
	Options  Options
	Adapter  *Adapter
}

func (s *ClusterStep) Name() cluster.StepName { return cluster.StepClusterCreate }

func (s *ClusterStep) Run(ctx context.Context) error {
	if err := s.Adapter.ensure(ctx, s.Options.K3dVersion); err != nil {
		return fmt.Errorf("ensure k3d: %w", err)
	}
	volumeSpec, err := recordingStateVolume(s.Selected)
	if err != nil {
		return fmt.Errorf("prepare recording mount: %w", err)
	}
	volumes := []string{volumeSpec}
	ports, err := dedicatedAPIPortMappings(s.Selected)
	if err != nil {
		return err
	}

	healthy, err := s.Adapter.client.ClusterHealthy(ctx, s.Selected.ClusterName)
	if err != nil {
		return fmt.Errorf("check cluster health: %w", err)
	}
	if !healthy {
		exists, err := s.Adapter.client.ClusterExists(ctx, s.Selected.ClusterName)
		if err != nil {
			return fmt.Errorf("check cluster: %w", err)
		}
		if err := clearLegacyPauseMarker(s.Selected.StateDir); err != nil {
			return err
		}
		if exists {
			if err := s.Adapter.client.DeleteCluster(ctx, s.Selected.ClusterName); err != nil {
				return fmt.Errorf("delete unhealthy cluster: %w", err)
			}
		}
		if s.Options.K3sImage != "" {
			if err := s.Adapter.client.CreateClusterWithImage(ctx,
				s.Selected.ClusterName, s.Options.Agents, s.Options.Memory, s.Options.K3sImage, volumes, ports); err != nil {
				return err
			}
		} else {
			if err := s.Adapter.client.CreateCluster(ctx,
				s.Selected.ClusterName, s.Options.Agents, s.Options.Memory, volumes, ports); err != nil {
				return err
			}
		}
	}

	if err := s.Adapter.client.ExportKubeconfig(ctx, s.Selected.ClusterName, s.Selected.KubeconfigPath); err != nil {
		return fmt.Errorf("export kubeconfig: %w", err)
	}
	if err := rewriteLocalKubeconfigEndpoint(s.Selected.KubeconfigPath); err != nil {
		return fmt.Errorf("normalize local kubeconfig endpoint: %w", err)
	}
	if err := host.Save(s.Selected.StateDir, s.Selected); err != nil {
		return fmt.Errorf("persist selected host: %w", err)
	}
	return nil
}

func dedicatedAPIPortMappings(selected *host.Session) ([]string, error) {
	if selected == nil || selected.DedicatedAPI == nil {
		return nil, nil
	}
	if selected.DedicatedAPI.HostPort < 1 || selected.DedicatedAPI.HostPort > 65535 {
		return nil, fmt.Errorf("prepare Dedicated API port mapping: invalid host port %d", selected.DedicatedAPI.HostPort)
	}
	if selected.DedicatedAPI.NodePort < 30000 || selected.DedicatedAPI.NodePort > 32767 {
		return nil, fmt.Errorf("prepare Dedicated API port mapping: invalid NodePort %d", selected.DedicatedAPI.NodePort)
	}
	return []string{fmt.Sprintf(
		"127.0.0.1:%d:%d@server:0",
		selected.DedicatedAPI.HostPort,
		selected.DedicatedAPI.NodePort,
	)}, nil
}

func clearLegacyPauseMarker(stateDir string) error {
	if err := os.Remove(filepath.Join(stateDir, legacyPauseMarkerFileName)); err != nil && !os.IsNotExist(err) {
		return fmt.Errorf("remove legacy pause marker: %w", err)
	}
	return nil
}

func rewriteLocalKubeconfigEndpoint(path string) error {
	data, err := os.ReadFile(path)
	if err != nil {
		return fmt.Errorf("read kubeconfig: %w", err)
	}

	var doc yaml.Node
	if err := yaml.Unmarshal(data, &doc); err != nil {
		return fmt.Errorf("parse kubeconfig: %w", err)
	}
	if !rewriteKubeconfigServerNodes(&doc) {
		return nil
	}
	out, err := yaml.Marshal(&doc)
	if err != nil {
		return fmt.Errorf("marshal kubeconfig: %w", err)
	}
	if err := os.WriteFile(path, out, 0o600); err != nil {
		return fmt.Errorf("write kubeconfig: %w", err)
	}
	return nil
}

func rewriteKubeconfigServerNodes(node *yaml.Node) bool {
	if node == nil {
		return false
	}
	changed := false
	if node.Kind == yaml.MappingNode {
		for i := 0; i+1 < len(node.Content); i += 2 {
			key := node.Content[i]
			value := node.Content[i+1]
			if key.Value == "server" && value.Kind == yaml.ScalarNode {
				if next, ok := localKubeconfigServer(value.Value); ok {
					value.Value = next
					changed = true
				}
			}
			if rewriteKubeconfigServerNodes(value) {
				changed = true
			}
		}
		return changed
	}
	for _, child := range node.Content {
		if rewriteKubeconfigServerNodes(child) {
			changed = true
		}
	}
	return changed
}

func localKubeconfigServer(server string) (string, bool) {
	u, err := url.Parse(strings.TrimSpace(server))
	if err != nil {
		return "", false
	}
	if !strings.EqualFold(u.Hostname(), "host.docker.internal") {
		return "", false
	}
	if port := u.Port(); port != "" {
		u.Host = net.JoinHostPort("127.0.0.1", port)
	} else {
		u.Host = "127.0.0.1"
	}
	return u.String(), true
}

func recordingStateVolume(selected *host.Session) (string, error) {
	if err := os.MkdirAll(selected.StateDir, 0o755); err != nil {
		return "", err
	}
	absStateDir, err := filepath.Abs(selected.StateDir)
	if err != nil {
		return "", err
	}
	return absStateDir + ":" + stub.RecordingNodeStateDir + "@all", nil
}

// NewClusterStep constructs the real cluster_create implementation for local-k3d.
func NewClusterStep(selected *host.Session, opts Options) *ClusterStep {
	return &ClusterStep{
		Selected: selected,
		Options:  opts,
		Adapter:  NewAdapter(),
	}
}