package regionalservice

import (
	"context"
	"fmt"
	"regexp"
	"sync"
	"time"

	"github.com/pingcap/log"
	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/utils"
	"github.com/tidbcloud/observability/pkg/resourcemanager"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	DataPathTmpl = "%s/log_data/data/%s/%s/%s"
)

var (
	awsAccessPrincipalPattern      = regexp.MustCompile(`^arn:[^:]+:iam::[0-9]{12}:(role/[^:]+|root)$`)
	aliCloudAccessPrincipalPattern = regexp.MustCompile(`^acs:ram::[0-9]+:(role/[^:]+|root)$`)
)

type IntegrationTaskHandler struct {
}

var integrationTaskHandler *IntegrationTaskHandler
var integrationTaskOnce sync.Once

func GetIntegrationTaskHandler() *IntegrationTaskHandler {
	integrationTaskOnce.Do(func() {
		if integrationTaskHandler == nil {
			integrationTaskHandler = new(IntegrationTaskHandler)
		}
	})
	return integrationTaskHandler
}

// CreateIntegrationTask creates a new monitoring pipeline
func (h *IntegrationTaskHandler) CreateIntegrationTask(ctx context.Context, req message.CreateIntegrationTaskReq) (resp message.CreateIntegrationTaskResp, err error) {
	return h.createIntegrationTask(ctx, req, false)
}

// CreateIntegrationTaskInternal creates a new monitoring pipeline for vector agent (internal)
func (h *IntegrationTaskHandler) CreateIntegrationTaskInternal(ctx context.Context, req message.CreateIntegrationTaskReq) (resp message.CreateIntegrationTaskResp, err error) {
	return h.createIntegrationTask(ctx, req, true)
}

