package migration

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/url"
	"reflect"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	appsclient "k8s.io/client-go/kubernetes/typed/apps/v1"
	coreclient "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/rest"

	"github.com/tidbcloud/observability/vm-migration/internal/retry"
)

// Deployment apply actions reported by Ensure and the deploy artifact.
const (
	ActionCreated = "created"
	ActionUpdated = "updated"
	ActionAbsent  = "absent"
	ActionDeleted = "deleted"
)

var (
	// Test-tunable knobs.
	kubePollInterval = 2 * time.Second
	kubeRetryPolicy  = retry.Policy{MaxAttempts: 3, Backoff: retry.ExponentialBackoff(time.Second, 4*time.Second)}
	// kubeRequestTimeout bounds every single REST call so a blackholed
	// connection cannot stall the poll loops past their deadlines.
	kubeRequestTimeout = 30 * time.Second
	// kubeConflictAttempts bounds the GET→PUT loop when another writer bumps
	// the resourceVersion between the read and the write.
	kubeConflictAttempts = 3
)

// VmselectProvisioner applies and removes the temporary vmselects on the
// cluster.
//
// Implementations must be safe for concurrent use: one provisioner is
// shared across all migrations and single-flight is enforced per migration
// only. Ensure and Remove are idempotent.
type VmselectProvisioner interface {
	// Ensure applies the Deployment (converge) and the Service
	// (create-if-absent) and waits for the Deployment rollout within
	// timeout, reporting progress lines to progress. It returns the
	// Deployment action taken: created or updated.
	Ensure(ctx context.Context, spec VmselectSpec, timeout time.Duration, progress io.Writer) (string, error)
	// Remove deletes the Deployment and Service (absent objects count as
	// success) and waits until both are gone within timeout. It returns the
	// Deployment action: deleted or absent.
	Remove(ctx context.Context, namespace, name string, timeout time.Duration, progress io.Writer) (string, error)
}

// NewKubeProvisioner builds a provisioner talking to the API server of cfg.
//
// The transport reloads the bearer token when its file changes, so
// kubelet-rotated projected service-account tokens keep working for the
// lifetime of a serve process.
func NewKubeProvisioner(cfg *rest.Config) (VmselectProvisioner, error) {
	cfg = rest.CopyConfig(cfg)
	cfg.Timeout = kubeRequestTimeout
	// Pin the wire format to JSON: requests stay debuggable and match the
	// rendered artifacts; protobuf buys nothing for a handful of objects.
	cfg.ContentType = "application/json"
	apps, err := appsclient.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	core, err := coreclient.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	return &kubeClient{apps: apps, core: core}, nil
}

type kubeClient struct {
	apps appsclient.AppsV1Interface
	core coreclient.CoreV1Interface
}

// Ensure applies the Service as create-if-absent and the Deployment as a
// converge PUT, then polls the Deployment rollout pinned to the generation
// of the apply response.
func (c *kubeClient) Ensure(ctx context.Context, spec VmselectSpec, timeout time.Duration, progress io.Writer) (string, error) {
	action, generation, err := c.applyDeployment(ctx, spec)
	if err != nil {
		return "", err
	}
	fmt.Fprintf(progress, "deployment %s/%s %s\n", spec.Namespace, spec.Name, action)
	if err := c.applyService(ctx, spec, progress); err != nil {
		return "", err
	}
	if err := c.waitReady(ctx, spec.Namespace, spec.Name, generation, spec.Replicas, timeout, progress); err != nil {
		return "", err
	}
	fmt.Fprintf(progress, "deployment %s/%s ready\n", spec.Namespace, spec.Name)
	return action, nil
}

