package server

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
)

// dedicatedReplayProvider implements replayClusterProvider for Dedicated-tier clusters.
// It holds only the fields extracted from replayContext at construction time, avoiding
// circular references and keeping its dependencies explicit.
type dedicatedReplayProvider struct {
	kubeClient      client.Client
	namespacePrefix string
	deployPosition  string
	sourceAuditLog  *apiv1.Replay_AuditLogAddr
	clusterRef      *apiv1.Replay_ClusterRef
	cloudCluster    *cloudv1.Cluster
	loadOnce        sync.Once
}

func newDedicatedReplayProvider(rctx *replayContext, ref *apiv1.Replay_ClusterRef) *dedicatedReplayProvider {
	return &dedicatedReplayProvider{
		kubeClient:      rctx.kubeClient,
		namespacePrefix: rctx.namespacePrefix,
		deployPosition:  rctx.deployPosition,
		sourceAuditLog:  rctx.sourceAuditLog,
		clusterRef:      ref,
	}
}

func (p *dedicatedReplayProvider) GetClusterType() apiv1.Replay_ClusterType {
	return p.clusterRef.ClusterType
}

func (p *dedicatedReplayProvider) LoadSourceMaterial(ctx context.Context) (replaySourceMaterial, error) {
	cluster, err := p.getCloudCluster(ctx)
	if err != nil {
		if errors.Is(err, errReplayCloudClusterMissing) {
			return replaySourceMaterial{}, errReplaySourceClusterMissing
		}
		return replaySourceMaterial{}, err
	}
	return p.buildSourceMaterial(cluster)
}

func (p *dedicatedReplayProvider) LoadTargetEndpoints(ctx context.Context) ([]string, error) {
	targetCluster, err := p.getCloudCluster(ctx)
	if err != nil {
		if errors.Is(err, errReplayCloudClusterMissing) {
			return nil, errReplayTargetClusterMissing
		}
		return nil, err
	}

	logger := tilog.WithContextL(ctx).With(
		zap.String("project", p.clusterRef.ProjectId),
		zap.String("cluster", p.clusterRef.ClusterId),
	)
	targetGroupIDs := make([]string, 0, len(targetCluster.Status.ExtraEndpoints)+1)
	targetGroupIDs = append(targetGroupIDs, targetCluster.GetDefaultGroupID())
	for _, group := range targetCluster.Status.ExtraEndpoints {
		targetGroupIDs = append(targetGroupIDs, group.Name)
	}
	endpoints := make([]string, 0, len(targetGroupIDs))
	for _, targetGroupID := range targetGroupIDs {
		endpoint, err := p.buildEndpoint(targetCluster, targetGroupID)
		if err != nil {
			logger.Error("Failed to build target endpoint for task group", zap.String("group", targetGroupID), zap.Error(err))
			endpoints = append(endpoints, "")
			continue
		}
		endpoints = append(endpoints, endpoint)
	}
	return endpoints, nil
}

func (p *dedicatedReplayProvider) ResolveDataplane(ctx context.Context) (replayDataplaneResolution, error) {
	cluster, err := p.getCloudCluster(ctx)
	if err != nil {
		return replayDataplaneResolution{}, err
	}

	return replayDataplaneResolution{
		kubernetes: cluster.Spec.Kubernetes,
	}, nil
}

func (p *dedicatedReplayProvider) buildSourceMaterial(cluster *cloudv1.Cluster) (replaySourceMaterial, error) {
	material := replaySourceMaterial{
		units: make([]replaySourceTaskUnit, 0, len(cluster.Status.ExtraEndpoints)+1),
	}

	defaultGroupID := cluster.GetDefaultGroupID()
	unit, err := p.buildSourceTaskUnit(cluster, defaultGroupID)
	if err != nil {
		return replaySourceMaterial{}, err
	}
	material.units = append(material.units, unit)

	if p.isSupportGroupsAuditLog() {
		for _, group := range cluster.Status.ExtraEndpoints {
			sourceUnit, err := p.buildSourceTaskUnit(cluster, group.Name)
			if err != nil {
				return replaySourceMaterial{}, err
			}
			material.units = append(material.units, sourceUnit)
		}
	}

	return material, nil
}

func (p *dedicatedReplayProvider) isSupportGroupsAuditLog() bool {
	if !p.supportDynamicLogPath(p.cloudCluster) {
		return false
	}
	if p.sourceAuditLog != nil && p.sourceAuditLog.Source != nil {
		if aws := p.sourceAuditLog.GetAws(); aws != nil && aws.S3Url != "" {
			return false
		}
	}
	return true
}

func (p *dedicatedReplayProvider) buildSourceTaskUnit(
	cluster *cloudv1.Cluster,
	groupID string,
) (replaySourceTaskUnit, error) {
	addr, err := p.buildAuditLogAddr(cluster, groupID)
	if err != nil {
		return replaySourceTaskUnit{}, err
	}

	return replaySourceTaskUnit{
		name:          fmt.Sprintf("group-%s", groupID),
		auditLogAddrs: []cloudv1.AuditLogAddr{addr},
	}, nil
}

