/*
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 integrationtask

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strings"

	"github.com/go-logr/logr"
	o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/resourcemanager"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/iam"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	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"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

// IntegrationTaskReconciler reconciles an IntegrationTask object
type IntegrationTaskReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

const (
	integrationTaskFinalizerName = "observability.pingcap.com/integrationtask"
	errNotIntegrationTask        = "not IntegrationTask resource"
	iamRolePolicyNameTmpl        = "etl-bucket-access-policy-%s-%s" // logicalRegion + tenantID
	configMapNameTmpl            = "job-manager-config-%s"          // config-<tenant>
	configFileName               = "config.json"
	deployNameTmpl               = "etl-job-manager-%s"
	serverlessDataPrefixTmpl     = "0/%s/serverless"
	serviceAccountNameTmpl       = "etl-job-manager-%s" // etl-job-manager-<tenant>
	tenantDataPrefixTmpl         = "log_data/data/%s"
	clusterDataPathTmpl          = "%s/log_data/data/%s/%s/%s"
)

func NewIntegrationTaskReconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *IntegrationTaskReconciler {
	return &IntegrationTaskReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: integrationTaskConnector{
				cli:         mgr.GetClient(),
				serviceInfo: serviceInfo,
			},
			NewResource: func() client.Object {
				return &o11yv1.IntegrationTask{}
			},
		},
	}
}

type integrationTaskConnector struct {
	serviceInfo structs.ServiceInfo
	cli         client.Client
}

func (itc integrationTaskConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	_, ok := obj.(*o11yv1.IntegrationTask)
	if !ok {
		return nil, fmt.Errorf(errNotIntegrationTask)
	}

	// Get S3 client
	s3Client := resourcemanager.GetS3Client()
	if s3Client == nil {
		return nil, fmt.Errorf("storage client not available")
	}

	return &IntegrationTask{
		cli:         itc.cli,
		s3Cli:       s3Client,
		serviceInfo: itc.serviceInfo,
	}, nil
}

// IntegrationTask handles the business logic for integration tasks
type IntegrationTask struct {
	cli         client.Client
	s3Cli       providers.S3Client
	serviceInfo structs.ServiceInfo
}

func (it *IntegrationTask) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	task, ok := obj.(*o11yv1.IntegrationTask)
	if !ok {
		return reconciler.ExternalObservation{Droped: true}, fmt.Errorf(errNotIntegrationTask)
	}

	log := log.FromContext(ctx)
	log.Info("Observing integration task", "name", task.Name, "phase", task.Status.Phase)

	// Check if destination resources exist
	resourceStatus, err := it.checkDestinationResourcesExistence(ctx, task)
	if err != nil {
		return reconciler.ExternalObservation{}, fmt.Errorf("failed to check destination resources existence: %w", err)
	}

	// Determine if task is up to date
	upToDate := false
	if resourceStatus.ResourceExists && task.Status.Phase == constants.ACTIVE {
		upToDate, err = it.checkDestinationResourcesUpToDate(ctx, task)
		if err != nil {
			return reconciler.ExternalObservation{}, fmt.Errorf("failed to check if destination resources are up to date: %w", err)
		}
	}

	observation := reconciler.ExternalObservation{
		ResourceExists:     resourceStatus.ResourceExists,
		PartResourceExists: resourceStatus.PartResourceExists,
		ResourceUpToDate:   upToDate,
	}

	return observation, nil
}

func (it *IntegrationTask) Create(ctx context.Context, obj client.Object) error {
	task, ok := obj.(*o11yv1.IntegrationTask)
	if !ok {
		return fmt.Errorf(errNotIntegrationTask)
	}

	log := log.FromContext(ctx)
	log.Info("Creating integration task", "name", task.Name, "type", task.Spec.TaskType)

	// Initialize task status
	task.Status.Phase = constants.CREATING
	if err := it.cli.Status().Update(ctx, task); err != nil {
		return err
	}

	// Create destination resources based on task type
	if err := it.createDestinationResources(ctx, task); err != nil {
		task.Status.LastError = err.Error()
		if err := it.cli.Status().Update(ctx, task); err != nil {
			return err
		}
		return fmt.Errorf("failed to create destination resources: %w", err)
	}

	task.Status.Phase = constants.ACTIVE
	return nil
}

func (it *IntegrationTask) Update(ctx context.Context, obj client.Object) error {
	task, ok := obj.(*o11yv1.IntegrationTask)
	if !ok {
		return fmt.Errorf(errNotIntegrationTask)
	}

	log := log.FromContext(ctx)
	log.Info("Updating integration task")

	// Update destination resources if needed
	if err := it.updateDestinationResources(ctx, task); err != nil {
		task.Status.LastError = err.Error()
		if err := it.cli.Status().Update(ctx, task); err != nil {
			return err
		}
		return fmt.Errorf("failed to update destination resources: %w", err)
	}

	task.Status.Phase = constants.ACTIVE
	task.Status.LastError = ""
	if err := it.cli.Status().Update(ctx, task); err != nil {
		return err
	}
	return nil
}

func (it *IntegrationTask) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	task, ok := obj.(*o11yv1.IntegrationTask)
	if !ok {
		return reconciler.ExternalDeletion{}, fmt.Errorf(errNotIntegrationTask)
	}

	log := log.FromContext(ctx)
	log.Info("Deleting integration task")

	//update status
	if task.Status.Phase != constants.DELETING {
		task.Status.Phase = constants.DELETING
		if err := it.cli.Status().Update(ctx, task); err != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}

	// Clean up destination resources
	if err := it.deleteDestinationResources(ctx, task); err != nil {
		log.Error(err, "Failed to clean up destination resources", "name", task.Name)
		return reconciler.ExternalDeletion{}, err
	}

	return reconciler.ExternalDeletion{
		ResourceDeleted: true,
	}, nil
}

// createDestinationResources creates resources for slow query ETL tasks
func (it *IntegrationTask) createDestinationResources(ctx context.Context, task *o11yv1.IntegrationTask) error {
	log := log.FromContext(ctx)

	// Use task's namespace directly
	namespace := task.Namespace

	// Extract tenant and biz type from task labels or namespace
	tenantID, bizType := it.extractTenantInfo(task)

	// 1. Create or update IAM role for slow query ETL
	iamRole, err := it.createOrUpdateIAMRole(ctx, task, namespace, tenantID, bizType)
	if err != nil {
		return fmt.Errorf("failed to create IAM role: %w", err)
	}

	// 2. Create or update ConfigMap for job manager configuration
	err = it.createOrUpdateJobManagerConfig(ctx, task, namespace, tenantID)
	if err != nil {
		return fmt.Errorf("failed to create job manager config: %w", err)
	}

	// 3. Create or update Deployment for ETL job manager
	err = it.createOrUpdateETLJobManager(ctx, task, namespace)
	if err != nil {
		return fmt.Errorf("failed to create ETL job manager: %w", err)
	}

	// Update destination info in status
	task.Status.DestinationInfo = o11yv1.DestinationInfo{
		AccessPoint:   it.buildS3Path(task),
		AssumeRoleARN: iamRole.Status.Arn,
	}

	log.Info("Slow query ETL resources created successfully",
		"accessPoint", task.Status.DestinationInfo.AccessPoint,
		"assumeRole", task.Status.DestinationInfo.AssumeRoleARN,
		"namespace", namespace)

	return nil
}

// checkDestinationResourcesExistence checks if slow query ETL resources exist
func (it *IntegrationTask) checkDestinationResourcesExistence(ctx context.Context, task *o11yv1.IntegrationTask) (reconciler.SubResourceObservation, error) {
	log := log.FromContext(ctx)

	log.V(1).Info("Checking slow query ETL resources existence", "cluster", task.Spec.ClusterID)

	// Extract tenant info
	tenantID, _ := it.extractTenantInfo(task)

	// Track individual resource status
	var resourceChecks []reconciler.SubResourceObservation

	// Check IAM role
	iamStatus, err := it.checkIAMRoleResource(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return reconciler.SubResourceObservation{}, fmt.Errorf("failed to check IAM role resource: %w", err)
	}
	resourceChecks = append(resourceChecks, iamStatus)

	// Check ConfigMap
	configMapStatus, err := it.checkConfigMapResource(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return reconciler.SubResourceObservation{}, fmt.Errorf("failed to check ConfigMap resource: %w", err)
	}
	resourceChecks = append(resourceChecks, configMapStatus)

	// Check Deployment
	deploymentStatus, err := it.checkDeploymentResource(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return reconciler.SubResourceObservation{}, fmt.Errorf("failed to check Deployment resource: %w", err)
	}
	resourceChecks = append(resourceChecks, deploymentStatus)

	// Combine all resource checks
	return reconciler.Combine(resourceChecks...), nil
}

// checkDestinationResourcesUpToDate checks if slow query ETL resources are up to date
func (it *IntegrationTask) checkDestinationResourcesUpToDate(ctx context.Context, task *o11yv1.IntegrationTask) (bool, error) {
	log := log.FromContext(ctx)

	log.V(1).Info("Checking if slow query ETL resources are up to date", "cluster", task.Spec.ClusterID)

	// Extract tenant info
	tenantID, _ := it.extractTenantInfo(task)

	// Check if IAM role is up to date
	iamUpToDate, err := it.isIAMRoleUpToDate(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return false, fmt.Errorf("failed to check if IAM role is up to date: %w", err)
	}
	if !iamUpToDate {
		log.V(1).Info("IAM role is not up to date")
		return false, nil
	}

	// Check if ConfigMap is up to date
	configMapUpToDate, err := it.isConfigMapUpToDate(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return false, fmt.Errorf("failed to check if ConfigMap is up to date: %w", err)
	}
	if !configMapUpToDate {
		log.V(1).Info("ConfigMap is not up to date")
		return false, nil
	}

	// Check if Deployment is up to date
	deploymentUpToDate, err := it.isDeploymentUpToDate(ctx, task, task.Namespace, tenantID)
	if err != nil {
		return false, fmt.Errorf("failed to check if Deployment is up to date: %w", err)
	}
	if !deploymentUpToDate {
		log.V(1).Info("Deployment is not up to date")
		return false, nil
	}

	log.V(1).Info("All resources are up to date")
	return true, nil
}

// checkIAMRoleResource checks if the IAM role resource exists
func (it *IntegrationTask) checkIAMRoleResource(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (reconciler.SubResourceObservation, error) {
	log := log.FromContext(ctx)

	roleName := task.Labels[constants.LabelAlias]
	iamKey := client.ObjectKey{Name: roleName, Namespace: namespace}
	iamRole := &o11yv1.IAM{}
	err := it.cli.Get(ctx, iamKey, iamRole)

	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			log.Info("IAM role not found", "roleName", roleName)
			return reconciler.SubResourceObservation{ResourceExists: false}, nil
		}
		return reconciler.SubResourceObservation{}, fmt.Errorf("failed to get IAM role %s: %w", roleName, err)
	}

	return reconciler.SubResourceObservation{ResourceExists: true, PartResourceExists: true}, nil
}

// checkConfigMapResource checks if the ConfigMap resource exists
func (it *IntegrationTask) checkConfigMapResource(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (reconciler.SubResourceObservation, error) {
	log := log.FromContext(ctx)

	configMapName := fmt.Sprintf(configMapNameTmpl, tenantID)
	configMapKey := client.ObjectKey{Name: configMapName, Namespace: namespace}
	configMap := &corev1.ConfigMap{}
	err := it.cli.Get(ctx, configMapKey, configMap)

	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			log.V(1).Info("ConfigMap not found", "configMapName", configMapName)
			return reconciler.SubResourceObservation{ResourceExists: false}, nil
		}
		return reconciler.SubResourceObservation{ResourceExists: false}, fmt.Errorf("failed to get ConfigMap %s: %w", configMapName, err)
	}

	log.V(1).Info("ConfigMap resource exists", "configMapName", configMapName)
	return reconciler.SubResourceObservation{ResourceExists: true, PartResourceExists: true}, nil
}

// checkDeploymentResource checks if the Deployment resource exists
func (it *IntegrationTask) checkDeploymentResource(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (reconciler.SubResourceObservation, error) {
	log := log.FromContext(ctx)

	deploymentName := fmt.Sprintf(deployNameTmpl, tenantID)
	deploymentKey := client.ObjectKey{Name: deploymentName, Namespace: namespace}
	deployment := &appsv1.Deployment{}
	err := it.cli.Get(ctx, deploymentKey, deployment)

	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			log.V(1).Info("Deployment not found", "deploymentName", deploymentName)
			return reconciler.SubResourceObservation{ResourceExists: false}, nil
		}
		return reconciler.SubResourceObservation{ResourceExists: false}, fmt.Errorf("failed to get Deployment %s: %w", deploymentName, err)
	}

	log.V(1).Info("Deployment resource exists", "deploymentName", deploymentName)
	return reconciler.SubResourceObservation{ResourceExists: true, PartResourceExists: true}, nil
}

// isIAMRoleUpToDate checks if the IAM role configuration is up to date
func (it *IntegrationTask) isIAMRoleUpToDate(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (bool, error) {
	log := log.FromContext(ctx)

	roleName := task.Labels[constants.LabelAlias]
	iamKey := client.ObjectKey{Name: roleName, Namespace: namespace}
	iamRole := &o11yv1.IAM{}
	err := it.cli.Get(ctx, iamKey, iamRole)
	if err != nil {
		return false, fmt.Errorf("failed to get IAM role %s: %w", roleName, err)
	}

	if iamRole.Status.Phase != constants.ACTIVE {
		return false, nil
	}

	// Verify IAM role has the correct cluster-specific policy with proper content
	policyValid, err := it.verifyIAMRolePolicy(ctx, iamRole, task, tenantID)
	if err != nil {
		return false, fmt.Errorf("failed to verify IAM role policy: %w", err)
	}

	if !policyValid {
		log.V(1).Info("IAM role policy is not up to date", "clusterID", task.Spec.ClusterID)
		return false, nil
	}

	log.V(1).Info("IAM role is up to date", "roleName", roleName)
	return true, nil
}

// isConfigMapUpToDate checks if the ConfigMap configuration is up to date
func (it *IntegrationTask) isConfigMapUpToDate(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (bool, error) {
	log := log.FromContext(ctx)

	configMapName := fmt.Sprintf(configMapNameTmpl, tenantID)
	configMapKey := client.ObjectKey{Name: configMapName, Namespace: namespace}
	configMap := &corev1.ConfigMap{}
	err := it.cli.Get(ctx, configMapKey, configMap)
	if err != nil {
		return false, fmt.Errorf("failed to get ConfigMap %s: %w", configMapName, err)
	}

	// Check if the cluster configuration matches the expected configuration
	if configMap.Data == nil || configMap.Data[configFileName] == "" {
		log.Info("ConfigMap data is empty")
		return false, nil
	}

	var configData map[string]interface{}
	parseErr := json.Unmarshal([]byte(configMap.Data[configFileName]), &configData)
	if parseErr != nil {
		return false, fmt.Errorf("failed to parse ConfigMap data: %w", parseErr)
	}

	// Verify the cluster configuration exists in the ConfigMap
	if configData["tenants"] == nil {
		log.Info("No tenants found in ConfigMap")
		return false, nil
	}

	tenants, ok := configData["tenants"].(map[string]interface{})
	if !ok {
		log.Info("Invalid tenants format in ConfigMap")
		return false, nil
	}

	if tenants[tenantID] == nil {
		log.Info("Tenant not found in ConfigMap", "tenantID", tenantID)
		return false, nil
	}

	tenantClusters, ok := tenants[tenantID].(map[string]interface{})
	if !ok {
		log.Info("Invalid tenant clusters format in ConfigMap", "tenantID", tenantID)
		return false, nil
	}

	_, exists := tenantClusters[task.Spec.ClusterID]
	if !exists {
		log.Info("Cluster not found in ConfigMap", "clusterID", task.Spec.ClusterID)
		return false, nil
	}

	log.V(1).Info("ConfigMap is up to date", "configMapName", configMapName)
	return true, nil
}

// isDeploymentUpToDate checks if the Deployment configuration is up to date
func (it *IntegrationTask) isDeploymentUpToDate(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) (bool, error) {
	log := log.FromContext(ctx)

	deploymentName := fmt.Sprintf(deployNameTmpl, tenantID)
	deploymentKey := client.ObjectKey{Name: deploymentName, Namespace: namespace}
	deployment := &appsv1.Deployment{}
	err := it.cli.Get(ctx, deploymentKey, deployment)
	if err != nil {
		return false, fmt.Errorf("failed to get Deployment %s: %w", deploymentName, err)
	}

	// TODO: Add more detailed validation of deployment spec if needed
	// For now, we check basic readiness
	if deployment.Status.ReadyReplicas == 0 {
		log.V(1).Info("Deployment has no ready replicas")
		return false, nil
	}

	if deployment.Status.UpdatedReplicas != *deployment.Spec.Replicas {
		log.V(1).Info("Deployment update in progress",
			"updatedReplicas", deployment.Status.UpdatedReplicas,
			"desiredReplicas", *deployment.Spec.Replicas)
		return false, nil
	}

	log.V(1).Info("Deployment is up to date", "deploymentName", deploymentName)
	return true, nil
}

// updateETLResources updates slow query ETL resources
func (it *IntegrationTask) updateDestinationResources(ctx context.Context, task *o11yv1.IntegrationTask) error {
	log := log.FromContext(ctx)

	log.Info("Updating slow query ETL resources",
		"cluster", task.Spec.ClusterID)

	// Use task's namespace directly
	namespace := task.Namespace

	// Extract tenant info
	tenantID, bizType := it.extractTenantInfo(task)

	// Update IAM role if needed (add new policies or update existing ones)
	iam, err := it.createOrUpdateIAMRole(ctx, task, namespace, tenantID, bizType)
	if err != nil {
		return fmt.Errorf("failed to update IAM role: %w", err)
	}

	// Update ConfigMap with new configuration
	err = it.createOrUpdateJobManagerConfig(ctx, task, namespace, tenantID)
	if err != nil {
		return fmt.Errorf("failed to update job manager config: %w", err)
	}

	// Update Deployment if needed
	err = it.createOrUpdateETLJobManager(ctx, task, namespace)
	if err != nil {
		return fmt.Errorf("failed to update ETL job manager: %w", err)
	}

	// Update destination info in status
	task.Status.DestinationInfo.AccessPoint = it.buildS3Path(task)
	task.Status.DestinationInfo.AssumeRoleARN = iam.Status.Arn

	log.Info("Slow query ETL resources updated successfully")
	return nil
}

// deleteETLResources cleans up slow query ETL resources
func (it *IntegrationTask) deleteDestinationResources(ctx context.Context, task *o11yv1.IntegrationTask) error {
	log := log.FromContext(ctx)

	log.Info("Deleting slow query ETL resources",
		"cluster", task.Spec.ClusterID,
		"accessPoint", task.Status.DestinationInfo.AccessPoint)

	// Use task's namespace directly
	namespace := task.Namespace

	// Extract tenant info
	tenantID, _ := it.extractTenantInfo(task)

	// Remove cluster-specific configuration from ConfigMap
	err := it.removeClusterFromConfig(ctx, task, namespace, tenantID)
	if err != nil {
		log.Error(err, "Failed to remove cluster config from ConfigMap", "cluster", task.Spec.ClusterID)
		return err
	}

	// Remove cluster-specific policy from IAM role
	err = it.removeClusterFromIAMRole(ctx, task, namespace, tenantID)
	if err != nil {
		log.Error(err, "Failed to remove cluster policy from IAM role", "cluster", task.Spec.ClusterID)
		return err
	}

	// Check if this is the last cluster for this tenant, and if so, clean up shared resources
	err = it.cleanupSharedResourcesIfNeeded(ctx, task, namespace, tenantID)
	if err != nil {
		log.Error(err, "Failed to cleanup shared resources", "tenant", tenantID)
		return err
	}

	log.Info("Slow query ETL resources cleaned up successfully")
	return nil
}

// removeClusterFromConfig removes cluster-specific configuration from the shared ConfigMap
func (it *IntegrationTask) removeClusterFromConfig(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) error {
	log := log.FromContext(ctx)

	configMapName := fmt.Sprintf(configMapNameTmpl, tenantID)
	key := client.ObjectKey{Name: configMapName, Namespace: namespace}

	configMap := &corev1.ConfigMap{}
	err := it.cli.Get(ctx, key, configMap)
	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			// ConfigMap doesn't exist, nothing to clean up
			return nil
		}
		return fmt.Errorf("failed to get ConfigMap: %w", err)
	}

	// Parse existing configuration
	if configMap.Data == nil || configMap.Data["config.json"] == "" {
		// No configuration to remove
		return nil
	}

	var configData map[string]interface{}
	err = json.Unmarshal([]byte(configMap.Data["config.json"]), &configData)
	if err != nil {
		return fmt.Errorf("failed to parse existing config: %w", err)
	}

	// Navigate to tenant's clusters and remove the specific cluster
	if configData["tenants"] != nil {
		tenants := configData["tenants"].(map[string]interface{})
		if tenants[tenantID] != nil {
			tenantClusters := tenants[tenantID].(map[string]interface{})

			// Remove the cluster configuration
			delete(tenantClusters, task.Spec.ClusterID)

			// If tenant has no more clusters, remove the tenant
			if len(tenantClusters) == 0 {
				delete(tenants, tenantID)
				log.Info("Removed tenant from config as it has no more clusters", "tenantID", tenantID)
			}
		}
	}

	// Marshal the updated configuration
	configJSON, err := json.MarshalIndent(configData, "", "  ")
	if err != nil {
		return fmt.Errorf("failed to marshal updated config: %w", err)
	}

	// Update the ConfigMap
	configMap.Data["config.json"] = string(configJSON)
	err = it.cli.Update(ctx, configMap)
	if err != nil {
		return fmt.Errorf("failed to update ConfigMap: %w", err)
	}

	log.Info("Removed cluster from ConfigMap",
		"tenantID", tenantID,
		"clusterID", task.Spec.ClusterID)

	return nil
}

// removeClusterFromIAMRole removes the principal from AssumeRole policy when cluster is deleted
func (it *IntegrationTask) removeClusterFromIAMRole(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) error {
	log := log.FromContext(ctx)

	roleName := task.Labels[constants.LabelAlias]
	key := client.ObjectKey{Name: roleName, Namespace: namespace}

	iamRole := &o11yv1.IAM{}
	err := it.cli.Get(ctx, key, iamRole)
	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			// IAM role doesn't exist, nothing to clean up
			return nil
		}
		return fmt.Errorf("failed to get IAM role: %w", err)
	}

	// Get the principal to remove
	principalToRemove := task.Spec.AccessRoleARN
	if principalToRemove == "" {
		log.V(1).Info("No AccessRoleARN specified, skipping principal removal")
		return nil
	}

	// Check if other IntegrationTasks in the namespace are using the same principal
	shouldRemove, err := it.shouldRemovePrincipal(ctx, namespace, principalToRemove, task.Name)
	if err != nil {
		return fmt.Errorf("failed to check if principal should be removed: %w", err)
	}

	if !shouldRemove {
		log.Info("Principal is still in use by other IntegrationTasks, skipping removal",
			"principal", principalToRemove,
			"namespace", namespace)
		return nil
	}

	// Remove the principal from AssumeRole policy
	// This will also delete the IAM role if no principals are left
	err = it.removePrincipalFromAssumeRolePolicy(ctx, iamRole, principalToRemove)
	if err != nil {
		return fmt.Errorf("failed to remove principal from AssumeRole policy: %w", err)
	}

	return nil
}

// shouldRemovePrincipal checks if the principal should be removed by verifying no other IntegrationTasks are using it
func (it *IntegrationTask) shouldRemovePrincipal(ctx context.Context, namespace, principal, currentTaskName string) (bool, error) {
	log := log.FromContext(ctx)

	// List all IntegrationTasks in the namespace
	taskList := &o11yv1.IntegrationTaskList{}
	err := it.cli.List(ctx, taskList, client.InNamespace(namespace))
	if err != nil {
		return false, fmt.Errorf("failed to list IntegrationTasks: %w", err)
	}

	// Check if any other task (excluding the current one being deleted) uses the same principal
	for _, task := range taskList.Items {
		// Skip the current task being deleted
		if task.Name == currentTaskName {
			continue
		}

		// Check if this task uses the same principal
		if task.Spec.AccessRoleARN == principal {
			log.V(1).Info("Found another IntegrationTask using the same principal",
				"taskName", task.Name,
				"principal", principal)
			return false, nil
		}
	}

	log.Info("No other IntegrationTasks using this principal, safe to remove",
		"principal", principal)
	return true, nil
}

// removePrincipalFromAssumeRolePolicy removes a principal from the AssumeRole policy
// If no principals remain after removal, it deletes the IAM role
func (it *IntegrationTask) removePrincipalFromAssumeRolePolicy(ctx context.Context, iamRole *o11yv1.IAM, principalToRemove string) error {
	log := log.FromContext(ctx)

	var assumeRolePolicyDocument string

	// Get current AssumeRole policy based on cloud vendor
	switch it.serviceInfo.Vendor {
	case "aws":
		assumeRolePolicyDocument = iamRole.Spec.Aws.AssumeRolePolicyDocument
	case "alicloud":
		assumeRolePolicyDocument = iamRole.Spec.AliCloud.AssumeRolePolicyDocument
	default:
		return fmt.Errorf("unsupported cloud vendor: %s", it.serviceInfo.Vendor)
	}

	if assumeRolePolicyDocument == "" {
		return fmt.Errorf("AssumeRolePolicyDocument is empty")
	}

	// Decode and parse existing policy
	decodedPolicy, err := base64.StdEncoding.DecodeString(assumeRolePolicyDocument)
	if err != nil {
		return fmt.Errorf("failed to decode AssumeRolePolicyDocument: %w", err)
	}

	var policyData iam.AssumeRolePolicy
	err = json.Unmarshal(decodedPolicy, &policyData)
	if err != nil {
		return fmt.Errorf("failed to parse AssumeRolePolicyDocument: %w", err)
	}

	// Remove principal from the policy and check if any principals remain
	removed, hasPrincipalsLeft, err := it.removePrincipalFromPolicy(&policyData, principalToRemove)
	if err != nil {
		return fmt.Errorf("failed to remove principal from policy: %w", err)
	}

	if !removed {
		log.V(1).Info("Principal not found in AssumeRole policy", "principal", principalToRemove)
		return nil
	}

	// If no principals left, delete the IAM role
	if !hasPrincipalsLeft {
		log.Info("No more principals in IAM role, deleting the role",
			"roleName", iamRole.Name,
			"namespace", iamRole.Namespace)
		err = it.cli.Delete(ctx, iamRole)
		if err != nil {
			return fmt.Errorf("failed to delete IAM role: %w", err)
		}
		log.Info("Successfully deleted IAM role", "roleName", iamRole.Name)
		return nil
	}

	// Marshal updated policy
	updatedPolicyJSON, err := json.Marshal(&policyData)
	if err != nil {
		return fmt.Errorf("failed to marshal updated policy: %w", err)
	}

	// Update the IAM role with new policy
	switch it.serviceInfo.Vendor {
	case "aws":
		iamRole.Spec.Aws.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(updatedPolicyJSON)
	case "alicloud":
		iamRole.Spec.AliCloud.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(updatedPolicyJSON)
	}

	err = it.cli.Update(ctx, iamRole)
	if err != nil {
		return fmt.Errorf("failed to update IAM role: %w", err)
	}

	log.Info("Successfully removed principal from AssumeRole policy", "principal", principalToRemove)
	return nil
}

// removePrincipalFromPolicy removes a principal from the policy document
// Returns: (removed bool, hasPrincipalsLeft bool, error)
func (it *IntegrationTask) removePrincipalFromPolicy(policyData *iam.AssumeRolePolicy, principalToRemove string) (bool, bool, error) {
	var principalKey string
	switch it.serviceInfo.Vendor {
	case "aws":
		principalKey = "AWS"
	case "alicloud":
		principalKey = "RAM"
	default:
		return false, false, fmt.Errorf("unsupported cloud vendor: %s", it.serviceInfo.Vendor)
	}

	// Find the statement with the principal we want to update
	for i := range policyData.Statement {
		statement := &policyData.Statement[i]

		// Check if this statement has the principal type we're looking for
		if principalList, exists := statement.Principal[principalKey]; exists {
			// Convert to string slice
			var principals []string
			switch p := principalList.(type) {
			case string:
				principals = []string{p}
			case []interface{}:
				for _, item := range p {
					if str, ok := item.(string); ok {
						principals = append(principals, str)
					}
				}
			case []string:
				principals = p
			default:
				continue
			}

			// Remove the principal
			updatedPrincipals := make([]string, 0)
			removed := false
			for _, existing := range principals {
				if !strings.EqualFold(existing, principalToRemove) {
					updatedPrincipals = append(updatedPrincipals, existing)
				} else {
					removed = true
				}
			}

			if removed {
				// Update the principal list
				if len(updatedPrincipals) == 0 {
					// If no principals left in this statement, remove the entire principal key
					delete(statement.Principal, principalKey)

					// Check if there are any principals left in any statement
					hasPrincipalsLeft := it.checkRemainingPrincipals(policyData, principalKey)
					return true, hasPrincipalsLeft, nil
				} else {
					// Multiple principals remain, store as array
					statement.Principal[principalKey] = updatedPrincipals
					return true, true, nil
				}
			}
		}
	}

	return false, true, nil // Principal not found, assume principals exist
}

// checkRemainingPrincipals checks if any principals remain in the policy
func (it *IntegrationTask) checkRemainingPrincipals(policyData *iam.AssumeRolePolicy, principalKey string) bool {
	for _, statement := range policyData.Statement {
		if principalList, exists := statement.Principal[principalKey]; exists {
			// Check if the principal list is not empty
			switch p := principalList.(type) {
			case string:
				if p != "" {
					return true
				}
			case []interface{}:
				if len(p) > 0 {
					return true
				}
			case []string:
				if len(p) > 0 {
					return true
				}
			}
		}
	}
	return false
}

// extractTenantInfo extracts tenant ID and biz type from the task namespace or labels
func (it *IntegrationTask) extractTenantInfo(task *o11yv1.IntegrationTask) (tenantID, bizType string) {
	if task.Labels != nil {
		if tid, ok := task.Labels[constants.LabelTenantID]; ok {
			tenantID = tid
		}
		if bt, ok := task.Labels[constants.LabelBizType]; ok {
			bizType = bt
		}
	}

	return tenantID, bizType
}

// buildS3Path builds the storage path for slow query logs based on cloud vendor
func (it *IntegrationTask) buildS3Path(task *o11yv1.IntegrationTask) string {
	dataPath := fmt.Sprintf(clusterDataPathTmpl, task.Spec.DestinationBucketName, task.Labels[constants.LabelTenantID], task.Labels[constants.LabelProjectID], task.Spec.ClusterID)
	switch it.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)
	}
}

// createOrUpdateIAMRole creates or updates the IAM role for slow query ETL
func (it *IntegrationTask) createOrUpdateIAMRole(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID, bizType string) (*o11yv1.IAM, error) {
	log := log.FromContext(ctx)

	// IAM role is shared per tenant, not per cluster
	roleName := task.Labels[constants.LabelAlias]

	// Check if IAM role already exists
	key := client.ObjectKey{Name: roleName, Namespace: namespace}
	existingRole := &o11yv1.IAM{}
	getErr := it.cli.Get(ctx, key, existingRole)

	if getErr != nil && client.IgnoreNotFound(getErr) != nil {
		return nil, fmt.Errorf("failed to check existing IAM role: %w", getErr)
	}

	if client.IgnoreNotFound(getErr) == nil {
		// Role doesn't exist, create it with tenant-wide access
		log.Info("Creating new tenant-wide IAM role for slow query ETL", "roleName", roleName, "namespace", namespace, "vendor", it.serviceInfo.Vendor)
		iamRole, err := it.createTenantWideIAMRole(ctx, task, namespace, tenantID, bizType)
		if err != nil {
			return nil, fmt.Errorf("failed to create tenant-wide IAM role: %w", err)
		}
		return iamRole, nil
	} else {
		// Role exists, update AssumeRole policy to add new principal
		log.Info("Updating existing IAM role AssumeRole policy", "roleName", roleName, "namespace", namespace, "vendor", it.serviceInfo.Vendor)
		err := it.updateAssumeRolePolicy(ctx, task, existingRole)
		if err != nil {
			return nil, fmt.Errorf("failed to update AssumeRole policy: %w", err)
		}
		return existingRole, nil
	}
}

// createTenantWideIAMRole creates a new IAM role with tenant-wide access
func (it *IntegrationTask) createTenantWideIAMRole(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID, bizType string) (*o11yv1.IAM, error) {
	roleName := task.Labels[constants.LabelAlias]

	// Create assume role policy with the initial principal
	assumeRolePolicy := it.createInitialAssumeRolePolicy(task.Spec.AccessRoleARN, namespace)
	policyJSON, err := json.Marshal(assumeRolePolicy)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal assume role policy: %w", err)
	}

	// Create tenant-wide access policy (covers all clusters in the tenant)
	accessPolicy := it.createTenantWideAccessPolicy(tenantID, bizType)
	accessPolicyJSON, err := json.Marshal(accessPolicy)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal access policy: %w", err)
	}

	var iamRole *o11yv1.IAM
	policyName := fmt.Sprintf(iamRolePolicyNameTmpl, it.serviceInfo.LogicalRegionName, tenantID)

	// Create IAM role based on cloud vendor
	switch it.serviceInfo.Vendor {
	case "aws":
		iamRole = &o11yv1.IAM{
			ObjectMeta: metav1.ObjectMeta{
				Name:      roleName,
				Namespace: namespace,
				Labels: map[string]string{
					constants.LabelAlias:   roleName,
					constants.LabelBizType: bizType,
				},
			},
			Spec: o11yv1.IAMSpec{
				Aws: o11yv1.AWSIAM{
					AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJSON),
					Policies: []o11yv1.AWSRolePolicy{
						{
							PolicyName:     policyName,
							PolicyDocument: base64.StdEncoding.EncodeToString(accessPolicyJSON),
						},
					},
				},
			},
		}
	case "alicloud":
		iamRole = &o11yv1.IAM{
			ObjectMeta: metav1.ObjectMeta{
				Name:      roleName,
				Namespace: namespace,
				Labels: map[string]string{
					constants.LabelAlias:   roleName,
					constants.LabelBizType: bizType,
				},
			},
			Spec: o11yv1.IAMSpec{
				AliCloud: o11yv1.AliCloudIAM{
					AssumeRolePolicyDocument: base64.StdEncoding.EncodeToString(policyJSON),
					Policies: []o11yv1.AliCloudPolicy{
						{
							PolicyName:     policyName,
							PolicyDocument: base64.StdEncoding.EncodeToString(accessPolicyJSON),
						},
					},
				},
			},
		}
	default:
		return nil, fmt.Errorf("unsupported cloud vendor: %s", it.serviceInfo.Vendor)
	}

	if _, err = controllerutil.CreateOrUpdate(ctx, it.cli, iamRole, func() error {
		return nil
	}); err != nil {
		return nil, fmt.Errorf("failed to create IAM role: %w", err)
	}

	return iamRole, nil
}

// updateAssumeRolePolicy updates the AssumeRole policy to add new principal
func (it *IntegrationTask) updateAssumeRolePolicy(ctx context.Context, task *o11yv1.IntegrationTask, existingRole *o11yv1.IAM) error {
	log := log.FromContext(ctx)

	newPrincipal := task.Spec.AccessRoleARN
	if newPrincipal == "" {
		log.V(1).Info("No AccessRoleARN specified, skipping AssumeRole policy update")
		return nil
	}

	var assumeRolePolicyDocument string
	var err error

	// Get current AssumeRole policy based on cloud vendor
	switch it.serviceInfo.Vendor {
	case "aws":
		assumeRolePolicyDocument = existingRole.Spec.Aws.AssumeRolePolicyDocument
	case "alicloud":
		assumeRolePolicyDocument = existingRole.Spec.AliCloud.AssumeRolePolicyDocument
	default:
		return fmt.Errorf("unsupported cloud vendor: %s", it.serviceInfo.Vendor)
	}

	if assumeRolePolicyDocument == "" {
		return fmt.Errorf("AssumeRolePolicyDocument is empty")
	}

	// Decode and parse existing policy
	decodedPolicy, err := base64.StdEncoding.DecodeString(assumeRolePolicyDocument)
	if err != nil {
		return fmt.Errorf("failed to decode AssumeRolePolicyDocument: %w", err)
	}

	var policyData iam.AssumeRolePolicy
	err = json.Unmarshal(decodedPolicy, &policyData)
	if err != nil {
		return fmt.Errorf("failed to parse AssumeRolePolicyDocument: %w", err)
	}

	// Add new principal to the policy
	updated, err := it.addPrincipalToPolicy(&policyData, newPrincipal)
	if err != nil {
		return fmt.Errorf("failed to add principal to policy: %w", err)
	}

	if !updated {
		log.V(1).Info("Principal already exists in AssumeRole policy", "principal", newPrincipal)
		return nil
	}

	// Marshal updated policy
	updatedPolicyJSON, err := json.Marshal(&policyData)
	if err != nil {
		return fmt.Errorf("failed to marshal updated policy: %w", err)
	}

	// Update the IAM role with new policy
	switch it.serviceInfo.Vendor {
	case "aws":
		existingRole.Spec.Aws.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(updatedPolicyJSON)
	case "alicloud":
		existingRole.Spec.AliCloud.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(updatedPolicyJSON)
	}

	err = it.cli.Update(ctx, existingRole)
	if err != nil {
		return fmt.Errorf("failed to update IAM role: %w", err)
	}

	log.Info("Successfully added principal to AssumeRole policy", "principal", newPrincipal)
	return nil
}

// createInitialAssumeRolePolicy creates the initial assume role policy with the first principal
func (it *IntegrationTask) createInitialAssumeRolePolicy(accessRoleARN, namespace string) iam.AssumeRolePolicy {
	switch it.serviceInfo.Vendor {
	case "aws":
		return iam.AssumeRolePolicy{
			Version: "2012-10-17",
			Statement: []iam.Statement{
				{
					Effect: "Allow",
					Principal: map[string]any{
						"AWS": []string{accessRoleARN},
					},
					Action: "sts:AssumeRole",
				},
			},
		}
	case "alicloud":
		return iam.AssumeRolePolicy{
			Version: "1",
			Statement: []iam.Statement{
				{
					Effect: "Allow",
					Principal: map[string]any{
						"RAM": []string{accessRoleARN},
					},
					Action: "sts:AssumeRole",
				},
			},
		}
	default:
		return iam.AssumeRolePolicy{}
	}
}

// createTenantWideAccessPolicy creates access policy that covers all clusters in the tenant
func (it *IntegrationTask) createTenantWideAccessPolicy(tenantID, bizType string) map[string]interface{} {
	bucketName := it.serviceInfo.BizTypeConfigs[bizType].ExternalBucket
	objPrefixPath := fmt.Sprintf(tenantDataPrefixTmpl, tenantID)
	switch it.serviceInfo.Vendor {
	case "aws":
		// Grant access to all slow_logs under the tenant path
		return map[string]interface{}{
			"Version": "2012-10-17",
			"Statement": []map[string]interface{}{
				{
					"Effect": "Allow",
					"Action": []string{
						"s3:GetObject",
						"s3:ListBucket",
					},
					"Resource": []string{
						fmt.Sprintf("arn:aws:s3:::%s/%s/*", bucketName, objPrefixPath),
					},
				},
			},
		}
	case "alicloud":
		return map[string]interface{}{
			"Version": "1",
			"Statement": []map[string]interface{}{
				{
					"Effect": "Allow",
					"Action": []string{
						"oss:GetObject",
						"oss:GetObjectAcl",
					},
					"Resource": []string{
						fmt.Sprintf("acs:oss:*:*:%s/%s/*", bucketName, objPrefixPath),
					},
				},
				{
					"Effect": "Allow",
					"Action": []string{
						"oss:ListObjects",
					},
					"Resource": []string{
						fmt.Sprintf("acs:oss:*:*:%s", bucketName),
					},
					"Condition": map[string]interface{}{
						"StringLike": map[string]interface{}{
							"oss:Prefix": []string{
								objPrefixPath,
								fmt.Sprintf("%s/*", objPrefixPath),
							},
						},
					},
				},
			},
		}
	default:
		return nil
	}
}

// addPrincipalToPolicy adds a new principal to the AssumeRole policy
func (it *IntegrationTask) addPrincipalToPolicy(policyData *iam.AssumeRolePolicy, newPrincipal string) (bool, error) {

	var principalKey string
	switch it.serviceInfo.Vendor {
	case "aws":
		principalKey = "AWS"
	case "alicloud":
		principalKey = "RAM"
	default:
		return false, fmt.Errorf("unsupported cloud vendor: %s", it.serviceInfo.Vendor)
	}

	// Find the statement with the principal we want to update
	for i := range policyData.Statement {
		statement := &policyData.Statement[i]

		// Check if this statement has the principal type we're looking for
		if principalList, exists := statement.Principal[principalKey]; exists {
			// Convert to string slice
			var principals []string
			switch p := principalList.(type) {
			case string:
				principals = []string{p}
			case []interface{}:
				for _, item := range p {
					if str, ok := item.(string); ok {
						principals = append(principals, str)
					}
				}
			case []string:
				principals = p
			default:
				continue
			}

			// Check if principal already exists
			for _, existing := range principals {
				if strings.EqualFold(existing, newPrincipal) {
					return false, nil // Principal already exists
				}
			}

			// Add new principal
			principals = append(principals, newPrincipal)
			statement.Principal[principalKey] = principals
			return true, nil
		}
	}

	return false, fmt.Errorf("no suitable statement found to add principal")
}

// createOrUpdateJobManagerConfig creates or updates the ConfigMap for job manager
func (it *IntegrationTask) createOrUpdateJobManagerConfig(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) error {
	log := log.FromContext(ctx)

	// Use a shared ConfigMap for all tenants in the namespace
	configMapName := fmt.Sprintf(configMapNameTmpl, tenantID)

	// Check if ConfigMap already exists
	key := client.ObjectKey{Name: configMapName, Namespace: namespace}
	existingConfigMap := &corev1.ConfigMap{}
	getErr := it.cli.Get(ctx, key, existingConfigMap)
	if getErr != nil && client.IgnoreNotFound(getErr) != nil {
		return fmt.Errorf("failed to get ConfigMap %s: %w", configMapName, getErr)
	}

	// Create cluster configuration for this specific cluster
	clusterConfig := map[string]interface{}{
		"frequency":  task.Spec.Frequency,
		"taskType":   task.Spec.TaskType,
		"dataFormat": task.Spec.DataFormat,
		"projectId":  task.Labels[constants.LabelProjectID],
	}

	// Build the hierarchical configuration structure
	var configData map[string]interface{}

	if getErr != nil || existingConfigMap.Data == nil || existingConfigMap.Data[configFileName] == "" {
		// ConfigMap doesn't exist, create it with initial structure
		log.Info("Creating new ConfigMap for slow query ETL", "configMapName", configMapName, "namespace", namespace)

		configData = map[string]interface{}{
			"tenants": map[string]interface{}{
				tenantID: map[string]interface{}{
					task.Spec.ClusterID: clusterConfig,
				},
			},
		}
	} else {
		// ConfigMap exists, parse existing data and update
		log.Info("Updating existing ConfigMap for slow query ETL", "configMapName", configMapName, "namespace", namespace)

		err := json.Unmarshal([]byte(existingConfigMap.Data[configFileName]), &configData)
		if err != nil {
			return fmt.Errorf("failed to parse existing config: %w", err)
		}

		// Ensure tenants section exists
		if configData["tenants"] == nil {
			configData["tenants"] = make(map[string]interface{})
		}

		tenants := configData["tenants"].(map[string]interface{})

		// Ensure tenant section exists
		if tenants[tenantID] == nil {
			tenants[tenantID] = make(map[string]interface{})
		}

		tenantClusters := tenants[tenantID].(map[string]interface{})

		// Add or update cluster configuration
		tenantClusters[task.Spec.ClusterID] = clusterConfig
	}

	// Marshal the complete configuration
	configJSON, err := json.MarshalIndent(configData, "", "  ")
	if err != nil {
		return fmt.Errorf("failed to marshal config data: %w", err)
	}

	configMap := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      configMapName,
			Namespace: namespace,
		},
	}

	_, err = controllerutil.CreateOrUpdate(ctx, it.cli, configMap, func() error {
		// Set labels
		if configMap.Labels == nil {
			configMap.Labels = make(map[string]string)
		}
		configMap.Labels["app"] = "slow-query-etl"
		configMap.Labels["version"] = "v1.0.0"

		// Set data
		if configMap.Data == nil {
			configMap.Data = make(map[string]string)
		}
		configMap.Data[configFileName] = string(configJSON)

		return nil
	})
	if err != nil {
		return fmt.Errorf("failed to create or update ConfigMap: %w", err)
	}

	log.Info("ConfigMap updated successfully",
		"configMapName", configMapName,
		"tenantID", tenantID,
		"clusterID", task.Spec.ClusterID)

	return nil
}

// createOrUpdateETLJobManager creates or updates the ETL job manager deployment
func (it *IntegrationTask) createOrUpdateETLJobManager(ctx context.Context, task *o11yv1.IntegrationTask, namespace string) error {
	log := log.FromContext(ctx)
	// Extract tenant info for naming
	tenantID, _ := it.extractTenantInfo(task)
	deploymentName := fmt.Sprintf(deployNameTmpl, tenantID)

	saName := fmt.Sprintf(serviceAccountNameTmpl, tenantID)

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

	_, err := controllerutil.CreateOrUpdate(ctx, it.cli, serviceAccount, func() error {
		// Set annotations for pod identity
		if serviceAccount.Annotations == nil {
			serviceAccount.Annotations = make(map[string]string)
		}

		// Add AliCloud pod identity annotation
		if it.serviceInfo.Vendor == "alicloud" {
			serviceAccount.Annotations["pod-identity.alibabacloud.com/role-name"] = it.serviceInfo.ControllerConfig.IntegrationTaskConfig.JobManagerIAMRole
		}
		return nil
	})
	if err != nil {
		return fmt.Errorf("failed to create or update ServiceAccount: %w", err)
	}

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

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

	err = it.cli.Get(ctx, key, existingDeployment)
	if err != nil {
		if client.IgnoreNotFound(err) == nil {
			// Deployment doesn't exist, create it
			log.Info("Creating new Deployment for slow query ETL job manager", "deploymentName", deploymentName, "namespace", namespace, "vendor", it.serviceInfo.Vendor)
			// srcPrefix is a path within the source bucket — do not include the bucket name.
			var dataPrefix = fmt.Sprintf(serverlessDataPrefixTmpl, task.Labels[constants.LabelBizType])

			containerArgs := []string{
				fmt.Sprintf("-srcBucket=%s", task.Spec.SourceBucketName),
				fmt.Sprintf("-srcPrefix=%s", dataPrefix),
				fmt.Sprintf("-dstBucket=%s", task.Spec.DestinationBucketName),
				fmt.Sprintf("-vendor=%s", it.serviceInfo.Vendor),
				fmt.Sprintf("-region=%s", it.serviceInfo.Region),
				fmt.Sprintf("-configPath=%s", "/etc/config/"+configFileName),
			}
			if task.Spec.DataPrefix != "" {
				containerArgs = append(containerArgs, fmt.Sprintf("-shareDataPrefix=%s", task.Spec.DataPrefix))
			}

			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, tenantID),
											},
										},
									},
								},
							},
						},
					},
				},
			}

			err = it.cli.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
		log.Info("Deployment already exists for slow query ETL job manager", "deploymentName", deploymentName, "namespace", namespace, "vendor", it.serviceInfo.Vendor)

		// Update the deployment if needed (e.g., if image or config changed)
		needsUpdate := false

		// 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 needsUpdate {
			err = it.cli.Update(ctx, existingDeployment)
			if err != nil {
				return fmt.Errorf("failed to update Deployment: %w", err)
			}
			log.Info("Updated Deployment for slow query ETL job manager", "deploymentName", deploymentName)
		}
	}

	return nil
}

// cleanupSharedResourcesIfNeeded cleans up shared resources if this is the last cluster for the tenant
func (it *IntegrationTask) cleanupSharedResourcesIfNeeded(ctx context.Context, task *o11yv1.IntegrationTask, namespace, tenantID string) error {
	log := log.FromContext(ctx)

	// Check if there are other integration tasks for this tenant
	taskList := &o11yv1.IntegrationTaskList{}
	err := it.cli.List(ctx, taskList, client.InNamespace(namespace))
	if err != nil {
		return fmt.Errorf("failed to list integration tasks: %w", err)
	}

	// Count active tasks (excluding the one being deleted)
	activeTasks := 0
	for _, t := range taskList.Items {
		if t.Name != task.Name && t.Status.Phase != constants.DELETING {
			activeTasks++
		}
	}

	// If this is the last task, clean up shared resources
	if activeTasks == 0 {
		log.Info("Cleaning up shared resources for tenant", "tenant", tenantID, "namespace", namespace)

		// Delete the shared Deployment
		deploymentName := fmt.Sprintf(deployNameTmpl, tenantID)
		deployment := &appsv1.Deployment{}
		err = it.cli.Get(ctx, client.ObjectKey{Name: deploymentName, Namespace: namespace}, deployment)
		if err == nil {
			err = it.cli.Delete(ctx, deployment)
			if err != nil {
				log.Error(err, "Failed to delete shared Deployment", "deploymentName", deploymentName)
				return err
			}
			return fmt.Errorf("Requeue as integration task not deleted")
		}

		// Check if the shared ConfigMap has any tenants left
		configMapName := "slow-query-etl-config"
		configMap := &corev1.ConfigMap{}
		err = it.cli.Get(ctx, client.ObjectKey{Name: configMapName, Namespace: namespace}, configMap)
		if err == nil {
			// Check if ConfigMap has any tenants with clusters left
			hasActiveTenants := false
			if configMap.Data != nil && configMap.Data["config.json"] != "" {
				var configData map[string]interface{}
				parseErr := json.Unmarshal([]byte(configMap.Data["config.json"]), &configData)
				if parseErr == nil && configData["tenants"] != nil {
					tenants := configData["tenants"].(map[string]interface{})
					for _, tenantData := range tenants {
						if tenantClusters, ok := tenantData.(map[string]interface{}); ok && len(tenantClusters) > 0 {
							hasActiveTenants = true
							break
						}
					}
				}
			}

			if !hasActiveTenants {
				err = it.cli.Delete(ctx, configMap)
				if err != nil {
					log.Error(err, "Failed to delete shared ConfigMap", "configMapName", configMapName)
					return err
				}
			}
		}

		// Delete the shared IAM role if it has no policies left
		roleName := task.Labels[constants.LabelAlias]
		iamRole := &o11yv1.IAM{}
		err = it.cli.Get(ctx, client.ObjectKey{Name: roleName, Namespace: namespace}, iamRole)
		if err == nil {
			hasPolicies := false
			switch it.serviceInfo.Vendor {
			case "aws":
				hasPolicies = len(iamRole.Spec.Aws.Policies) > 0
			case "alicloud":
				hasPolicies = len(iamRole.Spec.AliCloud.Policies) > 0
			}

			if !hasPolicies {
				err = it.cli.Delete(ctx, iamRole)
				if err != nil {
					log.Error(err, "Failed to delete shared IAM role", "roleName", roleName)
					return err
				}
			}
		}
	}

	return nil
}

// verifyIAMRolePolicy checks if the IAM role has the correct tenant-wide policy and AssumeRole policy
func (it *IntegrationTask) verifyIAMRolePolicy(ctx context.Context, iamRole *o11yv1.IAM, task *o11yv1.IntegrationTask, tenantID string) (bool, error) {
	found := false
	expectedPolicyName := fmt.Sprintf(iamRolePolicyNameTmpl, it.serviceInfo.LogicalRegionName, tenantID)

	switch it.serviceInfo.Vendor {
	case "aws":
		for _, policy := range iamRole.Spec.Aws.Policies {
			if policy.PolicyName == expectedPolicyName {
				found = true
				break
			}
		}
	case "alicloud":
		for _, policy := range iamRole.Spec.AliCloud.Policies {
			if policy.PolicyName == expectedPolicyName {
				found = true
				break
			}
		}
	default:
		return false, fmt.Errorf("unsupported cloud vendor for policy verification: %s", it.serviceInfo.Vendor)
	}
	if !found {
		return false, nil
	}

	// Verify that the AssumeRolePolicyDocument contains the expected AccessRoleARN
	return it.verifyAssumeRolePolicyDocument(ctx, iamRole, task)
}

// verifyAssumeRolePolicyDocument checks if the AssumeRolePolicyDocument contains the expected AccessRoleARN
func (it *IntegrationTask) verifyAssumeRolePolicyDocument(ctx context.Context, iamRole *o11yv1.IAM, task *o11yv1.IntegrationTask) (bool, error) {
	log := log.FromContext(ctx)

	var assumeRolePolicyDocument string

	// Get the AssumeRolePolicyDocument based on cloud vendor
	switch it.serviceInfo.Vendor {
	case "aws":
		assumeRolePolicyDocument = iamRole.Spec.Aws.AssumeRolePolicyDocument
	case "alicloud":
		assumeRolePolicyDocument = iamRole.Spec.AliCloud.AssumeRolePolicyDocument
	default:
		return false, fmt.Errorf("unsupported cloud vendor for assume role policy verification: %s", it.serviceInfo.Vendor)
	}

	if assumeRolePolicyDocument == "" {
		log.V(1).Info("AssumeRolePolicyDocument is empty")
		return false, nil
	}

	// Decode the base64 encoded policy document
	decodedPolicy, err := base64.StdEncoding.DecodeString(assumeRolePolicyDocument)
	if err != nil {
		return false, fmt.Errorf("failed to decode AssumeRolePolicyDocument: %w", err)
	}

	// Parse the policy document
	var policyData map[string]interface{}
	err = json.Unmarshal(decodedPolicy, &policyData)
	if err != nil {
		return false, fmt.Errorf("failed to parse AssumeRolePolicyDocument: %w", err)
	}

	// Check if the policy contains the expected AccessRoleARN
	expectedRoleARN := task.Spec.AccessRoleARN
	if expectedRoleARN == "" {
		log.V(1).Info("No AccessRoleARN specified in task spec")
		return true, nil // If no specific role is required, consider it valid
	}

	// Convert policy to lowercase string for case-insensitive search
	policyStr := strings.ToLower(string(decodedPolicy))
	expectedRoleStr := strings.ToLower(expectedRoleARN)

	// Check if the expected role ARN is present in the policy document
	if !strings.Contains(policyStr, expectedRoleStr) {
		log.V(1).Info("AssumeRolePolicyDocument does not contain expected AccessRoleARN",
			"expectedRoleARN", expectedRoleARN,
			"vendor", it.serviceInfo.Vendor)
		return false, nil
	}

	log.V(1).Info("AssumeRolePolicyDocument does not properly reference expected AccessRoleARN in Principal section",
		"expectedRoleARN", expectedRoleARN,
		"vendor", it.serviceInfo.Vendor)
	return false, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *IntegrationTaskReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		Named("integrationtask-v1").
		For(
			&o11yv1.IntegrationTask{},
			builder.WithPredicates(predicate.NewPredicateFuncs(func(obj client.Object) bool {
				return obj.GetLabels()[constants.LabelInternal] != "true"
			})),
		).
		Complete(r)
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *IntegrationTaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.Log.WithValues("IntegrationTask", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)

	var task o11yv1.IntegrationTask
	// use r.reconciler.Client because IntegrationTaskReconciler does not embed client.Client
	if err := r.reconciler.Client.Get(ctx, req.NamespacedName, &task); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	if task.DeletionTimestamp.IsZero() {
		if _, ok := task.Annotations[constants.AnnotationKeyPauseReconcile]; ok {
			logger.Info("IntegrationTask has pause annotation, skipping reconciliation")
			return ctrl.Result{}, nil
		}
	}

	return r.reconciler.Reconcile(ctx, req, integrationTaskFinalizerName)
}