package auditlog

import (
	"context"
	"fmt"
	"net/url"
	"path"
	"strconv"
	"strings"

	"github.com/samber/lo"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	tidbgrouprepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/tidbgroup"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/resourcepoolpolicy"
	servicecommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterproxy"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	QueryKernelAuditLogConfigStmt = `SELECT @@GLOBAL.tidb_audit_enabled          AS audit_enabled,
       @@GLOBAL.tidb_audit_log_redacted     AS log_redacted,
       @@GLOBAL.tidb_audit_log_format       AS log_format,
       @@GLOBAL.tidb_audit_log_max_filesize AS max_filesize,
       @@GLOBAL.tidb_audit_log_max_lifetime AS max_lifetime;`
	SetAuditEnabledStmt         = "SET @@GLOBAL.tidb_audit_enabled = '%s'"
	SetAuditLogRedactedStmt     = "SET @@GLOBAL.tidb_audit_log_redacted = '%s'"
	SetAuditLogFormatStmt       = "SET @@GLOBAL.tidb_audit_log_format = '%s'"
	SetAuditLogMaxFilesizeStmt  = "SET @@GLOBAL.tidb_audit_log_max_filesize = %d"
	SetAuditLogMaxLifetimeStmt  = "SET @@GLOBAL.tidb_audit_log_max_lifetime = %d"
	SetAuditLogFilePathStmt     = "SET @@GLOBAL.tidb_audit_log = '/var/log/auditlog/tidb-audit.log'"
	SetAuditLogFilePathJsonStmt = "SET @@GLOBAL.tidb_audit_log = '/var/log/auditlog/tidb-audit.log'"
	TriggerAuditLogRotationStmt = "SELECT audit_log_rotate()"

	// SetAuditLogReservedBackupsStmt is used to set the number of reserved audit log files.
	SetAuditLogReservedBackupsStmt = "SET @@GLOBAL.tidb_audit_log_reserved_backups = %d"
	DefaultAuditLogReservedBackups = 1024

	MaxAuditLogSizeMiB   = 100     // Maximum size of audit log in MiB.
	MinAuditLogSizeMiB   = 1       // Minimum size of audit log in MiB.
	MaxAuditLogAgeSecond = 2592000 // Maximum age of audit log in seconds (30 days).
)

func (svc *service) GetAuditLogSetting(
	ctx context.Context,
	clusterID uint64,
) (*globalngv1.AuditLogSetting, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "GetAuditLogSetting")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, clusterID, false)
	if err != nil {
		return nil, err
	}

	region, err := svc.regionRepo.Get(ctx, cluster.RegionId)
	if err != nil {
		return nil, err
	}

	return svc.getAuditLogSetting(ctx, cluster, region)
}

func (svc *service) getAuditLogSetting(
	ctx context.Context,
	cluster *entity.Cluster,
	region *entity.Region,
) (*globalngv1.AuditLogSetting, bizerr.BizError) {
	clusterID := cluster.ClusterId

	// Get the audit log status from the region server.
	regionalTidbCluster, err := svc.regionalServerCli.GetTidbCluster(ctx, region, &clusterngv1beta.GetTiDBClusterRequest{
		ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
	})
	if err != nil {
		return nil, err
	}

	// Convert the regional config to the central config.
	globalConfig := &globalngv1.AuditLogSetting{
		BucketUrl: regionalTidbCluster.GetAuditLogSetting().GetBucketUrl(),
		Region:    regionalTidbCluster.GetAuditLogSetting().GetRegion(),
		ClusterId: clusterID,
	}

	// Query DB
	dbConfig, err := svc.auditLogSettingRepo.GetByClusterID(ctx, clusterID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err
	}
	// Merge DB config fields into globalConfig if present
	if dbConfig != nil {
		if dbConfig.Id != 0 {
			globalConfig.Id = dbConfig.Id
		}
		if dbConfig.BucketUrl != "" {
			globalConfig.BucketUrl = dbConfig.BucketUrl
		}
		if dbConfig.Region != "" {
			globalConfig.Region = dbConfig.Region
		}
		if dbConfig.CloudRoleIdentity != nil {
			globalConfig.CloudRoleIdentity = dbConfig.CloudRoleIdentity
		}
		if dbConfig.Enabled != nil {
			globalConfig.Enabled = dbConfig.Enabled
		}
		if dbConfig.Unredacted != nil {
			globalConfig.Unredacted = dbConfig.Unredacted
		}
		if dbConfig.Format != "" {
			globalConfig.Format = dbConfig.Format
		}
		if dbConfig.MaxSizeMib != 0 {
			globalConfig.MaxSizeMib = dbConfig.MaxSizeMib
		}
		if dbConfig.MaxAgeSecond != nil {
			globalConfig.MaxAgeSecond = dbConfig.MaxAgeSecond
		}
	}

	// Remove our-added cluster id suffix from the bucket url.
	if strings.HasSuffix(globalConfig.BucketUrl, fmt.Sprint(clusterID)) {
		bucketURL, err := url.Parse(globalConfig.BucketUrl)
		if err != nil {
			return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse bucket url")
		}
		bucketURL.Path = path.Dir(bucketURL.Path)
		globalConfig.BucketUrl = bucketURL.String()
	}

	return globalConfig, nil
}

func (svc *service) UpdateAuditLogSetting(
	ctx context.Context,
	reqConfig *globalngv1.AuditLogSetting,
	validateOnly bool,
) (*globalngv1.AuditLogSetting, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "UpdateAuditLogSetting")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, reqConfig.GetClusterId(), false)
	if err != nil {
		return nil, err
	}
	if berr := servicecommon.CheckClusterResourcePoolLifecycle(
		ctx, svc.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateInstance,
	); berr != nil {
		return nil, berr
	}

	region, err := svc.regionRepo.Get(ctx, cluster.RegionId)
	if err != nil {
		return nil, err
	}

	// Get current config.
	currentConfig, err := svc.getAuditLogSetting(ctx, cluster, region)
	if err != nil {
		return nil, err
	}

	// Merge the request config with the current config.
	if reqConfig.BucketUrl != "" {
		currentConfig.BucketUrl = reqConfig.BucketUrl
	}
	if reqConfig.Region != "" && reqConfig.Region != globalcommonv1.Region_NAME_UNSPECIFIED.String() {
		currentConfig.Region = reqConfig.Region
	}
	if reqConfig.CloudRoleIdentity != nil {
		currentConfig.CloudRoleIdentity = reqConfig.CloudRoleIdentity
	}
	if reqConfig.Unredacted != nil {
		currentConfig.Unredacted = reqConfig.Unredacted
	}
	if reqConfig.Format != "" &&
		reqConfig.Format != openapidedicatedv1beta1.AuditLogConfig_FORMAT_UNSPECIFIED.String() {
		currentConfig.Format = reqConfig.Format
	}
	if reqConfig.MaxSizeMib != 0 {
		currentConfig.MaxSizeMib = reqConfig.MaxSizeMib
	}
	if reqConfig.MaxAgeSecond != nil {
		currentConfig.MaxAgeSecond = reqConfig.MaxAgeSecond
	}
	if reqConfig.Enabled != nil {
		currentConfig.Enabled = reqConfig.Enabled
	}
	reqConfig = currentConfig

	// Set default values for the config if not provided.
	if reqConfig.GetFormat() == "" ||
		reqConfig.GetFormat() == openapidedicatedv1beta1.AuditLogConfig_FORMAT_UNSPECIFIED.String() {
		reqConfig.Format = openapidedicatedv1beta1.AuditLogConfig_TEXT.String()
	}
	if reqConfig.GetMaxSizeMib() == 0 {
		reqConfig.MaxSizeMib = 10
	}
	if reqConfig.MaxAgeSecond == nil {
		reqConfig.MaxAgeSecond = lo.ToPtr(int32(86400)) // Default to 1 day.
	}
	originalReqBucketUrl := reqConfig.GetBucketUrl()
	// Append cluster to path of bucket url.
	bucketURL, err_ := url.Parse(reqConfig.GetBucketUrl())
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeParams, "failed to parse bucket url")
	}
	if !strings.Contains(bucketURL.Path, fmt.Sprint(reqConfig.GetClusterId())) {
		bucketURL.Path = path.Join(bucketURL.Path, fmt.Sprint(reqConfig.GetClusterId()))
		reqConfig.BucketUrl = bucketURL.String()
	}

	// Check the request parameters.
	if reqConfig.GetMaxSizeMib() < MinAuditLogSizeMiB || reqConfig.GetMaxSizeMib() > MaxAuditLogSizeMiB {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"max size must be between %d and %d MiB",
			MinAuditLogSizeMiB,
			MaxAuditLogSizeMiB,
		)
	}
	if reqConfig.GetMaxAgeSecond() > MaxAuditLogAgeSecond {
		return nil, bizerr.New(bizerr.ErrCodeParams, "max age must be less than %d seconds", MaxAuditLogAgeSecond)
	}
	if reqConfig.GetBucketUrl() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "bucket uri is required")
	}
	switch cluster.GetCloudProviderId() {
	case globalcommonv1.CloudProvider_AZURE:
		// Azure supports two auth modes:
		// 1. non-empty cloud_role_identity => SAS token auth
		// 2. empty cloud_role_identity => kubelet identity auth
	case globalcommonv1.CloudProvider_AWS:
		if reqConfig.GetCloudRoleIdentity() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "aws role arn is required")
		}
	case globalcommonv1.CloudProvider_GCP:
		if reqConfig.GetRegion() == "" || reqConfig.GetRegion() == globalcommonv1.Region_NAME_UNSPECIFIED.String() {
			return nil, bizerr.New(bizerr.ErrCodeParams, "region is required")
		}
	case globalcommonv1.CloudProvider_ALICLOUD:
		if reqConfig.GetCloudRoleIdentity() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "alicloud role arn is required")
		}
	}

	// Validate the audit log config.
	if err := ValidateAuditLogSetting(cluster, reqConfig); err != nil {
		return nil, err
	}

	tidbgroups, err := svc.tidbGroupRepo.Query(ctx, &tidbgrouprepo.Filter{
		ClusterIds: []uint64{cluster.ClusterId},
	})
	if err != nil {
		return nil, err
	}
	if len(tidbgroups) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "tidb group not found for cluster %d", cluster.ClusterId)
	}
	tidbgroup := tidbgroups[0]
	proxy, err := svc.GetPlatformAdminClusterProxy(ctx, cluster, region, tidbgroup)
	if err != nil {
		return nil, err
	}
	providerRole, err := svc.regionalServerCli.GetProviderRole(ctx, region, &clusterngv1beta.GetProviderRoleRequest{
		ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
	})
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to get provider role: %v", err)
	}
	sharedPool, err := svc.regionalServerCli.GetSharedPool(ctx, region, &clusterngv1beta.GetSharedPoolRequest{
		SharedPoolId: cluster.GetSharedPoolId(),
	})
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to get shared pool: %v", err)
	}
	reqConfig.ExternalId = providerRole.GetExternalId()
	// Check bucket access.
	pass, err := proxy.AuditLogCheck(ctx, &clusterproxy.AuditCheckModel{
		TargetType: "audit",
		Audit: &clusterproxy.AuditParams{
			BucketURL: reqConfig.GetBucketUrl(),
			Region:    reqConfig.GetRegion(),
			AWS: &clusterproxy.AWSAuth{
				RoleARN:    reqConfig.GetCloudRoleIdentity(),
				ExternalID: reqConfig.GetExternalId(),
			},
			GCP: &clusterproxy.GCPAuth{
				ServiceAccount: reqConfig.GetCloudRoleIdentity(),
			},
			Azure: &clusterproxy.AzureAuth{
				SASToken: reqConfig.GetCloudRoleIdentity(),
				ClientID: sharedPool.GetProvider().GetNodeClientId(),
			},
			AliCloud: &clusterproxy.AliCloudAuth{
				RoleARN:    reqConfig.GetCloudRoleIdentity(),
				ExternalID: reqConfig.GetExternalId(),
			},
		},
	})
	if err != nil {
		return nil, err
	}
	if !pass {
		return nil, BuildPrecheckAuditLogConfigBizError(cluster.GetCloudProviderId())
	}

	if validateOnly {
		// If validate only, return the config without applying it.
		reqConfig.BucketUrl = originalReqBucketUrl // Restore original bucket URL.
		return reqConfig, nil
	}

	// If there is already a running workflow, return an error.
	dagIns, err_ := svc.wf.ListRunningDagInstancesByNamePrefix(
		ctx,
		common.UpdateAuditLogSetting,
		strconv.FormatUint(cluster.ClusterId, 10),
	)
	if err_ != nil {
		return nil, bizerr.WrapError(err_, "failed to list running dag instances")
	}
	if len(dagIns) > 0 {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"the last updating audit log config is still running, please wait for it to finish",
		)
	}

	errx := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		if berr := servicecommon.LockClusterResourcePoolLifecycle(
			ctx, svc.resourcePoolRepo, cluster, resourcepoolpolicy.OperationUpdateInstance,
		); berr != nil {
			return berr
		}
		// Ensure DB record is up-to-date before calling regional or cluster proxy.
		if err := svc.auditLogSettingRepo.Upsert(ctx, reqConfig); err != nil {
			return err
		}

		// Start async reconciliation workflow.
		workFlowErr := svc.wf.RunDagWithBizId(
			ctx,
			common.UpdateAuditLogSetting,
			strconv.FormatUint(cluster.OrganizationId, 10),
			"0",
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:         workflow.GetTraceId(ctx),
				common.KeyClusterId:         strconv.FormatUint(cluster.ClusterId, 10),
				common.KeyAuditLogSettingId: strconv.FormatUint(reqConfig.Id, 10),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if errx != nil {
		return nil, bizerr.WrapError(errx, "update audit log config failed")
	}

	reqConfig.BucketUrl = originalReqBucketUrl // Restore original bucket URL.
	return reqConfig, nil
}

func ValidateAuditLogSetting(cluster *entity.Cluster, config *globalngv1.AuditLogSetting) bizerr.BizError {
	if config == nil {
		return bizerr.New(bizerr.ErrCodeParams, "config is nil")
	}

	// Check cluster status.
	if cluster.GetState() != globalngv1.Cluster_AVAILABLE &&
		cluster.GetState() != globalngv1.Cluster_MODIFYING {
		return bizerr.New(bizerr.ErrCodeParams, "the cluster's status is invalid: %s", cluster.GetState())
	}

	u, err := url.Parse(config.BucketUrl)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse bucket url")
	}

	switch cluster.GetCloudProviderId() {
	case globalcommonv1.CloudProvider_AWS:
		if u.Scheme != "s3" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be s3, but is: %s", u.Scheme)
		}
		if config.GetCloudRoleIdentity() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "role ARN is required")
		}

		region, err := enum.RegionFromString(globalcommonv1.CloudProvider_AWS.String(), config.Region)

		if err != nil || region == globalcommonv1.Region_NAME_UNSPECIFIED {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region: %s", config.GetRegion())
		}
	case globalcommonv1.CloudProvider_GCP:
		if u.Scheme != "gcs" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be gcs, but is: %s", u.Scheme)
		}

		region, err := enum.RegionFromString(globalcommonv1.CloudProvider_GCP.String(), config.Region)
		if err != nil || region == globalcommonv1.Region_NAME_UNSPECIFIED {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region: %s", config.GetRegion())
		}
	case globalcommonv1.CloudProvider_AZURE:
		if u.Scheme != "https" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be https, but is: %s", u.Scheme)
		}
	case globalcommonv1.CloudProvider_ALICLOUD:
		if u.Scheme != "oss" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be oss, but is: %s", u.Scheme)
		}

		region, err := enum.RegionFromString(globalcommonv1.CloudProvider_ALICLOUD.String(), config.Region)
		if err != nil || region == globalcommonv1.Region_NAME_UNSPECIFIED {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region: %s", config.GetRegion())
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported source type: %s", cluster.GetCloudProviderId())
	}

	return nil
}

func CvtToRegionalConfig(
	config *globalngv1.AuditLogSetting,
	cloudProvider globalcommonv1.CloudProvider_Name,
) *clusterngv1beta.Cluster_AuditLogSetting {
	c := &clusterngv1beta.Cluster_AuditLogSetting{
		BucketUrl: config.GetBucketUrl(),
		ProviderIdentity: &clusterngv1beta.Cluster_AuditLogSetting_ProviderIdentity{
			CloudRoleIdentity: config.GetCloudRoleIdentity(),
			ExternalId:        strconv.FormatUint(config.GetClusterId(), 10),
		},
		Region: config.GetRegion(),
	}
	return c
}

func BuildPrecheckAuditLogConfigBizError(cloudProvider globalcommonv1.CloudProvider_Name) bizerr.BizError {
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL and role ARN are correct, then try again.",
		)
	case globalcommonv1.CloudProvider_GCP:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL is correct, then try again.",
		)
	case globalcommonv1.CloudProvider_AZURE:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL and Azure auth configuration, then try again.",
		)
	case globalcommonv1.CloudProvider_ALICLOUD:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL and role ARN are correct, then try again.",
		)
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported cloud provider: %s", cloudProvider)
	}
}