package legacyimpl

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/url"
	"path"
	"strconv"
	"strings"

	"github.com/samber/lo"
	clusterdedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clusterproxy"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/auditlog"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const (
	// Note: All sql statements here are accessible to customers' users.
	enableAuditPluginStmt         = "admin plugins enable audit"
	DisableAuditPluginStmt        = "admin plugins disable audit"
	setAuditLogRedactStatusStmt   = "SET @@GLOBAL.tidb_audit_redact_log = '%s'"
	setAuditLogMaxSizeStmt        = "SET @@GLOBAL.tidb_audit_log_max_size = 10485760"
	flushStmt                     = "FLUSH tidb plugins audit"
	showPluginsStmt               = "SHOW PLUGINS"
	queryAuditLogRedactStatusStmt = "select @@GLOBAL.tidb_audit_redact_log AS log_redacted"
)

func (svc *service) GetAuditLogConfig(
	ctx context.Context,
	clusterID uint64,
) (*globaldedicatedv1.AuditLogConfig, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Service.GetAuditLogConfig")
	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.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	// Get the audit log status from the region server.
	regionalConfig, err := svc.regionServerCli.GetAuditLogConfiguration(
		ctx,
		region,
		&clusterdedicatedv1.GetAuditLogConfigurationRequest{ClusterId: strconv.FormatUint(clusterID, 10)},
	)
	if err != nil {
		return nil, err
	}

	// Convert the regional config to the central config.
	globalConfig := &globaldedicatedv1.AuditLogConfig{
		Id:           0, // leave empty for legacy
		ClusterId:    clusterID,
		BucketUrl:    regionalConfig.GetBucketUrl(),
		Region:       regionalConfig.GetBucketRegion(),
		RoleArn:      regionalConfig.GetS3().GetRoleArn(),
		SasToken:     regionalConfig.GetAzBlob().GetSasToken(),
		Unredacted:   nil, // fill later
		Format:       "",  // leave empty for legacy
		MaxSizeMib:   0,   // leave empty for legacy
		MaxAgeSecond: nil, // leave empty for legacy
		BucketManager: lo.Ternary(
			regionalConfig.GetIsManagedByPingcap(),
			openapidedicatedv1beta1.AuditLogConfig_PINGCAP.String(),
			openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String(),
		),
		Enabled: nil, // fill later
		Legacy:  true,
	}

	// 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()
	}

	// If the cluster is in the paused, pausing or resuming state, return config only.
	if cluster.GetState() == globaldedicatedv1.Cluster_PAUSED ||
		cluster.GetState() == globaldedicatedv1.Cluster_PAUSING ||
		cluster.GetState() == globaldedicatedv1.Cluster_RESUMING {
		return globalConfig, nil
	}

	proxy, err := svc.v2Impl.GetCloudAdminClusterProxy(ctx, cluster, region)
	if err != nil {
		return nil, err
	}

	enable, err := svc.queryPluginStatus(ctx, proxy)
	if err != nil {
		return nil, err
	}
	globalConfig.Enabled = lo.ToPtr(enable)

	redacted, err := svc.queryAuditLogRedactStatus(ctx, proxy)
	if err != nil {
		return nil, err
	}
	globalConfig.Unredacted = lo.ToPtr(!redacted)

	return globalConfig, nil
}