// createIntegrationTask is the internal implementation for creating monitoring pipelines
func (h *IntegrationTaskHandler) createIntegrationTask(ctx context.Context, req message.CreateIntegrationTaskReq, isInternal bool) (resp message.CreateIntegrationTaskResp, err error) {
	if len(req.Frequency) != 0 {
		freq, err := time.ParseDuration(req.Frequency)
		if err != nil {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "invalid frequency format")
		}
		if freq < 10*time.Minute || freq > 24*time.Hour {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "invalid frequency %s, should be between 10 minutes and 24 hours", req.Frequency)
		}
	} else {
		req.Frequency = message.DefaultIntegrationTaskFrequency
	}
	if len(req.DataFormat) == 0 {
		req.DataFormat = message.DefaultIntegrationTaskDataFormat
	}
	if isInternal {
		if err := defaultTimeRange(&req); err != nil {
			return resp, err
		}
	} else {
		if req.AccessPrincipal == "" {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "accessPrincipal is required for external tasks")
		}
		if err := validateAccessPrincipalForVendor(configs.GetServiceInfo().Vendor, req.AccessPrincipal); err != nil {
			return resp, err
		}
	}

	ctx = dealuseRDS(ctx, req.BizType)
	clusters, err := resourcemanager.GetClusterReaderWriter(ctx).ListClusters(ctx, client.MatchingLabels{
		constants.LabelApplicationID: req.ClusterID,
		constants.LabelTenantID:      req.TenantID,
	})
	if err != nil {
		return resp, err
	}
	if len(clusters.Items) == 0 {
		return resp, errors.NewErrorf(errors.ParameterInvalid, "o11y for cluster %s not available", req.ClusterID)
	}
	cluster := clusters.Items[0]

	// Check if monitoring pipeline already exists for this cluster and task type
	taskID := fmt.Sprintf("%s-%s", req.DataType, utils.FnvHash(fmt.Sprintf("%s-%s-%s", req.TenantID, req.ClusterID, req.DataType)))
	if isInternal {
		taskID = fmt.Sprintf("%s-%s", taskID, utils.FnvHash(req.UserID))
	}
	// Create IntegrationTask resource
	task := &v1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Name:      taskID,
			Namespace: cluster.Namespace,
		},
	}

	exist, err := resourcemanager.GetIntegrationTaskReaderWriter().IsIntegrationTaskExists(ctx, client.ObjectKeyFromObject(task))
	if err != nil {
		log.Error("failed to query existing pipelines", zap.Error(err))
		return resp, err
	}
	if exist {
		return resp, errors.NewErrorf(errors.ParameterInvalid, "pipeline with type %s already exists", req.DataType)
	}

	projectID := cluster.Labels[constants.LabelProjectID]

	// Determine AccessRoleARN and bucket names based on internal/external task type
	var accessRoleARN string
	var sourceBucket, destBucket string

	if isInternal {
		// Use pre-configured vector role ARN
		vectorRoleARN := configs.GetServiceInfo().ControllerConfig.IntegrationTaskConfig.VectorRoleARN
		if vectorRoleARN == "" {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "VectorRoleARN not configured for internal tasks")
		}
		accessRoleARN = vectorRoleARN
	} else {
		accessRoleARN = req.AccessPrincipal
	}
	sourceBucket, destBucket = dataBucketName(req.BizType, isInternal, req.UserID, req.TenantID, "")

	task.Labels = map[string]string{
		constants.LabelApplicationID:       req.ClusterID,
		constants.LabelIntegrationTaskID:   taskID,
		constants.LabelIntegrationTaskType: req.DataType,
		constants.LabelTenantID:            req.TenantID,
		constants.LabelProjectID:           projectID,
		constants.LabelBizType:             req.BizType,
		constants.LabelO11yID:              cluster.Name,
		constants.LabelAlias:               fmt.Sprintf(constants.IntegrationTaskRoleNameTemplate, req.TenantID),
		constants.LabelInternal:            "false",
	}
	if isInternal {
		task.Labels[constants.LabelInternal] = "true"
		if req.UserID != "" {
			task.Labels[constants.LabelUserID] = req.UserID
		}
	}
	task.Annotations = map[string]string{
		constants.AnnotationIntegrationTaskDisplayName: req.DisplayName,
	}

	task.Spec = v1.IntegrationTaskSpec{
		TaskType:              v1.TaskType(req.DataType),
		ClusterID:             req.ClusterID,
		AccessRoleARN:         accessRoleARN,
		Frequency:             req.Frequency,
		SourceBucketName:      sourceBucket,
		DestinationBucketName: destBucket,
		DataFormat:            req.DataFormat,
		DataPrefix:            req.DataPrefix,
		JobType:               v1.JobType(req.JobType),
		StartTime:             req.StartTime,
		EndTime:               req.EndTime,
		RawLogComponents:      req.RawLogComponents,
	}

	// Create the monitoring pipeline
	err = resourcemanager.GetIntegrationTaskReaderWriter().CreateIntegrationTask(ctx, task)
	if err != nil {
		log.Error("failed to create monitoring pipeline", zap.Error(err))
		return resp, err
	}

	// Build response
	resp = message.CreateIntegrationTaskResp{
		ID:               taskID,
		Name:             fmt.Sprintf("clusters/%s/monitoringPipelines/%s", req.ClusterID, taskID),
		DisplayName:      req.DisplayName,
		DataType:         req.DataType,
		ClusterID:        req.ClusterID,
		AccessPrincipal:  accessRoleARN,
		Frequency:        req.Frequency,
		DataFormat:       req.DataFormat,
		DataPrefix:       req.DataPrefix,
		JobType:          req.JobType,
		StartTime:        req.StartTime,
		EndTime:          req.EndTime,
		RawLogComponents: req.RawLogComponents,
		DestinationInfo:  buildDestinationInfo(task),
		CreateTime:       task.CreationTimestamp.Format(time.RFC3339),
	}

	log.Info("monitoring pipeline created successfully",
		zap.String("task_id", taskID),
		zap.String("cluster_id", req.ClusterID),
		zap.String("task_type", req.DataType))

	return resp, nil
}

