/*
Copyright 2024.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package legacy

import (
	"context"
	"fmt"
	"slices"
	"strings"

	o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/deploy"
	task "github.com/tidbcloud/observability/pkg/resourceworker/controllers/integrationtask/v2/task"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

const (
	deployNameTmpl         = "etl-job-manager-%s"
	serviceAccountNameTmpl = "etl-job-manager-%s"
	clusterDataPathTmpl    = "%s/log_data/data/%s/%s/%s"
)

type Deployment struct {
	*task.BaseTask
}

func NewDeployment(ctx *task.Context) *Deployment {
	return &Deployment{BaseTask: &task.BaseTask{Context: ctx}}
}

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

func (t *Deployment) Create(ctx context.Context) deploy.Result {
	namespace := t.Task.Namespace

	err := t.createOrUpdateETLJobManager(ctx, t.Task, namespace)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *Deployment) Delete(ctx context.Context) deploy.Result {
	id, err := t.Identity()
	if err != nil {
		return deploy.Error(err)
	}
	namespace := id.Namespace

	if id.IsTenant() {
		// Tenant-scope resources are 1:1 with the (tenant, dataType) pipeline, so
		// delete them directly without cross-counting other tasks in the namespace.
		if err := t.deleteJobManager(ctx, namespace, id.ResourceSuffix()); err != nil {
			return deploy.Error(err)
		}
		return deploy.Done()
	}

	// Cluster scope: the Deployment/ServiceAccount are shared per tenant. Only
	// delete them when no other cluster-scoped task for this tenant remains.
	// Tenant-scoped tasks are isolated pipelines and must not gate this cleanup.
	taskList := &o11yv1.IntegrationTaskList{}
	if err := t.Client.List(ctx, taskList, client.InNamespace(namespace)); err != nil {
		return deploy.Error(fmt.Errorf("failed to list integration tasks: %w", err))
	}

	activeTasks := 0
	for _, other := range taskList.Items {
		if other.Name == t.Task.Name {
			continue
		}
		if other.Spec.Scope == o11yv1.ScopeTenant {
			continue
		}
		if other.Status.Phase != constants.DELETING {
			activeTasks++
		}
	}

	if activeTasks == 0 {
		logger := log.FromContext(ctx)
		logger.Info("Cleaning up shared Deployment and ServiceAccount for tenant", "tenant", id.TenantID, "namespace", namespace)
		if err := t.deleteJobManager(ctx, namespace, id.ResourceSuffix()); err != nil {
			return deploy.Error(err)
		}
	}

	return deploy.Done()
}

// deleteJobManager deletes the ETL job manager Deployment and ServiceAccount for a
// given resource-name suffix. It is used both for tenant-scope direct cleanup and
// cluster-scope last-task cleanup.
func (t *Deployment) deleteJobManager(ctx context.Context, namespace, suffix string) error {
	logger := log.FromContext(ctx)

	deploymentName := fmt.Sprintf(deployNameTmpl, suffix)
	deployment := &appsv1.Deployment{}
	err := t.Client.Get(ctx, client.ObjectKey{Name: deploymentName, Namespace: namespace}, deployment)
	if err == nil {
		if err := t.Client.Delete(ctx, deployment); err != nil {
			return fmt.Errorf("failed to delete Deployment: %w", err)
		}
	} else if client.IgnoreNotFound(err) != nil {
		return fmt.Errorf("failed to get Deployment: %w", err)
	}

	saName := fmt.Sprintf(serviceAccountNameTmpl, suffix)
	serviceAccount := &corev1.ServiceAccount{}
	err = t.Client.Get(ctx, client.ObjectKey{Name: saName, Namespace: namespace}, serviceAccount)
	if err == nil {
		if err := t.Client.Delete(ctx, serviceAccount); err != nil {
			return fmt.Errorf("failed to delete ServiceAccount: %w", err)
		}
	} else if client.IgnoreNotFound(err) != nil {
		return fmt.Errorf("failed to get ServiceAccount: %w", err)
	}

	logger.Info("cleaned up job manager", "deploymentName", deploymentName, "serviceAccountName", saName, "namespace", namespace)
	return nil
}

// createOrUpdateETLJobManager creates or updates the ETL job manager Deployment and ServiceAccount
func (t *Deployment) createOrUpdateETLJobManager(ctx context.Context, task *o11yv1.IntegrationTask, namespace string) error {
	logger := log.FromContext(ctx)
	// Scope-aware naming: cluster scope keeps the per-tenant name, tenant scope
	// gets a per-(tenant, dataType) name so pipelines are isolated.
	id, idErr := t.Identity()
	if idErr != nil {
		return idErr
	}
	suffix := id.ResourceSuffix()
	deploymentName := fmt.Sprintf(deployNameTmpl, suffix)

	saName := fmt.Sprintf(serviceAccountNameTmpl, suffix)

	// Create or update ServiceAccount for ETL job manager
	serviceAccount := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      saName,
			Namespace: namespace,
		},
	}

	_, err := controllerutil.CreateOrUpdate(ctx, t.Client, serviceAccount, func() error {
		// Set annotations so the job-manager pod assumes its cloud IAM identity.
		if serviceAccount.Annotations == nil {
			serviceAccount.Annotations = make(map[string]string)
		}

		switch t.ServiceInfo.Vendor {
		case "aws":
			// IRSA: use VectorRoleARN (the regional-server's own role) as the ETL execution
			// identity. This role has write access to the external export bucket. The tenant's
			// accessRoleArn is a separate external role trusted by o11y-ext-<tenantID> for
			// READ-ONLY access by the tenant — it must NOT be used as the ETL identity.
			if roleARN := t.ServiceInfo.ControllerConfig.IntegrationTaskConfig.VectorRoleARN; roleARN != "" {
				serviceAccount.Annotations[constants.AnnotationServiceAccountRole] = roleARN
			}
		case "alicloud":
			serviceAccount.Annotations["pod-identity.alibabacloud.com/role-name"] = t.ServiceInfo.ControllerConfig.IntegrationTaskConfig.JobManagerIAMRole
		}
		return nil
	})
	if err != nil {
		return fmt.Errorf("failed to create or update ServiceAccount: %w", err)
	}

	logger.Info("ServiceAccount created/updated successfully", "serviceAccountName", saName)

	// Check if Deployment already exists
	key := client.ObjectKey{Name: deploymentName, Namespace: namespace}
	existingDeployment := &appsv1.Deployment{}
	expectedImage := t.ServiceInfo.ControllerConfig.IntegrationTaskConfig.JobManagerImage

	err = t.Client.Get(ctx, key, existingDeployment)
	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			// Deployment doesn't exist, create it
			logger.Info("Creating new Deployment for slow query ETL job manager", "deploymentName", deploymentName, "namespace", namespace, "vendor", t.ServiceInfo.Vendor)
			// Build container args; -shareDataPrefix controls the checkpoint marker and
			// output key prefix within the destination bucket.
			containerArgs := buildETLContainerArgs(task, t.ServiceInfo.Vendor, t.ServiceInfo.Region)

			replicas := int32(1)
			deployment := &appsv1.Deployment{
				ObjectMeta: metav1.ObjectMeta{
					Name:      deploymentName,
					Namespace: namespace,
					Labels: map[string]string{
						"app": "etl-job-manager",
					},
				},
				Spec: appsv1.DeploymentSpec{
					Replicas: &replicas,
					Selector: &metav1.LabelSelector{
						MatchLabels: map[string]string{
							"app": "etl-job-manager",
						},
					},
					Template: corev1.PodTemplateSpec{
						ObjectMeta: metav1.ObjectMeta{
							Labels: map[string]string{
								"app": "etl-job-manager",
							},
						},
						Spec: corev1.PodSpec{
							ServiceAccountName: saName,
							RestartPolicy:      corev1.RestartPolicyAlways,
							Containers: []corev1.Container{
								{
									Name:  "etl-job-manager",
									Image: expectedImage,
									Args:  containerArgs,
									VolumeMounts: []corev1.VolumeMount{
										{
											Name:      "config",
											MountPath: "/etc/config",
											ReadOnly:  true,
										},
									},
									Resources: corev1.ResourceRequirements{
										Requests: corev1.ResourceList{
											corev1.ResourceCPU:    resource.MustParse("100m"),
											corev1.ResourceMemory: resource.MustParse("128Mi"),
										},
										Limits: corev1.ResourceList{
											corev1.ResourceCPU:    resource.MustParse("500m"),
											corev1.ResourceMemory: resource.MustParse("512Mi"),
										},
									},
									// LivenessProbe: &corev1.Probe{
									// 	ProbeHandler: corev1.ProbeHandler{
									// 		HTTPGet: &corev1.HTTPGetAction{
									// 			Path: "/health",
									// 			Port: intstr.FromInt(8080),
									// 		},
									// 	},
									// 	InitialDelaySeconds: 30,
									// 	PeriodSeconds:       10,
									// },
									// ReadinessProbe: &corev1.Probe{
									// 	ProbeHandler: corev1.ProbeHandler{
									// 		HTTPGet: &corev1.HTTPGetAction{
									// 			Path: "/ready",
									// 			Port: intstr.FromInt(8080),
									// 		},
									// 	},
									// 	InitialDelaySeconds: 5,
									// 	PeriodSeconds:       5,
									// },
								},
							},
							Volumes: []corev1.Volume{
								{
									Name: "config",
									VolumeSource: corev1.VolumeSource{
										ConfigMap: &corev1.ConfigMapVolumeSource{
											LocalObjectReference: corev1.LocalObjectReference{
												Name: fmt.Sprintf(configMapNameTmpl, suffix),
											},
										},
									},
								},
							},
						},
					},
				},
			}

			err = t.Client.Create(ctx, deployment)
			if err != nil {
				return fmt.Errorf("failed to create Deployment: %w", err)
			}
		} else {
			return err
		}
	} else {
		// Deployment exists, check if it needs to be updated
		logger.Info("Deployment already exists for slow query ETL job manager", "deploymentName", deploymentName, "namespace", namespace, "vendor", t.ServiceInfo.Vendor)

		// Update the deployment if needed (e.g., if image or config changed)
		needsUpdate := false
		expectedArgs := buildETLContainerArgs(task, t.ServiceInfo.Vendor, t.ServiceInfo.Region)

		// Check if vendor-specific image needs updating

		if len(existingDeployment.Spec.Template.Spec.Containers) > 0 {
			if existingDeployment.Spec.Template.Spec.Containers[0].Image != expectedImage {
				existingDeployment.Spec.Template.Spec.Containers[0].Image = expectedImage
				needsUpdate = true
			}
			if !slices.Equal(existingDeployment.Spec.Template.Spec.Containers[0].Args, expectedArgs) {
				existingDeployment.Spec.Template.Spec.Containers[0].Args = expectedArgs
				needsUpdate = true
			}
		}

		if needsUpdate {
			err = t.Client.Update(ctx, existingDeployment)
			if err != nil {
				return fmt.Errorf("failed to update Deployment: %w", err)
			}
			logger.Info("Updated Deployment for slow query ETL job manager", "deploymentName", deploymentName)
		}
	}

	return nil
}

func buildETLContainerArgs(task *o11yv1.IntegrationTask, vendor, region string) []string {
	containerArgs := []string{
		fmt.Sprintf("-srcBucket=%s", task.Spec.SourceBucketName),
		fmt.Sprintf("-srcPrefix=%s", formatSrcPrefixForVendor(task, vendor)),
		fmt.Sprintf("-dstBucket=%s", task.Spec.DestinationBucketName),
		fmt.Sprintf("-vendor=%s", vendor),
		fmt.Sprintf("-region=%s", region),
		fmt.Sprintf("-configPath=%s", "/etc/config/"+configFileName),
	}
	if task.Spec.DataPrefix != "" {
		containerArgs = append(containerArgs, fmt.Sprintf("-shareDataPrefix=%s", task.Spec.DataPrefix))
	}
	return containerArgs
}

func formatSrcPrefixForVendor(it *o11yv1.IntegrationTask, vendor string) string {
	srcPrefix := resolveSrcPrefix(it)
	if !shouldIncludeSourceBucketInSrcPrefix(it, vendor) {
		return srcPrefix
	}
	return strings.TrimSuffix(it.Spec.SourceBucketName, "/") + "/" + strings.TrimPrefix(srcPrefix, "/")
}

func shouldIncludeSourceBucketInSrcPrefix(it *o11yv1.IntegrationTask, vendor string) bool {
	if vendor != string(constants.AliCloud) || it.Spec.SourceBucketName == "" {
		return false
	}
	// Alicloud serverless keeps a legacy source layout where the ETL expects the
	// bucket name in srcPrefix; premium and dedicated use bucket-internal prefixes.
	return resolveDeployType(it.Labels[constants.LabelBizType]) == task.DeployTypeServerless
}

// buildS3Path builds the storage path for slow query logs based on cloud vendor
func (t *Deployment) buildS3Path(task *o11yv1.IntegrationTask) string {
	dataPath := fmt.Sprintf(clusterDataPathTmpl, task.Spec.DestinationBucketName, task.Labels[constants.LabelTenantID], task.Labels[constants.LabelProjectID], task.Spec.ClusterID)
	switch t.ServiceInfo.Vendor {
	case "aws":
		return fmt.Sprintf("s3://%s", dataPath)
	case "alicloud":
		return fmt.Sprintf("oss://%s", dataPath)
	default:
		return fmt.Sprintf("s3://%s", dataPath)
	}
}