// UpdateAuditLogConfig modifies the audit log configuration for a cluster, idempotently.
// One must ensure all fields in the config are set, even if they are not changed.
func (svc *service) UpdateAuditLogConfig(
	ctx context.Context,
	config *globaldedicatedv1.AuditLogConfig,
) (*globaldedicatedv1.AuditLogConfig, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "Service.UpdateAuditLogConfig")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, config.GetClusterId(), false)
	if err != nil {
		return nil, err
	}
	region, err := svc.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	// Prepare the cluster proxy.
	proxy, err := svc.v2Impl.GetCloudAdminClusterProxy(ctx, cluster, region)
	if err != nil {
		return nil, err
	}

	if !config.GetEnabled() {
		// Disable SQL plugin execution in the cluster through the cluster proxy.
		stmts := []string{
			DisableAuditPluginStmt,
			fmt.Sprintf(setAuditLogRedactStatusStmt, lo.Ternary(config.GetUnredacted(), "OFF", "ON")),
			setAuditLogMaxSizeStmt,
			flushStmt,
		}
		if err := proxy.ExecuteMultiSQL(ctx, stmts); err != nil {
			return nil, err
		}
		// Then return immediately.
		return config, nil
	}

	// Set default values for the audit log config if not set.
	if config.GetBucketManager() == "" ||
		config.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_BUCKET_MANAGER_UNSPECIFIED.String() {
		config.BucketManager = openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String()
	}

	if config.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String() {
		// Validate the audit log config.
		if err := auditlog.ValidateAuditLogConfig(cluster, config, LegacyMinVersion); err != nil {
			return nil, err
		}

		// Check bucket access.
		resp, err := svc.regionServerCli.ValidateAuditLogConfiguration(
			ctx,
			region,
			&clusterdedicatedv1.ValidateAuditLogConfigurationRequest{
				ClusterId:             strconv.FormatUint(config.GetClusterId(), 10),
				AuditLogConfiguration: auditlog.CvtToRegionalConfig(config, cluster.GetCloudProviderId()),
			},
		)
		if err != nil {
			return nil, err
		}
		if !resp.GetIsValid() {
			return nil, auditlog.BuildPrecheckAuditLogConfigBizError(cluster.GetCloudProviderId())
		}

		// Append cluster to path of bucket url.
		bucketURL, err_ := url.Parse(config.GetBucketUrl())
		if err_ != nil {
			return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeParams, "failed to parse bucket url")
		}
		if !strings.Contains(bucketURL.Path, fmt.Sprint(config.GetClusterId())) {
			bucketURL.Path = path.Join(bucketURL.Path, fmt.Sprint(config.GetClusterId()))
			config.BucketUrl = bucketURL.String()
		}
	}

	// Update the audit log status in the region server.
	_, err = svc.regionServerCli.SetAuditLogConfiguration(
		ctx,
		region,
		&clusterdedicatedv1.SetAuditLogConfigurationRequest{
			ClusterId:             strconv.FormatUint(config.GetClusterId(), 10),
			AuditLogConfiguration: auditlog.CvtToRegionalConfig(config, cluster.GetCloudProviderId()),
			ManagedBy: lo.Ternary(config.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_PINGCAP.String(),
				clusterdedicatedv1.SetAuditLogConfigurationRequest_BUCKET_MANAGED_BY_PINGCAP,
				clusterdedicatedv1.SetAuditLogConfigurationRequest_BUCKET_MANAGED_BY_CUSTOMER,
			),
		},
	)
	if err != nil {
		return nil, err
	}

	// Exec enable plugin sql in the cluster via cluster proxy.
	stmts := []string{
		enableAuditPluginStmt,
		fmt.Sprintf(setAuditLogRedactStatusStmt, lo.Ternary(config.GetUnredacted(), "OFF", "ON")),
		setAuditLogMaxSizeStmt,
		flushStmt,
	}
	if err := proxy.ExecuteMultiSQL(ctx, stmts); err != nil {
		return nil, err
	}

	config.Legacy = true // Mark the config as legacy.

	return config, nil
}

