package infra

import (
	"context"
	"errors"
	"fmt"

	"github.com/pingcap/tidb/pkg/util"
	auditlogutil "github.com/tidbcloud/aws-shared-provider/pkg/auditlog/util"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	infracache "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/cache"
	infradynamo "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/dynamo"
	_ "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/metrics/prom"
	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
	infraots "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/tablestore"
	_ "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/metrics/prom"
	consumerredis "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/redis"
	unmarshaldynamo "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/dynamo"
	unmarshalots "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/tablestore"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	ctrl "sigs.k8s.io/controller-runtime"
)

const (
	infraClusterCacheSize = 10000
)

type regionalInfraImpl struct {
	infraClusterCli     infraclient.Client
	LocalKubernetesName string
}

func NewRegionalInfraWithClient(client infraclient.Client, localKubernetesName string) RegionalInfra {
	return &regionalInfraImpl{infraClusterCli: client, LocalKubernetesName: localKubernetesName}
}

func NewRegionalInfra(
	ctx context.Context,
	conf *RegionalInfraConfig,
) (RegionalInfra, *consumerredis.Consumer, error) {
	tlsConfig, err := util.NewTLSConfig(
		util.WithCAPath(conf.InfraRedisClientTLSConfig.CA),
		util.WithCertAndKeyPath(
			conf.InfraRedisClientTLSConfig.Cert,
			conf.InfraRedisClientTLSConfig.Key,
		),
	)
	if err != nil {
		return nil, nil, fmt.Errorf("load infra redis client tls config err: %w", err)
	}

	var infraClusterCli infraclient.Client
	var infraClusterRedisConsumer *consumerredis.Consumer
	if conf.EnableInfraMysqlClient {
		if conf.InfraRdsSecretArn == "" {
			return nil, nil, fmt.Errorf("infra rds secret arn is required when enable infra mysql client")
		}
		rdsConfig, err := auditlogutil.GetRdsConfig(conf.CloudProviderType, conf.InfraRdsSecretArn, conf.Region)
		if err != nil {
			return nil, nil, fmt.Errorf("get rds config err: %w", err)
		}
		underlayClient, err := inframysql.NewMySQLClient(
			rdsConfig.ToGormutilConfig(),
			"gateway",
			conf.KubeClusterName,
			inframysql.DefaultIndexField2name,
			serverless_infra.DefaultClusterEventTTL,
			true,
		)
		if err != nil {
			return nil, nil, fmt.Errorf("create infra mysql client err: %w", err)
		}
		underlayClient.SetLogger(ctrl.Log)

		index2key := map[string]string{
			"status.keyspace_token-index":         "keyspace_token",
			"status.zone_redundant_enabled-index": "zone_redundant_enabled",
		}
		infraClusterCli, _ = infracache.NewCacheClientV2(conf.CacheServerAddr, index2key, []string{"status.keyspace_token-index"}, infraClusterCacheSize, underlayClient, true)
		infraClusterCli.SetLogger(ctrl.Log)
	} else {
		var underlayClient infraclient.Client
		switch {
		case conf.AwsCfg != nil:
			underlayClient = infradynamo.NewDynamoClient(conf.AwsCfg, conf.InfraClusterTable,
				map[string]string{"status.keyspace_token-index": "status.keyspace_token"},
			)
		case conf.AlicloudCred != nil:
			otsEndpoint := fmt.Sprintf("https://%s.%s.ots.aliyuncs.com", conf.OtsInstanceName, conf.Region)
			underlayClient = infraots.NewTableStoreClient(conf.AlicloudCred, otsEndpoint, conf.OtsInstanceName,
				conf.InfraClusterTable,
				map[string]string{"status.keyspace_token-index": "status.keyspace_token"},
			)
		default:
			return nil, nil, fmt.Errorf("unsupported underlay infra cluster client")
		}

		var infraClusterCo infraclient.CacheOperate
		infraClusterCli, infraClusterCo = infracache.NewCacheClient(infraClusterCacheSize, underlayClient, true)
		infraClusterCli.SetLogger(ctrl.Log)
		infraClusterRedisConsumer = &consumerredis.Consumer{
			Addr:         conf.InfraRedisAddr,
			TLSConfig:    tlsConfig,
			Stream:       conf.InfraClusterTable + "_" + conf.KubeClusterName,
			CacheOperate: infraClusterCo,
			Unmarshal: func(data any) (serverless_infra.Object, error) {
				if conf.AwsCfg != nil {
					return unmarshaldynamo.Unmarshal(data, func() serverless_infra.Object { return &serverless_infra.Cluster{} })
				} else if conf.AlicloudCred != nil {
					return unmarshalots.Unmarshal(data, func() serverless_infra.Object { return &serverless_infra.Cluster{} })
				}
				return nil, fmt.Errorf("unknown infra cluster")
			},
		}
		infraClusterRedisConsumer.SetLogger(ctrl.Log)
		if err := infraClusterRedisConsumer.Init(ctx); err != nil {
			return nil, nil, err
		}
	}

	return NewRegionalInfraWithClient(infraClusterCli, conf.KubeClusterName), infraClusterRedisConsumer, nil
}

func (i *regionalInfraImpl) GetCluster(ctx context.Context, namespace, name string) (*serverless_infra.Cluster, error) {
	infraCluster := &serverless_infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: namespace, Name: name}}
	if err := i.infraClusterCli.Get(ctx, infraCluster); err != nil {
		return nil, err
	}
	return infraCluster, nil
}

func (i *regionalInfraImpl) GetClusterWithToken(ctx context.Context, token string) (*serverless_infra.Cluster, error) {
	clusterKey := &meta.ObjectKey{}
	if err := i.infraClusterCli.GetIndex(ctx, token, "status.keyspace_token-index", clusterKey); err != nil {
		if errors.Is(err, infraclient.ErrNotFound) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	c := &serverless_infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: clusterKey.Namespace, Name: clusterKey.Name}}
	if err := i.infraClusterCli.Get(ctx, c); err != nil {
		if errors.Is(err, infraclient.ErrNotFound) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	return c, nil
}