func validateAccessPrincipalForVendor(vendor, accessPrincipal string) error {
	switch constants.ProvidedVendor(vendor) {
	case constants.AWS:
		if awsAccessPrincipalPattern.MatchString(accessPrincipal) {
			return nil
		}
		return errors.NewErrorf(errors.ParameterInvalid,
			"accessPrincipal must be a valid aws role or root ARN, expected format: arn:aws:iam::<account-id>:(role/<role-name>|root)")
	case constants.AliCloud:
		if aliCloudAccessPrincipalPattern.MatchString(accessPrincipal) {
			return nil
		}
		return errors.NewErrorf(errors.ParameterInvalid,
			"accessPrincipal must be a valid alicloud role or root ARN, expected format: acs:ram::<account-id>:(role/<role-name>|root)")
	default:
		return errors.NewErrorf(errors.ParameterInvalid,
			"accessPrincipal is not supported for vendor %s", vendor)
	}
}

func defaultTimeRange(req *message.CreateIntegrationTaskReq) error {
	if req.StartTime == "" {
		return errors.NewErrorf(errors.ParameterInvalid, "invalid startTime, startTime is required")
	}
	if req.StartTime != "" {
		_, err := time.Parse(time.RFC3339, req.StartTime)
		if err != nil {
			return errors.NewErrorf(errors.ParameterInvalid, "invalid startTime format, should be RFC3339")
		}
	}

	if req.EndTime == "" {
		now := time.Now().UTC()
		req.EndTime = now.Format(time.RFC3339)
		return nil
	}

	_, err := time.Parse(time.RFC3339, req.EndTime)
	if err != nil {
		return errors.NewErrorf(errors.ParameterInvalid, "invalid endTime format, should be RFC3339")
	}

	return nil
}

// DetailIntegrationTask gets details of an monitoring pipeline
func (h *IntegrationTaskHandler) DetailIntegrationTask(ctx context.Context, req message.DetailIntegrationTaskReq) (resp message.DetailIntegrationTaskResp, err error) {
	if req.ID == "" {
		return resp, errors.NewErrorf(errors.ParameterInvalid, "id is required")
	}

	task, err := resourcemanager.GetIntegrationTaskReaderWriter().GetIntegrationTaskByID(ctx, req.ClusterID, req.ID)
	if err != nil {
		log.Error("failed to get monitoring pipeline",
			zap.Error(err),
			zap.String("task_id", req.ID))
		return resp, err
	}

	resp = convertToDetailResp(task)
	return resp, nil
}

// DeleteIntegrationTask deletes an monitoring pipeline
func (h *IntegrationTaskHandler) DeleteIntegrationTask(ctx context.Context, req message.DeleteIntegrationTaskReq) (resp message.DeleteIntegrationTaskResp, err error) {
	if req.ID == "" {
		return resp, errors.NewErrorf(errors.ParameterInvalid, "id is required")
	}

	task, err := resourcemanager.GetIntegrationTaskReaderWriter().GetIntegrationTaskByID(ctx, req.ClusterID, req.ID)
	if err != nil {
		log.Error("failed to get monitoring pipeline for deletion",
			zap.Error(err),
			zap.String("task_id", req.ID))
		return resp, err
	}

	err = resourcemanager.GetIntegrationTaskReaderWriter().DeleteIntegrationTask(ctx, client.ObjectKeyFromObject(task))
	if err != nil {
		log.Error("failed to delete monitoring pipeline",
			zap.Error(err),
			zap.String("task_id", req.ID))
		return resp, err
	}

	resp = message.DeleteIntegrationTaskResp{
		ID:      req.ID,
		Message: "monitoring pipeline deleted successfully",
	}

	log.Info("monitoring pipeline deleted successfully", zap.String("id", req.ID))
	return resp, nil
}

