package repository

import (
	"context"
	"fmt"
	"sort"

	"github.com/aws/aws-sdk-go-v2/aws"
	"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/pingcap/log"
	"go.uber.org/zap"
)

func (r *PDMetaStoreClient) listClusterBranches(ctx context.Context) ([]*ClusterBranch, error) {
	url := fmt.Sprintf("http://%s:%d/api/v1/meta:list", r.pdMetaStoreHost, r.pdMetaStorePort)
	return r.requestPDMetaStore(ctx, url)
}

func (r *DynamoClusterRepo) listClusters(ctx context.Context) ([]*ClusterBranch, []*OriginClusterRecord, error) {
	// scan all clusters with filter using pagination
	expr, err := expression.NewBuilder().WithFilter(expression.Name("tags_kubernetes").Equal(expression.Value(r.k8sName))).Build()
	if err != nil {
		return nil, nil, err
	}

	var allClusters []*ClusterBranch
	var deletingRecords []*OriginClusterRecord
	var lastEvaluatedKey map[string]types.AttributeValue

	for {
		scanInput := &dynamodb.ScanInput{
			TableName:                 aws.String(r.tableName),
			ExpressionAttributeNames:  expr.Names(),
			ExpressionAttributeValues: expr.Values(),
			FilterExpression:          expr.Filter(),
		}

		// Set pagination parameters
		if lastEvaluatedKey != nil {
			scanInput.ExclusiveStartKey = lastEvaluatedKey
		}

		output, err := r.ddbClient.Scan(ctx, scanInput)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to scan items: %s", err)
		}

		// Process current page
		for _, item := range output.Items {
			var record OriginClusterRecord
			err = attributevalue.UnmarshalMap(item, &record)
			if err != nil {
				return nil, nil, fmt.Errorf("failed to unmarshal cluster: %s", err)
			}
			if record.Phase == "Deleting" {
				log.Info("found deleting record in dynamodb", zap.Any("record", record))
				deletingRecords = append(deletingRecords, &record)
			}
			cluster, err := record.ToCluster()
			if err != nil {
				return nil, nil, fmt.Errorf("failed to convert cluster: %s", err)
			}
			allClusters = append(allClusters, cluster)
		}

		// Check if there are more pages
		lastEvaluatedKey = output.LastEvaluatedKey
		if lastEvaluatedKey == nil {
			break
		}
	}

	if len(allClusters) == 0 {
		return nil, nil, fmt.Errorf("no clusters found")
	}
	return allClusters, deletingRecords, nil
}

type DiffResult struct {
	InfraExtra    []*ClusterBranch       `json:"infra_extra"`
	PdExtra       []*ClusterBranch       `json:"pd_extra"`
	Mismatches    []*MismatchedItem      `json:"mismatches"`
	InfraDeleting []*OriginClusterRecord `json:"infra_deleting"`
	Error         string                 `json:"error,omitempty"`
}

type MismatchedItem struct {
	PrimaryID string         `json:"primary_id"`
	InfraItem *ClusterBranch `json:"infra_item"`
	PdItem    *ClusterBranch `json:"pd_item"`
}

