package impl

import (
	"context"
	"fmt"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

type premiumReplayClusterProvider struct {
	auditLogURLStrategy    replayAuditLogURLStrategy
	mgmtClient             grpcli.MgmtSvcGrpcClient
	clusterID              uint64
	infraProvider          string
	sourceAuditLogAddr     *entity.AuditLogAddr
	targetClusterEndpoints []*globalngv1.Endpoint
}

func newPremiumReplayClusterProvider(
	mgmtClient grpcli.MgmtSvcGrpcClient,
	cluster *mgmt.Cluster,
	replayEnt *entity.Replay,
) *premiumReplayClusterProvider {
	var sourceAuditLogAddr *entity.AuditLogAddr
	if replayEnt != nil {
		sourceAuditLogAddr = replayEnt.SourceAuditCredential
	}
	infraProvider := domain.CloudProviderToInfraProvider(cluster.GetCloudProvider())
	provider := &premiumReplayClusterProvider{
		auditLogURLStrategy:    newReplayAuditLogURLStrategy(cluster.GetRegion(), infraProvider),
		mgmtClient:             mgmtClient,
		clusterID:              cluster.GetClusterId(),
		infraProvider:          infraProvider,
		sourceAuditLogAddr:     sourceAuditLogAddr,
		targetClusterEndpoints: cluster.GetNextGenCluster().GetEndpoints(),
	}
	return provider
}

// PrepareTaskSourceAuditLogs normalizes task-level source audit-log URLs.
// In manual task mode, each explicit replay task owns its source inputs, so
// replay-level SourceAuditCredential is not forwarded as Defaults.SourceAuditLog.
func (p *premiumReplayClusterProvider) PrepareTaskSourceAuditLogs(tasks entity.ReplayTaskList) bizerr.BizError {
	for i := range tasks {
		for j := range tasks[i].DataSource.SourceAuditLogAddrs {
			addr := &tasks[i].DataSource.SourceAuditLogAddrs[j]
			if err := p.auditLogURLStrategy.normalizeSourceAuditLogURL(addr); err != nil {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"replay_tasks[%d].data_source.source_audit_log_addrs[%d] is invalid: %s",
					i, j, err.Error(),
				)
			}
		}
	}
	return nil
}

// PrepareSourceAuditLog prepares replay-level SourceAuditCredential for auto
// mode. Premium sources resolve supported source audit-log material from
// management settings; manual task mode skips this method because each task owns
// its input. The returned material is normalized before being forwarded as
// Defaults.SourceAuditLog. GKE is rejected until its replay credential contract
// is defined.
func (p *premiumReplayClusterProvider) PrepareSourceAuditLog(ctx context.Context) (*entity.AuditLogAddr, bizerr.BizError) {
	if p.infraProvider == string(domain.InfraProviderTypeGKE) {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"GKE Premium source audit log setting is not supported yet for source cluster %d",
			p.clusterID,
		)
	}
	sourceAuditLogAddr, bizErr := p.loadSourceAuditLogAddrFromSetting(ctx)
	if bizErr != nil {
		return nil, bizErr
	}
	if err := p.auditLogURLStrategy.normalizeSourceAuditLogURL(sourceAuditLogAddr); err != nil {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"source cluster %d has invalid source_audit_credential: %s",
			p.clusterID, err.Error(),
		)
	}
	return sourceAuditLogAddr, nil
}

// loadSourceAuditLogAddrFromSetting maps management audit-log settings into the
// replay-level source audit-log format expected by infra-provider. ACK preserves
// caller-provided Aliyun AK/SK while taking the bucket URL from mgmt.
func (p *premiumReplayClusterProvider) loadSourceAuditLogAddrFromSetting(ctx context.Context) (*entity.AuditLogAddr, bizerr.BizError) {
	if p.mgmtClient == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "mgmt client is required to resolve source audit log material")
	}
	setting, bizErr := p.mgmtClient.GetNextGenDatabaseAuditLogSetting(ctx, p.clusterID)
	if bizErr != nil {
		return nil, bizErr
	}
	if setting == nil {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"source cluster %d does not have audit log setting",
			p.clusterID,
		)
	}
	bucketURL := setting.GetBucketUrl()
	if bucketURL == "" {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"source cluster %d does not have audit log bucket url",
			p.clusterID,
		)
	}

	response := p.sourceAuditLogAddr
	if response == nil {
		response = &entity.AuditLogAddr{}
	}

	if bizErr := p.fillSourceAuditLogAddrFromSetting(response, setting); bizErr != nil {
		return nil, bizErr
	}
	return response, nil
}