// Remove deletes both objects with foreground propagation and waits until
// they are gone, so a re-deploy never hits a Terminating object.
func (c *kubeClient) Remove(ctx context.Context, namespace, name string, timeout time.Duration, progress io.Writer) (string, error) {
	foreground := metav1.DeletePropagationForeground
	deleteOptions := metav1.DeleteOptions{PropagationPolicy: &foreground}
	action := ActionAbsent
	for _, target := range []struct {
		kind   string
		delete func(ctx context.Context, opts metav1.DeleteOptions) error
	}{
		{"deployment", func(ctx context.Context, opts metav1.DeleteOptions) error {
			return c.apps.Deployments(namespace).Delete(ctx, name, opts)
		}},
		{"service", func(ctx context.Context, opts metav1.DeleteOptions) error {
			return c.core.Services(namespace).Delete(ctx, name, opts)
		}},
	} {
		err := kubeRetry(ctx, func() error { return target.delete(ctx, deleteOptions) })
		switch {
		case err == nil:
			if target.kind == "deployment" {
				action = ActionDeleted
			}
			fmt.Fprintf(progress, "%s %s/%s deleted\n", target.kind, namespace, name)
		case apierrors.IsNotFound(err):
			fmt.Fprintf(progress, "%s %s/%s absent\n", target.kind, namespace, name)
		default:
			return "", fmt.Errorf("delete %s %s/%s: %w", target.kind, namespace, name, err)
		}
	}
	deadline := time.Now().Add(timeout)
	for {
		depGone, err := c.deploymentAbsent(ctx, namespace, name)
		if err != nil {
			return "", err
		}
		svcGone, err := c.serviceAbsent(ctx, namespace, name)
		if err != nil {
			return "", err
		}
		if depGone && svcGone {
			return action, nil
		}
		if time.Now().After(deadline) {
			return "", fmt.Errorf("%s/%s still terminating after %s", namespace, name, timeout)
		}
		select {
		case <-ctx.Done():
			return "", ctx.Err()
		case <-time.After(kubePollInterval):
		}
	}
}

// applyDeployment converges the Deployment to spec. An update replaces the
// spec and labels; the rest of the stored object (resourceVersion and
// server-assigned metadata) round-trips. Returns the action and the
// generation of the apply response.
func (c *kubeClient) applyDeployment(ctx context.Context, spec VmselectSpec) (string, int64, error) {
	want := spec.Deployment()
	deployments := c.apps.Deployments(spec.Namespace)
	for attempt := 0; attempt < kubeConflictAttempts; attempt++ {
		existing, err := kubeCall(ctx, func() (*appsv1.Deployment, error) {
			return deployments.Get(ctx, spec.Name, metav1.GetOptions{})
		})
		if apierrors.IsNotFound(err) {
			created, err := kubeCall(ctx, func() (*appsv1.Deployment, error) {
				return deployments.Create(ctx, &want, metav1.CreateOptions{})
			})
			if err != nil {
				return "", 0, fmt.Errorf("create deployment %s/%s: %w", spec.Namespace, spec.Name, err)
			}
			return ActionCreated, created.Generation, nil
		}
		if err != nil {
			return "", 0, fmt.Errorf("get deployment %s/%s: %w", spec.Namespace, spec.Name, err)
		}
		existing.Spec = want.Spec
		existing.Labels = want.Labels
		updated, err := kubeCall(ctx, func() (*appsv1.Deployment, error) {
			return deployments.Update(ctx, existing, metav1.UpdateOptions{})
		})
		switch {
		case err == nil:
			return ActionUpdated, updated.Generation, nil
		case apierrors.IsConflict(err):
			continue
		default:
			return "", 0, fmt.Errorf("update deployment %s/%s: %w", spec.Namespace, spec.Name, err)
		}
	}
	return "", 0, fmt.Errorf("apply deployment %s/%s: resourceVersion conflict persisted after %d attempts",
		spec.Namespace, spec.Name, kubeConflictAttempts)
}

// applyService is create-if-absent: the Service spec is invariant, so an
// existing object is validated, never updated. Mismatched selector or ports
// fail loudly — drift can only be external.
func (c *kubeClient) applyService(ctx context.Context, spec VmselectSpec, progress io.Writer) error {
	want := spec.Service()
	services := c.core.Services(spec.Namespace)
	existing, err := kubeCall(ctx, func() (*corev1.Service, error) {
		return services.Get(ctx, spec.Name, metav1.GetOptions{})
	})
	switch {
	case err == nil:
		if !serviceMatches(existing, &want) {
			return fmt.Errorf("service %s/%s exists with unexpected selector or ports; delete it or reconcile manually",
				spec.Namespace, spec.Name)
		}
		fmt.Fprintf(progress, "service %s/%s present\n", spec.Namespace, spec.Name)
		return nil
	case apierrors.IsNotFound(err):
		if _, err := kubeCall(ctx, func() (*corev1.Service, error) {
			fresh := spec.Service()
			return services.Create(ctx, &fresh, metav1.CreateOptions{})
		}); err != nil {
			return fmt.Errorf("create service %s/%s: %w", spec.Namespace, spec.Name, err)
		}
		fmt.Fprintf(progress, "service %s/%s created\n", spec.Namespace, spec.Name)
		return nil
	default:
		return fmt.Errorf("get service %s/%s: %w", spec.Namespace, spec.Name, err)
	}
}