// QueryIntegrationTasks queries monitoring pipelines with filters
func (h *IntegrationTaskHandler) QueryIntegrationTasks(ctx context.Context, req message.QueryIntegrationTasksReq) (resp message.QueryIntegrationTasksResp, err error) {
	return h.queryIntegrationTasks(ctx, req, false)
}

// QueryIntegrationTasksInternal queries monitoring pipelines for internal services
func (h *IntegrationTaskHandler) QueryIntegrationTasksInternal(ctx context.Context, req message.QueryIntegrationTasksReq) (resp message.QueryIntegrationTasksResp, err error) {
	return h.queryIntegrationTasks(ctx, req, true)
}

// queryIntegrationTasks is the shared implementation for querying monitoring pipelines
func (h *IntegrationTaskHandler) queryIntegrationTasks(ctx context.Context, req message.QueryIntegrationTasksReq, isInternal bool) (resp message.QueryIntegrationTasksResp, err error) {
	// Build filters
	filters := make(map[string]string)
	filters[constants.LabelApplicationID] = req.ClusterID
	filters[constants.LabelTenantID] = req.TenantID
	if req.DataType != nil && *req.DataType != "" {
		filters[constants.LabelIntegrationTaskType] = *req.DataType
	}
	if isInternal {
		filters[constants.LabelInternal] = "true"
		if req.UserID != "" {
			filters[constants.LabelUserID] = req.UserID
		}
	} else {
		filters[constants.LabelInternal] = "false"
	}

	// Set pagination defaults
	pageSize := req.PageSize
	if pageSize <= 0 {
		pageSize = 20
	}

	// Query tasks
	taskList, err := resourcemanager.GetIntegrationTaskReaderWriter().QueryIntegrationTasks(ctx, filters, message.ContinueRequest{
		Limit:         int64(pageSize),
		ContinueToken: req.PageToken,
	})
	if err != nil {
		log.Error("failed to query monitoring pipelines", zap.Error(err))
		return resp, err
	}

	// Convert to response format
	tasks := make([]message.IntegrationTaskSummary, len(taskList.Items))
	for i, task := range taskList.Items {
		tasks[i] = convertToSummary(&task)
	}

	resp = message.QueryIntegrationTasksResp{
		Pipelines: tasks,
		OAPIPagination: message.OAPIPagination{
			NextPageToken: taskList.Continue,
			TotalSize:     len(taskList.Items),
		},
	}

	return resp, nil
}

// UpdateIntegrationTask updates an monitoring pipeline
func (h *IntegrationTaskHandler) UpdateIntegrationTask(ctx context.Context, req message.UpdateIntegrationTaskReq) (resp message.UpdateIntegrationTaskResp, err error) {
	panic("Implement me")
}

func convertToDetailResp(task *v1.IntegrationTask) message.DetailIntegrationTaskResp {
	return message.DetailIntegrationTaskResp{
		ID:               task.Name,
		Name:             fmt.Sprintf("clusters/%s/monitoringPipelines/%s", task.Spec.ClusterID, task.Name),
		DisplayName:      task.Annotations[constants.AnnotationIntegrationTaskDisplayName],
		DataType:         string(task.Spec.TaskType),
		ClusterID:        task.Spec.ClusterID,
		AccessPrincipal:  task.Spec.AccessRoleARN,
		Frequency:        task.Spec.Frequency,
		DataFormat:       task.Spec.DataFormat,
		DataPrefix:       task.Spec.DataPrefix,
		JobType:          string(task.Spec.JobType),
		StartTime:        task.Spec.StartTime,
		EndTime:          task.Spec.EndTime,
		RawLogComponents: task.Spec.RawLogComponents,
		Phase:            string(task.Status.Phase),
		CreateTime:       task.CreationTimestamp.Format(time.RFC3339),
		DestinationInfo:  buildDestinationInfo(task),
	}
}