func (svc *service) PrecheckAuditLogConfig(
	ctx context.Context,
	config *globaldedicatedv1.AuditLogConfig,
) (*globaldedicatedv1.PrecheckAuditLogConfigResponse, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "PrecheckAuditLogConfig")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, config.GetClusterId(), false)
	if err != nil {
		return nil, err
	}
	region, err := svc.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	if err := auditlog.ValidateAuditLogConfig(cluster, config, LegacyMinVersion); err != nil {
		return nil, err
	}

	// Check bucket access.
	resp, err := svc.regionServerCli.ValidateAuditLogConfiguration(
		ctx,
		region,
		&clusterdedicatedv1.ValidateAuditLogConfigurationRequest{
			ClusterId:             strconv.FormatUint(config.GetClusterId(), 10),
			AuditLogConfiguration: auditlog.CvtToRegionalConfig(config, cluster.GetCloudProviderId()),
		},
	)
	if err != nil {
		return nil, err
	}

	result := &globaldedicatedv1.PrecheckAuditLogConfigResponse{
		Valid: resp.GetIsValid(),
	}
	if !resp.GetIsValid() {
		result.Message = auditlog.BuildPrecheckAuditLogConfigBizError(cluster.GetCloudProviderId()).ErrMsg()
	}

	return result, nil
}

// +-------+---------------+-------+---------------------+------------+---------+
// | Name  | Status        | Type  | Library             | License    | Version |
// +-------+---------------+-------+---------------------+------------+---------+
// | audit | Ready-enable  | Audit | /plugins/audit-1.so | Commercial | 1       |
// +-------+---------------+-------+---------------------+------------+---------+
// | audit | Ready-disable | Audit | /plugins/audit-1.so | Commercial | 1       |
// +-------+---------------+-------+---------------------+------------+---------+
func (svc *service) queryPluginStatus(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (enable bool, err bizerr.BizError) {
	rawBytes, err := proxy.QuerySQL(ctx, showPluginsStmt)
	if err != nil {
		return false, err
	}

	type raw struct {
		Name   []string `json:"Name"`
		Status []string `json:"Status"`
		Type   []string `json:"Type"`
	}

	// The result of show plugins is like:
	// {
	//    "Library": [
	//        "L3BsdWdpbnMvYXVkaXQtMS5zbw=="
	//    ],
	//    "License": [
	//        "Q29tbWVyY2lhbA=="
	//    ],
	//    "Name": [
	//        "YXVkaXQ="
	//    ],
	//    "Status": [
	//        "UmVhZHktZW5hYmxl"
	//    ],
	//    "Type": [
	//        "QXVkaXQ="
	//    ],
	//    "Version": [
	//        "MQ=="
	//    ]
	// }
	var rawObj raw
	if err := json.Unmarshal(rawBytes, &rawObj); err != nil {
		return false, bizerr.WrapError(err, "failed to unmarshal json")
	}

	for i, t := range rawObj.Type {
		t_, err := base64.StdEncoding.DecodeString(t)
		if err != nil {
			return false, bizerr.WrapError(err, "failed to decode base64: %s", t)
		}
		// Find the audit plugin row.
		if strings.EqualFold("Audit", string(t_)) {
			s_, err := base64.StdEncoding.DecodeString(rawObj.Status[i])
			if err != nil {
				return false, bizerr.WrapError(err, "failed to decode base64: %s", rawObj.Status[i])
			}
			// Check the status of the audit plugin.
			if strings.Contains(strings.ToLower(string(s_)), "enable") {
				return true, nil
			} else if strings.Contains(strings.ToLower(string(s_)), "disable") {
				return false, nil
			}
		}
	}

	// Plugin not found, so we assume it is disabled.
	return false, nil
}

func (svc *service) queryAuditLogRedactStatus(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (redacted bool, err bizerr.BizError) {
	rawBytes, err := proxy.QuerySQL(ctx, queryAuditLogRedactStatusStmt)
	if err != nil {
		return false, err
	}

	type raw struct {
		LogRedacted []int `json:"log_redacted"`
	}

	var rawObj raw
	if err := json.Unmarshal(rawBytes, &rawObj); err != nil {
		return false, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "failed to unmarshal json")
	}

	if len(rawObj.LogRedacted) == 0 {
		return false, bizerr.New(bizerr.ErrCodeUnknown, "no redact status found")
	}

	return rawObj.LogRedacted[0] == 1, nil
}