// serviceMatches compares the fields the rendered Service owns: the
// selector and each port's name, number, and target. Server-assigned fields
// (clusterIP, protocol defaults, session affinity) are ignored.
func serviceMatches(existing, want *corev1.Service) bool {
	if !reflect.DeepEqual(existing.Spec.Selector, want.Spec.Selector) ||
		len(existing.Spec.Ports) != len(want.Spec.Ports) {
		return false
	}
	for i := range want.Spec.Ports {
		e, w := existing.Spec.Ports[i], want.Spec.Ports[i]
		if e.Name != w.Name || e.Port != w.Port || e.TargetPort.String() != w.TargetPort.String() {
			return false
		}
	}
	return true
}

// waitReady polls until the Deployment controller has observed the apply
// generation and that many pods are available. Pinning the generation keeps
// a concurrent writer's stale-ready status from ending the wait early.
func (c *kubeClient) waitReady(ctx context.Context, namespace, name string, generation int64, replicas int, timeout time.Duration, progress io.Writer) error {
	deployments := c.apps.Deployments(namespace)
	deadline := time.Now().Add(timeout)
	for {
		dep, err := kubeCall(ctx, func() (*appsv1.Deployment, error) {
			return deployments.Get(ctx, name, metav1.GetOptions{})
		})
		if apierrors.IsNotFound(err) {
			return fmt.Errorf("deployment %s/%s vanished during rollout wait", namespace, name)
		}
		if err != nil {
			return fmt.Errorf("get deployment %s/%s: %w", namespace, name, err)
		}
		if dep.Status.ObservedGeneration >= generation && dep.Status.AvailableReplicas >= int32(replicas) {
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("deployment %s/%s not ready after %s", namespace, name, timeout)
		}
		fmt.Fprintf(progress, "waiting: %d/%d available (observedGeneration %d/%d)\n",
			dep.Status.AvailableReplicas, replicas, dep.Status.ObservedGeneration, generation)
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(kubePollInterval):
		}
	}
}

func (c *kubeClient) deploymentAbsent(ctx context.Context, namespace, name string) (bool, error) {
	_, err := kubeCall(ctx, func() (*appsv1.Deployment, error) {
		return c.apps.Deployments(namespace).Get(ctx, name, metav1.GetOptions{})
	})
	return objectAbsent(err)
}

func (c *kubeClient) serviceAbsent(ctx context.Context, namespace, name string) (bool, error) {
	_, err := kubeCall(ctx, func() (*corev1.Service, error) {
		return c.core.Services(namespace).Get(ctx, name, metav1.GetOptions{})
	})
	return objectAbsent(err)
}

func objectAbsent(err error) (bool, error) {
	switch {
	case err == nil:
		return false, nil
	case apierrors.IsNotFound(err):
		return true, nil
	default:
		return false, err
	}
}

// kubeRetry runs call under kubeRetryPolicy. Transport failures and 5xx
// responses are transient; every other error aborts immediately.
func kubeRetry(ctx context.Context, call func() error) error {
	return retry.Do(ctx, kubeRetryPolicy, func() error {
		err := call()
		if err != nil && kubeTransient(err) {
			return &kubeTransientError{err: err}
		}
		return err
	})
}

// kubeCall is kubeRetry for calls that return a value.
func kubeCall[T any](ctx context.Context, call func() (T, error)) (T, error) {
	var out T
	err := kubeRetry(ctx, func() error {
		var err error
		out, err = call()
		return err
	})
	return out, err
}

// kubeTransient reports whether err is worth retrying: transport failures
// surface as *url.Error, server-side 5xx as API status errors.
func kubeTransient(err error) bool {
	var urlErr *url.Error
	if errors.As(err, &urlErr) {
		return true
	}
	return apierrors.IsInternalError(err) ||
		apierrors.IsServiceUnavailable(err) ||
		apierrors.IsServerTimeout(err) ||
		apierrors.IsTimeout(err)
}

// kubeTransientError marks an error as always retryable for retry.Do.
type kubeTransientError struct {
	err error
}

func (e *kubeTransientError) Error() string { return e.err.Error() }

func (e *kubeTransientError) Is(target error) bool { return target == retry.ErrTransient }

func (e *kubeTransientError) Unwrap() error { return e.err }