func convertToSummary(task *v1.IntegrationTask) message.IntegrationTaskSummary {
	return message.IntegrationTaskSummary{
		ID:               task.Name,
		DataType:         string(task.Spec.TaskType),
		AccessPrincipal:  task.Spec.AccessRoleARN,
		Frequency:        task.Spec.Frequency,
		DataFormat:       task.Spec.DataFormat,
		DataPrefix:       task.Spec.DataPrefix,
		JobType:          string(task.Spec.JobType),
		StartTime:        task.Spec.StartTime,
		EndTime:          task.Spec.EndTime,
		RawLogComponents: task.Spec.RawLogComponents,
		DestinationInfo:  buildDestinationInfo(task),
		Phase:            string(task.Status.Phase),
		CreateTime:       task.CreationTimestamp.Format(time.RFC3339),
	}
}

func dataBucketName(bizType string, internal bool, userID, tenantID string, bucketSharePolicy v1.BucketSharePolicy) (string, string) {
	svcInfo := configs.GetServiceInfo()
	src, dst := svcInfo.BizTypeConfigs[bizType].SpecificBucket, svcInfo.BizTypeConfigs[bizType].ExternalBucket
	if len(src) == 0 {
		src = fmt.Sprintf(constants.StorageBucketNameTemplate, svcInfo.Env, constants.DefaultBucketGroup, svcInfo.LogicalRegionName)
	}
	if len(dst) == 0 {
		dst = fmt.Sprintf(constants.ExternalStorageBucketNameTmpl, svcInfo.Env, constants.DefaultBucketGroup, svcInfo.LogicalRegionName)
	}
	if internal && userID != "" {
		dst = fmt.Sprintf("o11y-%s-%s-%s", userID, svcInfo.Env, svcInfo.Region)
	}
	// Exclusive external tasks get a dedicated per-(tenant, logical region)
	// destination bucket so each tenant's exported data is isolated and billed
	// independently. Empty policy resolves from controller config before the
	// exclusive default is applied.
	sharedBucketTenant := svcInfo.ControllerConfig.IntegrationTaskConfig.IsSharedBucketTenant(tenantID)
	if !internal && tenantID != "" && v1.ResolveBucketSharePolicy(bucketSharePolicy, sharedBucketTenant) == v1.BucketSharePolicyExclusive {
		dst = fmt.Sprintf(constants.ExternalStorageBucketNameTmpl, svcInfo.Env, tenantID, svcInfo.LogicalRegionName)
	}
	return src, dst
}

func buildDestinationInfo(task *v1.IntegrationTask) message.DestinationInfo {
	// For internal tasks, use the destination bucket directly and the pre-configured role ARN
	if task.Labels[constants.LabelInternal] == "true" {
		return message.DestinationInfo{
			AccessPoint: fmt.Sprintf("s3://%s", task.Spec.DestinationBucketName),
		}
	}

	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		return message.DestinationInfo{
			AccessPoint:   fmt.Sprintf("s3://%s", buildClusterDataPath(task.Spec.DestinationBucketName, task.Labels[constants.LabelTenantID], task.Labels[constants.LabelProjectID], task.Spec.ClusterID)),
			AssumeRoleARN: fmt.Sprintf(roleArnTemplate, configs.GetAccountID(), task.Labels[constants.LabelAlias]),
		}
	case string(constants.AliCloud):
		return message.DestinationInfo{
			AccessPoint:   fmt.Sprintf("oss://%s", buildClusterDataPath(task.Spec.DestinationBucketName, task.Labels[constants.LabelTenantID], task.Labels[constants.LabelProjectID], task.Spec.ClusterID)),
			AssumeRoleARN: fmt.Sprintf(aliRoleArnTemplate, configs.GetAccountID(), task.Labels[constants.LabelAlias]),
		}
	}
	return message.DestinationInfo{}
}

func buildClusterDataPath(bucketName, tenantID, projectID, clusterID string) string {
	if projectID == "" {
		return fmt.Sprintf("%s/log_data/data/%s/%s", bucketName, tenantID, clusterID)
	}
	return fmt.Sprintf(DataPathTmpl, bucketName, tenantID, projectID, clusterID)
}