func (r *CacheClusterBranchRepo) DiffMetaData() *DiffResult {
	if r.cloudProvider != "aws" {
		return &DiffResult{Error: "only aws cloud provider is supported"}
	}
	logger := log.L().With(zap.String("component", "cache_repo"))
	ctx := context.Background()

	dynamoClusters, deleteingRecords, err := r.dynamoClusterRepo.listClusters(ctx)
	if err != nil {
		logger.Warn("failed to list dynamo clusters", zap.Error(err))
		return &DiffResult{Error: err.Error()}
	}
	deletingClusterIds := make(map[string]struct{}, len(deleteingRecords))
	for _, record := range deleteingRecords {
		deletingClusterIds[record.ClusterId] = struct{}{}
	}
	logger.Info("fetched dynamo clusters", zap.Int("count", len(dynamoClusters)))

	pdClusterBranches, err := r.pdMetaStoreRepo.listClusterBranches(ctx)
	if err != nil {
		logger.Warn("failed to list pd clusters", zap.Error(err))
		return &DiffResult{Error: err.Error()}
	}
	logger.Info("fetched pd clusters", zap.Int("count", len(pdClusterBranches)))

	// Create infra sorted list (dynamo + infra, removing duplicates)
	infraMap := make(map[string]*ClusterBranch)
	for _, cb := range dynamoClusters {
		if cb.KeyspaceID == 0 {
			continue
		}
		if _, ok := deletingClusterIds[cb.PrimaryID]; ok {
			// skip deleting clusters
			continue
		}
		infraMap[cb.PrimaryID] = cb
	}

	pdMap := make(map[string]*ClusterBranch)
	for _, cb := range pdClusterBranches {
		if cb.Type == CBTypeBranch {
			continue
		}
		pdMap[cb.PrimaryID] = cb
	}

	// Check items that exist in both and compare details
	matchCount := 0
	var mismatches []*MismatchedItem

	for primaryID, pdItem := range pdMap {
		if infraItem, exists := infraMap[primaryID]; exists {
			// Item exists in both, compare details
			if r.compareClusterBranchItems(pdItem, infraItem) {
				matchCount++
			} else {
				mismatches = append(mismatches, &MismatchedItem{
					PrimaryID: primaryID,
					InfraItem: infraItem,
					PdItem:    pdItem,
				})
				logger.Warn("Item mismatch found",
					zap.Any("infra", infraItem),
					zap.Any("pd", pdItem),
				)
			}
		}
	}

	logger.Info("Item comparison completed",
		zap.Int("matches", matchCount),
		zap.Int("mismatches", len(mismatches)),
	)

	// Items only in PD (missing from infra)
	var pdExtraItems []*ClusterBranch
	for primaryID, pdItem := range pdMap {
		if _, exists := infraMap[primaryID]; !exists {
			pdExtraItems = append(pdExtraItems, pdItem)
		}
	}

	// Items only in infra (missing from PD)
	var infraExtraItems []*ClusterBranch
	for primaryID, infraItem := range infraMap {
		if _, exists := pdMap[primaryID]; !exists {
			infraExtraItems = append(infraExtraItems, infraItem)
		}
	}

	// Sort extra items for consistent output
	sort.Slice(pdExtraItems, func(i, j int) bool {
		return pdExtraItems[i].PrimaryID < pdExtraItems[j].PrimaryID
	})
	sort.Slice(infraExtraItems, func(i, j int) bool {
		return infraExtraItems[i].PrimaryID < infraExtraItems[j].PrimaryID
	})

	// Final summary
	logger.Info("Diff process completed",
		zap.Int("pd_total", len(pdClusterBranches)),
		zap.Int("infra_total", len(infraMap)),
		zap.Int("matches", matchCount),
		zap.Int("mismatches", len(mismatches)),
		zap.Int("pd_only", len(pdExtraItems)),
		zap.Int("infra_only", len(infraExtraItems)),
	)

	return &DiffResult{
		InfraExtra: infraExtraItems,
		PdExtra:    pdExtraItems,
		Mismatches: mismatches,
		// InfraDeleting: deleteingRecords,
	}
}

// compareClusterBranchItems compares two ClusterBranch items for equality
func (r *CacheClusterBranchRepo) compareClusterBranchItems(item1, item2 *ClusterBranch) bool {
	return item1.Type == item2.Type &&
		item1.PrimaryID == item2.PrimaryID &&
		item1.Name == item2.Name &&
		item1.ClusterID == item2.ClusterID &&
		item1.TenantID == item2.TenantID &&
		item1.ProjectID == item2.ProjectID &&
		item1.KeyspaceID == item2.KeyspaceID &&
		item1.KeyspaceToken == item2.KeyspaceToken &&
		item1.EncryptKeyspace == item2.EncryptKeyspace &&
		item1.MaxRcu == item2.MaxRcu &&
		item1.MinRcu == item2.MinRcu
}