package common

import (
	"context"
	"sort"
	"strings"
	"time"

	o11yv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

// Provider injects cloud-specific fields into the KubernetesAgentSpec.
type Provider interface {
	// FillProviderSpec reads provider-specific cluster resources and fills cloud-only fields.
	FillProviderSpec(ctx context.Context, t *task.BaseTask, spec *o11yv2.KubernetesAgentSpec) error
}

// O11y is the common task to deploy the observability KubernetesAgent reused by ACK, AKS, and EKS.
type O11y struct {
	*task.BaseTask
	provider Provider
}

var _ deploy.Task = &O11y{}

func (t *O11y) Name() string { return "O11y" }

// NewO11y constructs the common O11y task via dependency injection of a Provider.
func NewO11y(ctx *task.Context, p Provider) *O11y {
	return &O11y{BaseTask: &task.BaseTask{Context: ctx}, provider: p}
}

func (t *O11y) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())

	annotations, res := t.ensureBYOCVPCPeering(ctx)
	if res != nil {
		return *res
	}

	appID := t.APPBox.Labels[v1alpha1.DataPlaneIDKey]
	if appID == "" {
		appID = t.APPBox.Name
	}
	tenantID := t.APPBox.Labels[consts.TenantLabelKey]

	spec := o11yv2.KubernetesAgentSpec{
		KubernetesClusterName: t.APPBox.Status.KubernetesClusterName,
		O11y: o11yv2.O11yConfigV2{
			BizType:       t.Config.APPBoxConfig.O11yBizType,
			TenantID:      tenantID,
			ApplicationID: appID,
		},
	}

	if err := t.provider.FillProviderSpec(ctx, t.BaseTask, &spec); err != nil {
		return deploy.Error(err)
	}

	annotations, err := t.mergeKubernetesAgentAnnotations(ctx, annotations)
	if err != nil {
		return deploy.Error(err)
	}
	annotations = setAnnotation(
		annotations,
		consts.AnnotationKeyO11yVectorHostNetwork,
		t.Config.APPBoxConfig.EnableO11yVectorHostNetwork,
	)

	nodeSpecs, err := t.collectDedicatedNodeSpecs(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	if nodeSpecs != "" {
		if annotations == nil {
			annotations = map[string]string{}
		}
		annotations[consts.AnnotationKeyDataplaneNodeSpecs] = nodeSpecs
	}

	k8sAgent := &o11yv2.KubernetesAgent{
		ObjectMeta: metav1.ObjectMeta{
			Name:        t.APPBox.Name,
			Namespace:   t.agentNamespace(),
			Labels:      t.APPBox.Labels,
			Annotations: annotations,
		},
		Spec: spec,
	}

	return deploy.ErrorOrDone(k8s.Apply(ctx, t.Local, k8sAgent))
}

func setAnnotation(annotations map[string]string, key string, enabled bool) map[string]string {
	if !enabled {
		delete(annotations, key)
		return annotations
	}
	if annotations == nil {
		annotations = map[string]string{}
	}
	annotations[key] = consts.AnnoValTrue
	return annotations
}

func (t *O11y) Delete(ctx context.Context) deploy.Result {
	k8sAgent := &o11yv2.KubernetesAgent{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.APPBox.Name,
			Namespace: t.agentNamespace(),
		},
	}
	// Block other deletion steps until the agent is fully deleted.
	if err := k8s.DeleteObjectCompleted(ctx, t.Local, k8sAgent); err != nil {
		return deploy.Panic(err)
	}
	return t.deleteBYOCVPCPeering(ctx)
}

// mergeKubernetesAgentAnnotations keeps existing KubernetesAgent annotations
// (including ACS-managed ones) and overlays the given ones.
func (t *O11y) mergeKubernetesAgentAnnotations(ctx context.Context, overlay map[string]string) (map[string]string, error) {
	existing := &o11yv2.KubernetesAgent{}
	err := t.Local.Get(ctx, client.ObjectKey{Name: t.APPBox.Name, Namespace: t.agentNamespace()}, existing)
	if err != nil && !apierrors.IsNotFound(err) {
		return nil, err
	}
	merged := map[string]string{}
	if err == nil {
		merged = util.MergeMaps(merged, existing.Annotations)
	}
	merged = util.MergeMaps(merged, overlay)
	if len(merged) == 0 {
		return nil, nil
	}
	return merged, nil
}

func (t *O11y) agentNamespace() string {
	if t.Config.APPBoxConfig.O11yAgentNamespace != "" {
		return t.Config.APPBoxConfig.O11yAgentNamespace
	}
	return t.APPBox.Namespace
}

// collectDedicatedNodeSpecs returns a sorted, comma-separated list of Dedicated
// worker node-spec values for this dataplane (e.g. "2C8G,4C8G,8C16G").
//
// Only Dedicated NodePools are included. Admin nodes (created by AppBox, NodeType
// admin/shared) do not carry the node-spec label and are intentionally not labeled
// with one. ACS deploys a fallback o11y DaemonSet without nodeSelector to cover
// admin nodes, so they are out of scope for this annotation.
func (t *O11y) collectDedicatedNodeSpecs(ctx context.Context) (string, error) {
	dataplaneID := t.APPBox.Labels[v1alpha1.DataPlaneIDKey]
	if dataplaneID == "" {
		return "", nil
	}

	ns := t.Config.Namespace
	if ns == "" {
		ns = t.APPBox.Namespace
	}

	var list v1alpha1.NodePoolList
	if err := t.Local.List(ctx, &list,
		client.InNamespace(ns),
		client.MatchingLabels{v1alpha1.DataPlaneIDKey: dataplaneID},
	); err != nil {
		return "", err
	}

	specs := map[string]struct{}{}
	for i := range list.Items {
		np := &list.Items[i]
		if !strings.EqualFold(np.Spec.NodeType, consts.NodeTypeDedicated) {
			continue
		}
		if np.Spec.CpuCore == 0 && np.Spec.MemoryGi == 0 {
			continue
		}
		specs[nodepool.NodeSpecLabel(np.Spec.CpuCore, np.Spec.MemoryGi)] = struct{}{}
	}
	if len(specs) == 0 {
		return "", nil
	}

	sorted := make([]string, 0, len(specs))
	for s := range specs {
		sorted = append(sorted, s)
	}
	sort.Strings(sorted)
	return strings.Join(sorted, ","), nil
}

// Provider implementations are defined in cloud-specific packages (ack/eks).