package service

import (
	"context"
	"fmt"
	"strings"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
)

const (
	storageClassAPIVersion = "storage.k8s.io/v1"
	storageClassKind       = "StorageClass"
)

// ensureNewArchStorageClass makes sure a StorageClass that matches the
// rcu tier's iops / throughput exists on the dataplane k8s cluster, and
// returns the SC name to put into the TidbCluster CR's PVC template.
//
// Behavior:
//   - Build a per-spec SC named cdc-{cloud}-{iops}-{throughput}
//     (e.g. cdc-aws-6000-500, cdc-alicloud-6000-500). Cloud-specific
//     provisioner / parameter layouts come from the same provider
//     handlers infra-provider/ticdc-service and
//     cluster-service-ng/internal/dataplane-manager use.
//   - StorageClass is cluster-scoped, so the cdc-{cloud}-{iops}-{tput}
//     name is the same across all serverless-service tenants on the
//     same dataplane and is shared.
func (s *cdcService) ensureNewArchStorageClass(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	cloudProvider string,
	spec newArchTiCDCInstanceType,
) (string, bizerr.BizError) {
	cfg := config.Global().CDCConfig.NewArch
	name := storageClassName(cloudProvider, spec)
	if name == "" {
		return "", nil
	}

	// StorageClass is cluster-scoped, but dbaas-proxy's request validator
	// requires Namespace on every K8sResourceRequest regardless of scope —
	// the value is ignored once the request reaches the apiserver, but the
	// proxy layer 400s without it. Reuse cfg.Namespace so the value is
	// well-formed and consistent with every other proxy call we make.
	if _, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  storageClassAPIVersion,
		Kind:        storageClassKind,
		Namespace:   cfg.Namespace,
		Name:        name,
	}); err == nil {
		return name, nil
	}

	obj, err := buildStorageClassForProvider(name, cloudProvider, spec)
	if err != nil {
		return "", bizerr.WrapError(err, "build storage class %s", name)
	}
	if _, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Create",
		APIVersion:  storageClassAPIVersion,
		Kind:        storageClassKind,
		Namespace:   cfg.Namespace,
		Object:      obj,
	}); err != nil {
		// Tolerate concurrent create — StorageClass is cluster-scoped
		// and another shard reconcile on this dataplane may have
		// created the exact same SC name (it's derived from spec).
		if isAlreadyExistsErr(err) {
			return name, nil
		}
		return "", bizerr.WrapError(err, "create storage class %s", name)
	}
	return name, nil
}

func storageClassName(cloudProvider string, spec newArchTiCDCInstanceType) string {
	if spec.Iops == 0 && spec.ThroughputMbs == 0 {
		return ""
	}
	return fmt.Sprintf("cdc-%s-%d-%d", strings.ToLower(cloudProvider), spec.Iops, spec.ThroughputMbs)
}

func buildStorageClassForProvider(
	name, cloudProvider string,
	spec newArchTiCDCInstanceType,
) (map[string]any, error) {
	switch strings.ToLower(cloudProvider) {
	case config.ProviderAWS:
		return buildAWSStorageClass(name, spec), nil
	case config.ProviderGCP:
		return buildGCPStorageClass(name, spec), nil
	case config.ProviderAzure:
		return buildAzureStorageClass(name, spec), nil
	case config.ProviderAliCloud:
		return buildAlicloudStorageClass(name, spec), nil
	default:
		return nil, fmt.Errorf("unsupported cloud provider %s for newarch ticdc storage class", cloudProvider)
	}
}

func buildAWSStorageClass(name string, spec newArchTiCDCInstanceType) map[string]any {
	params := map[string]any{
		"type":                      "gp3",
		"csi.storage.k8s.io/fstype": "ext4",
		"encrypted":                 "true",
	}
	if spec.Iops > 0 {
		params["iops"] = fmt.Sprintf("%d", spec.Iops)
	}
	if spec.ThroughputMbs > 0 {
		params["throughput"] = fmt.Sprintf("%d", spec.ThroughputMbs)
	}
	return baseStorageClass(name, "ebs.csi.aws.com", params)
}

func buildGCPStorageClass(name string, _ newArchTiCDCInstanceType) map[string]any {
	// GCP pd-balanced throughput / iops scale with volume size; the same
	// fixed type is used in infra-provider/ticdc-service. If a higher tier
	// (pd-ssd / hyperdisk) is needed later, add a switch on rcu here.
	params := map[string]any{
		"type":                      "pd-balanced",
		"csi.storage.k8s.io/fstype": "ext4",
	}
	return baseStorageClass(name, "pd.csi.storage.gke.io", params)
}

func buildAzureStorageClass(name string, spec newArchTiCDCInstanceType) map[string]any {
	// Azure PremiumV2_LRS lets us set IOPS / throughput per-disk directly.
	params := map[string]any{
		"skuName":                   "PremiumV2_LRS",
		"csi.storage.k8s.io/fstype": "ext4",
	}
	if spec.Iops > 0 {
		params["DiskIOPSReadWrite"] = fmt.Sprintf("%d", spec.Iops)
	}
	if spec.ThroughputMbs > 0 {
		params["DiskMBpsReadWrite"] = fmt.Sprintf("%d", spec.ThroughputMbs)
	}
	return baseStorageClass(name, "disk.csi.azure.com", params)
}

func buildAlicloudStorageClass(name string, spec newArchTiCDCInstanceType) map[string]any {
	// cloud_auto picks the right ESSD performance level automatically.
	// PL1 ≈ up to 50k iops / 350 MB/s — enough for any rcu tier we have.
	params := map[string]any{
		"type":            "cloud_auto",
		"fstype":          "ext4",
		"encrypted":       "true",
		"burstingEnabled": "true",
	}
	if spec.Iops > 0 {
		params["provisionedIops"] = fmt.Sprintf("%d", spec.Iops)
	}
	return baseStorageClass(name, "diskplugin.csi.alibabacloud.com", params)
}

func baseStorageClass(name, provisioner string, params map[string]any) map[string]any {
	return map[string]any{
		"apiVersion": storageClassAPIVersion,
		"kind":       storageClassKind,
		"metadata": map[string]any{
			"name": name,
			"labels": map[string]any{
				"app.kubernetes.io/managed-by": "serverless-service",
				"tidbcloud.com/cdc-arch":       "ticdc_newarch",
			},
		},
		"provisioner":          provisioner,
		"reclaimPolicy":        "Delete",
		"volumeBindingMode":    "WaitForFirstConsumer",
		"allowVolumeExpansion": true,
		"mountOptions":         []string{"nodelalloc", "noatime"},
		"parameters":           params,
	}
}