func (p *dedicatedReplayProvider) buildEndpoint(cluster *cloudv1.Cluster, groupID string) (string, error) {
	// Same project: use internal service address.
	if p.deployPosition == DeployPositionTarget {
		if cluster.GetDefaultGroupID() == groupID {
			return fmt.Sprintf("db-tidb.tidb%s.svc:4000", cluster.Name), nil
		}
		extraGroup, ok := cluster.GetExtraGroupStatus(groupID)
		if !ok || extraGroup.TiDB == nil || len(extraGroup.TiDB.Members) == 0 {
			return "", fmt.Errorf("extra tidb group not found for cluster %s", cluster.Name)
		}
		extraTiDBName := extractPrefix(extraGroup.TiDB.Members[0].Name)
		return fmt.Sprintf("%s-tidb.tidb%s.svc:4000", extraTiDBName, cluster.Name), nil
	}

	// Cross-cluster: use private endpoint.
	sqlEndpoints, ok := cluster.GetSQLEndpointStatus(groupID)
	if !ok {
		return "", fmt.Errorf("sql endpoint status not found for cluster %s group %s", cluster.Name, groupID)
	}
	if sqlEndpoints.PrivateEndpoint == nil {
		return "", fmt.Errorf("private endpoint is not available for cluster %s", cluster.Name)
	}
	return fmt.Sprintf("%s:%d", sqlEndpoints.PrivateEndpoint.Host, sqlEndpoints.PrivateEndpoint.Port), nil
}

func (p *dedicatedReplayProvider) getCloudCluster(ctx context.Context) (*cloudv1.Cluster, error) {
	if p.clusterRef == nil {
		return nil, errReplayCloudClusterMissing
	}
	p.loadOnce.Do(func() {
		clusterNamespace := genNamespace(p.namespacePrefix, p.clusterRef.ProjectId)
		cluster := &cloudv1.Cluster{}
		if err := p.kubeClient.Get(ctx, client.ObjectKey{Name: p.clusterRef.ClusterId, Namespace: clusterNamespace}, cluster); err != nil {
			tilog.WithContextL(ctx).Error("Failed to get cloud cluster",
				zap.String("project", p.clusterRef.ProjectId),
				zap.String("cluster", p.clusterRef.ClusterId),
				zap.String("namespace", clusterNamespace),
				zap.Error(err))
			return
		}

		validExtraEndpoints := make([]cloudv1.ExtraGroupStatus, 0, len(cluster.Status.ExtraEndpoints))
		for _, endpoint := range cluster.Status.ExtraEndpoints {
			if endpoint.Phase == cloudv1.ClusterPhaseRunning &&
				endpoint.TiDB != nil && len(endpoint.TiDB.Members) > 0 {
				validExtraEndpoints = append(validExtraEndpoints, endpoint)
			}
		}
		cluster.Status.ExtraEndpoints = validExtraEndpoints
		p.cloudCluster = cluster
	})
	if p.cloudCluster == nil {
		return nil, fmt.Errorf("unable to load cloud cluster %s: %w", p.clusterRef.ClusterId, errReplayCloudClusterMissing)
	}
	return p.cloudCluster, nil
}

func (p *dedicatedReplayProvider) buildAuditLogAddr(
	sourceCluster *cloudv1.Cluster,
	groupID string,
) (cloudv1.AuditLogAddr, error) {
	if sourceCluster == nil {
		return cloudv1.AuditLogAddr{}, fmt.Errorf("source cluster is nil")
	}
	audit := sourceCluster.Spec.Cluster.TiDB.Audit
	if audit == nil || !audit.Enable || audit.BucketURL == "" {
		return cloudv1.AuditLogAddr{}, fmt.Errorf("audit log is not enabled for source cluster %s", sourceCluster.Name)
	}

	switch sourceCluster.GetProviderType() {
	case consts.ProviderTypeEKS:
		return p.buildEKSAuditLogAddr(sourceCluster, groupID)
	default:
		return cloudv1.AuditLogAddr{}, unsupportedReplayCloudProvider(sourceCluster.GetProviderType())
	}
}

func (p *dedicatedReplayProvider) buildEKSAuditLogAddr(cluster *cloudv1.Cluster,
	groupID string) (cloudv1.AuditLogAddr, error) {
	audit := cluster.Spec.Cluster.TiDB.Audit
	bucketURL := strings.TrimSuffix(audit.BucketURL, "/")
	addr := cloudv1.AuditLogAddr{
		AWS: &cloudv1.AWSAuditLog{
			S3URL:      bucketURL + "/tidb-",
			RoleArn:    audit.RoleArn,
			ExternalID: ptr.To(cluster.GetProjectID()),
		},
	}
	if p.supportDynamicLogPath(cluster) {
		addr.AWS.S3URL = fmt.Sprintf("%s/%s_tidb-", bucketURL, groupID)
	}
	err := p.overrideSourceAuditLog(&addr)
	return addr, err
}

func (p *dedicatedReplayProvider) supportDynamicLogPath(cluster *cloudv1.Cluster) bool {
	auditSidecarVersion := cluster.GetAnnotations()[consts.AuditLogSidecarImageTag]
	return CheckIfAuditSidecarSupportDynamicLogPath(context.Background(), auditSidecarVersion)
}

func (p *dedicatedReplayProvider) overrideSourceAuditLog(addr *cloudv1.AuditLogAddr) error {
	if aws := p.sourceAuditLog.GetAws(); aws != nil {
		if addr.AWS == nil {
			return fmt.Errorf("%w: aws audit log", errReplaySourceAuditLogProviderMismatch)
		}
		if err := validateAWSAuditLog(aws); err != nil {
			return err
		}
		if aws.S3Url != "" {
			addr.AWS.S3URL = aws.S3Url
		}
		if aws.RoleArn != "" {
			addr.AWS.RoleArn = aws.RoleArn
		}
		if aws.ExternalId != nil {
			addr.AWS.ExternalID = aws.ExternalId
		}
	}
	return nil
}