package repository

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/expression"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
	"github.com/tidbcloud/serverless-metering/pkg/config"
)

type OriginClusterRecord struct {
	ClusterId       string            `dynamodbav:"name"`
	TenantID        string            `dynamodbav:"tags_tenant"`
	ProjectID       string            `dynamodbav:"tags_project"`
	KeyspaceID      uint32            `dynamodbav:"status.keyspace_id"`
	KeyspaceToken   string            `dynamodbav:"status.keyspace_token"`
	EncryptKeyspace bool              `dynamodbav:"spec.encrypt_keyspace"`
	Annotations     map[string]string `dynamodbav:"annotations"`
	K8sName         string            `dynamodbav:"tags_kubernetes"`
	Phase           string            `dynamodbav:"status.phase"`
	TiDB            TiDBRecord        `dynamodbav:"spec.tidb"`
}

type TiDBRecord struct {
	MinRcu int64 `dynamodbav:"min_rcu"`
	MaxRcu int64 `dynamodbav:"max_rcu"`
}

func (o *OriginClusterRecord) ToCluster() (*ClusterBranch, error) {
	clusterId, err := strconv.ParseUint(o.ClusterId, 10, 64)
	if err != nil {
		return nil, err
	}
	projectId, err := strconv.ParseUint(o.ProjectID, 10, 64)
	if err != nil {
		return nil, err
	}
	tenantId, err := strconv.ParseUint(o.TenantID, 10, 64)
	if err != nil {
		return nil, err
	}
	return &ClusterBranch{
		Type:            CBTypeCluster,
		PrimaryID:       o.ClusterId,
		ClusterID:       clusterId,
		TenantID:        tenantId,
		ProjectID:       projectId,
		KeyspaceID:      o.KeyspaceID,
		KeyspaceToken:   o.KeyspaceToken,
		EncryptKeyspace: o.EncryptKeyspace,
		MinRcu:          o.TiDB.MinRcu,
		MaxRcu:          o.TiDB.MaxRcu,
	}, nil
}

type DynamoClusterRepo struct {
	ddbClient *dynamodb.Client
	tableName string
	k8sName   string
}

func NewDynamoClusterRepo() (*DynamoClusterRepo, error) {
	cfg := config.Global()
	optFns := []func(*awsconfig.LoadOptions) error{
		awsconfig.WithRegion(cfg.Region),
	}
	if cfg.Stage == "local" {
		optFns = append(optFns, awsconfig.WithSharedConfigProfile("dbaas-dev"))
	}
	awscfg, err := awsconfig.LoadDefaultConfig(context.TODO(), optFns...)
	if err != nil {
		return nil, err
	}

	ddbClient := dynamodb.NewFromConfig(awscfg)

	return &DynamoClusterRepo{
		ddbClient: ddbClient,
		tableName: cfg.Repo.DynamodbTable,
		k8sName:   cfg.K8sName,
	}, nil
}

func (r *DynamoClusterRepo) GetClusterByID(ctx context.Context, clusterID string) (*ClusterBranch, error) {
	keyCond := expression.KeyEqual(expression.Key("name"), expression.Value(clusterID))
	return r.getCluster(ctx, keyCond, "name-index")
}

func (r *DynamoClusterRepo) GetClusterByKeyspaceID(ctx context.Context, keyspaceID string) (*ClusterBranch, error) {
	kid, err := strconv.Atoi(keyspaceID)
	if err != nil {
		return nil, fmt.Errorf("not found")
	}
	keyCond := expression.KeyEqual(expression.Key("status.keyspace_id"), expression.Value(kid))
	return r.getCluster(ctx, keyCond, "status.keyspace_id-index")
}

func (r *DynamoClusterRepo) getCluster(ctx context.Context, keyCond expression.KeyConditionBuilder, indexName string) (*ClusterBranch, error) {
	// access with secondary index
	expr, err := expression.NewBuilder().WithKeyCondition(keyCond).Build()
	if err != nil {
		return nil, err
	}
	output, err := r.ddbClient.Query(ctx, &dynamodb.QueryInput{
		TableName:                 aws.String(r.tableName),
		ExpressionAttributeNames:  expr.Names(),
		ExpressionAttributeValues: expr.Values(),
		KeyConditionExpression:    expr.KeyCondition(),
		IndexName:                 aws.String(indexName),
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get item with secondary index: %s", err)
	}
	if len(output.Items) == 0 {
		return nil, fmt.Errorf("cluster not found with secondary index")
	}

	clusters := make([]*ClusterBranch, 0)
	for _, item := range output.Items {
		// access with primary index
		keyConds := make(map[string]types.Condition)
		for k, v := range item {
			// only need primary key, but the output may contains secondary index
			if k != "name" && k != "namespace" {
				continue
			}
			keyConds[k] = types.Condition{
				ComparisonOperator: types.ComparisonOperatorEq,
				AttributeValueList: []types.AttributeValue{v},
			}
		}
		expr, _ = expression.NewBuilder().WithFilter(expression.Name("tags_kubernetes").Equal(expression.Value(r.k8sName))).Build()
		keyOutput, err := r.ddbClient.Query(ctx, &dynamodb.QueryInput{
			TableName:                 aws.String(r.tableName),
			ExpressionAttributeNames:  expr.Names(),
			ExpressionAttributeValues: expr.Values(),
			KeyConditions:             keyConds,
			FilterExpression:          expr.Filter(),
		})
		if err != nil {
			return nil, fmt.Errorf("failed to get item with primary index: %s", err)
		}
		if len(keyOutput.Items) == 0 {
			continue
		}

		var record OriginClusterRecord
		err = attributevalue.UnmarshalMap(keyOutput.Items[0], &record)
		if err != nil {
			return nil, fmt.Errorf("failed to unmarshal cluster: %s", err)
		}
		if strings.HasPrefix(record.ClusterId, "e2e") {
			continue
		}
		cluster, err := record.ToCluster()
		if err != nil {
			return nil, fmt.Errorf("failed to convert cluster: %s", err)
		}
		clusters = append(clusters, cluster)
	}
	if len(clusters) == 0 {
		return nil, fmt.Errorf("cluster not found")
	}
	if len(clusters) > 1 {
		return nil, fmt.Errorf("multiple clusters found")
	}
	return clusters[0], nil
}