func (p *premiumReplayClusterProvider) fillSourceAuditLogAddrFromSetting(
	response *entity.AuditLogAddr,
	setting *globalngv1.AuditLogSetting,
) bizerr.BizError {
	bucketURL := setting.GetBucketUrl()
	switch p.infraProvider {
	case string(domain.InfraProviderTypeEKS):
		roleARN := setting.GetCloudRoleIdentity()
		if roleARN == "" {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"source cluster %d does not have audit log cloud role identity",
				p.clusterID,
			)
		}
		var externalID *string
		if setting.GetExternalId() != "" {
			externalIDValue := setting.GetExternalId()
			externalID = &externalIDValue
		}
		response.AWS = &entity.AWSAuditLog{
			S3URL:      bucketURL,
			RoleARN:    roleARN,
			ExternalID: externalID,
		}
		return nil
	case string(domain.InfraProviderTypeGKE):
		return bizerr.New(
			bizerr.ErrCodeParams,
			"GKE Premium source audit log setting is not supported yet for source cluster %d",
			p.clusterID,
		)
	case string(domain.InfraProviderTypeACK):
		aliyunAuditLog := &entity.AliyunAuditLog{BucketURL: bucketURL}
		if response.Aliyun != nil {
			aliyunAuditLog.AccessKeyID = response.Aliyun.AccessKeyID
			aliyunAuditLog.SecretAccessKey = response.Aliyun.SecretAccessKey
		}
		response.Aliyun = aliyunAuditLog
		return nil
	default:
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported infra provider %s for source audit log setting",
			p.infraProvider,
		)
	}
}

// PrepareTargetEndpoint resolves the Premium target endpoint for auto-mode
// callers. Manual task mode skips this method because each task carries target
// connection details.
func (p *premiumReplayClusterProvider) PrepareTargetEndpoint() (string, bizerr.BizError) {
	if len(p.targetClusterEndpoints) == 0 {
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"missing nextgen cluster endpoints",
		)
	}

	// Replay traffic must stay on private Premium connectivity. Prefer the most
	// constrained private endpoint first across all endpoint records.
	for _, endpoint := range p.targetClusterEndpoints {
		if endpoint == nil {
			continue
		}
		for _, privateLinkEndpoint := range endpoint.GetPrivateLinkEndpoints() {
			if privateLinkEndpoint == nil {
				continue
			}
			if addr, ok := p.formatServerlessHostPort(privateLinkEndpoint.GetHost(), int64(privateLinkEndpoint.GetPort())); ok {
				return addr, nil
			}
		}
	}

	for _, endpoint := range p.targetClusterEndpoints {
		if endpoint == nil {
			continue
		}
		if vpcPeeringEndpoint := endpoint.GetVpcPeeringEndpoint(); vpcPeeringEndpoint != nil {
			if addr, ok := p.formatServerlessHostPort(vpcPeeringEndpoint.GetHost(), int64(vpcPeeringEndpoint.GetPort())); ok {
				return addr, nil
			}
		}
	}

	return "", bizerr.New(
		bizerr.ErrCodeParams,
		"has no usable nextgen private endpoint (private link and vpc peering endpoints are missing or invalid)",
	)
}
func (p *premiumReplayClusterProvider) formatServerlessHostPort(host string, port int64) (string, bool) {
	if host == "" || port <= 0 {
		return "", false
	}
	return fmt.Sprintf("%s:%d", host, port), true
}