package server

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dataplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"k8s.io/client-go/tools/clientcmd"

	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"
)

const (
	premiumClusterServiceBaseURL   = "http://cluster-ng-server.cluster-ng-service"
	premiumDataPlaneManagerBaseURL = "http://dataplane-manager.dataplane-manager"
	premiumClusterServiceTimeout   = 5 * time.Second
)

// premiumReplayProvider implements replayClusterProvider for Premium-tier clusters.
// It holds only the fields extracted from replayContext at construction time, avoiding
// circular references and keeping its dependencies explicit.
type premiumReplayProvider struct {
	premiumTargetEndpoint string
	sourceAuditLog        *apiv1.Replay_AuditLogAddr
	clusterRef            *apiv1.Replay_ClusterRef
	client                *http.Client
	loadOnce              sync.Once
	kubernetes            *cloudv1.Kubernetes
	kubeconfigData        []byte
	loadErr               error
}

func newPremiumReplayProvider(rctx *replayContext, ref *apiv1.Replay_ClusterRef) *premiumReplayProvider {
	return &premiumReplayProvider{
		premiumTargetEndpoint: rctx.premiumTargetEndpoint,
		sourceAuditLog:        rctx.sourceAuditLog,
		clusterRef:            ref,
		client:                &http.Client{Timeout: premiumClusterServiceTimeout},
	}
}

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

func (p *premiumReplayProvider) LoadSourceMaterial(context.Context) (replaySourceMaterial, error) {
	return p.buildSourceMaterial()
}

func (p *premiumReplayProvider) LoadTargetEndpoints(context.Context) ([]string, error) {
	if p.premiumTargetEndpoint == "" {
		return nil, errPremiumReplayMissingTargetEndpoint
	}
	return []string{p.premiumTargetEndpoint}, nil
}

func (p *premiumReplayProvider) ResolveDataplane(ctx context.Context) (replayDataplaneResolution, error) {
	if p.clusterRef == nil {
		return replayDataplaneResolution{}, errReplayDataplaneMissingDataNode
	}

	kubernetes, kubeconfigData, err := p.resolveKubernetes(ctx)
	if err != nil {
		return replayDataplaneResolution{}, err
	}

	return replayDataplaneResolution{
		kubernetes:     *kubernetes,
		kubeconfigData: kubeconfigData,
	}, nil
}

func (p *premiumReplayProvider) buildSourceMaterial() (replaySourceMaterial, error) {
	addr, ok, err := p.explicitPrimaryAuditLogAddr()
	if err != nil {
		return replaySourceMaterial{}, err
	}
	if !ok {
		return replaySourceMaterial{}, errPremiumReplayMissingSourceAuditLog
	}

	return replaySourceMaterial{
		units: []replaySourceTaskUnit{
			{auditLogAddrs: []cloudv1.AuditLogAddr{addr}},
		},
	}, nil
}

func (p *premiumReplayProvider) explicitPrimaryAuditLogAddr() (cloudv1.AuditLogAddr, bool, error) {
	if p.sourceAuditLog == nil {
		return cloudv1.AuditLogAddr{}, false, nil
	}

	switch source := p.sourceAuditLog.Source.(type) {
	case *apiv1.Replay_AuditLogAddr_Aws:
		return p.explicitAWSPrimaryAuditLogAddr(source.Aws)
	case *apiv1.Replay_AuditLogAddr_Aliyun:
		return p.explicitAliyunPrimaryAuditLogAddr(source.Aliyun)
	default:
		return cloudv1.AuditLogAddr{}, false, nil
	}
}

func (p *premiumReplayProvider) explicitAWSPrimaryAuditLogAddr(aws *apiv1.Replay_AWSAuditLog) (cloudv1.AuditLogAddr, bool, error) {
	if aws == nil {
		return cloudv1.AuditLogAddr{}, false, nil
	}
	if err := validateAWSAuditLog(aws); err != nil {
		return cloudv1.AuditLogAddr{}, false, err
	}
	if aws.S3Url == "" {
		return cloudv1.AuditLogAddr{}, false, errPremiumReplayMissingAWSAuditLogS3URL
	}
	return cloudv1.AuditLogAddr{
		AWS: &cloudv1.AWSAuditLog{
			S3URL:      aws.S3Url,
			RoleArn:    aws.RoleArn,
			ExternalID: aws.ExternalId,
		},
	}, true, nil
}

func (p *premiumReplayProvider) explicitAliyunPrimaryAuditLogAddr(aliyun *apiv1.Replay_AliyunAuditLog) (cloudv1.AuditLogAddr, bool, error) {
	if aliyun == nil || aliyun.BucketUrl == "" {
		return cloudv1.AuditLogAddr{}, false, nil
	}
	return cloudv1.AuditLogAddr{
		Aliyun: &cloudv1.AliyunAuditLog{
			BucketURL: aliyun.BucketUrl,
		},
	}, true, nil
}

func (p *premiumReplayProvider) resolveKubernetes(ctx context.Context) (*cloudv1.Kubernetes, []byte, error) {
	p.loadOnce.Do(func() {
		p.kubernetes, p.kubeconfigData, p.loadErr = p.loadKubernetes(ctx)
	})
	if p.loadErr != nil {
		return nil, nil, p.loadErr
	}
	return p.kubernetes, p.kubeconfigData, nil
}

func (p *premiumReplayProvider) loadKubernetes(ctx context.Context) (*cloudv1.Kubernetes, []byte, error) {
	provider, err := p.validateClusterRef()
	if err != nil {
		return nil, nil, err
	}

	dataPlane, dataplaneName, err := p.fetchDataplaneMetadata(ctx)
	if err != nil {
		return nil, nil, err
	}
	return p.buildKubernetes(dataPlane, dataplaneName, provider)
}

func (p *premiumReplayProvider) validateClusterRef() (string, error) {
	if p.clusterRef == nil || p.clusterRef.ClusterId == "" {
		return "", premiumReplayMissingDataNode()
	}
	if err := validatePremiumReplayProvider(p.clusterRef.Provider); err != nil {
		return "", err
	}
	return p.clusterRef.Provider, nil
}

func (p *premiumReplayProvider) fetchDataplaneMetadata(ctx context.Context) (*dataplanemgrv1.DataPlane, string, error) {
	clusterResp := &clusterngv1beta.Cluster{}

	err := p.getProto(ctx, premiumClusterServiceBaseURL,
		fmt.Sprintf("/api/v1beta/clusters/%s", p.clusterRef.ClusterId), clusterResp)
	if err != nil {
		return nil, "", fmt.Errorf("failed to get premium cluster %s: %w", p.clusterRef.ClusterId, err)
	}
	if clusterResp.GetSharedPoolId() == "" {
		return nil, "", fmt.Errorf("premium cluster %s does not contain sharedPoolId", p.clusterRef.ClusterId)
	}

	sharedPoolResp := &clusterngv1beta.SharedPool{}
	err = p.getProto(ctx, premiumClusterServiceBaseURL,
		fmt.Sprintf("/api/v1beta/sharedPools/%s", clusterResp.GetSharedPoolId()), sharedPoolResp)
	if err != nil {
		return nil, "", fmt.Errorf("failed to get shared pool %s: %w", clusterResp.GetSharedPoolId(), err)
	}
	if sharedPoolResp.GetDataplaneName() == "" {
		return nil, "", fmt.Errorf("shared pool %s does not contain dataplaneName", clusterResp.GetSharedPoolId())
	}

	dataPlaneResp := &dataplanemgrv1.DataPlane{}
	err = p.getProto(ctx, premiumDataPlaneManagerBaseURL,
		fmt.Sprintf("/api/v1/dataplanes/%s", sharedPoolResp.GetDataplaneName()), dataPlaneResp)
	if err != nil {
		return nil, "", fmt.Errorf("failed to get dataplane %s: %w", sharedPoolResp.GetDataplaneName(), err)
	}
	if dataPlaneResp.GetKubernetes() == nil {
		return nil, "", fmt.Errorf("premium dataplane %s does not contain kubernetes info", sharedPoolResp.GetDataplaneName())
	}

	return dataPlaneResp, sharedPoolResp.GetDataplaneName(), nil
}

func (p *premiumReplayProvider) getProto(ctx context.Context, baseURL, path string, out proto.Message) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(baseURL, "/")+path, http.NoBody)
	if err != nil {
		return err
	}

	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close() //nolint:errcheck // TODO: handle response body close error

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("request %s %s failed with status %d: %s", http.MethodGet, req.URL.String(), resp.StatusCode, strings.TrimSpace(string(body)))
	}

	unmarshalOpt := protojson.UnmarshalOptions{DiscardUnknown: true}
	if err := unmarshalOpt.Unmarshal(body, out); err != nil {
		return fmt.Errorf("failed to decode response from %s: %w", req.URL.String(), err)
	}
	return nil
}

func (p *premiumReplayProvider) buildKubernetes(
	dataPlane *dataplanemgrv1.DataPlane,
	dataplaneName string,
	provider string,
) (*cloudv1.Kubernetes, []byte, error) {
	if dataPlane == nil || dataPlane.GetKubernetes() == nil {
		return nil, nil, fmt.Errorf("premium replay metadata is incomplete")
	}

	resolved := &cloudv1.Kubernetes{
		Provider: provider,
		Name:     dataPlane.GetKubernetes().GetName(),
		Account:  dataPlane.GetAccountId(),
		Version:  dataPlane.GetKubernetes().GetVersion(),
	}

	if err := validatePremiumReplayProvider(provider); err != nil {
		return nil, nil, err
	}
	if applyDefaults := premiumDataplaneDefaultsFor(provider); applyDefaults != nil {
		applyDefaults(resolved)
	}

	kubeconfigData := decodeKubeconfigPayload(dataPlane.GetKubernetes().GetKubeconfig())
	if len(kubeconfigData) == 0 {
		return nil, nil, fmt.Errorf("premium dataplane %s does not contain kubeconfig", dataplaneName)
	}

	endpointFromKubeconfig, caDataFromKubeconfig, clusterNameFromKubeconfig, err := extractKubernetesMetaFromKubeconfig(kubeconfigData)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to extract kubernetes meta from kubeconfig: %w", err)
	}

	resolved.Name = firstNonEmpty(clusterNameFromKubeconfig, resolved.Name)
	resolved.Endpoint = endpointFromKubeconfig
	resolved.CAData = caDataFromKubeconfig

	if resolved.Name == "" {
		return nil, nil, fmt.Errorf("premium dataplane %s does not contain kubernetes name", dataplaneName)
	}
	if resolved.Account == "" {
		return nil, nil, fmt.Errorf("premium dataplane %s does not contain cloud account id", dataplaneName)
	}
	if resolved.Endpoint == "" {
		return nil, nil, fmt.Errorf("premium dataplane %s kubeconfig does not contain endpoint", dataplaneName)
	}

	return resolved, kubeconfigData, nil
}

func validatePremiumReplayProvider(provider string) error {
	switch provider {
	case consts.ProviderTypeEKS, consts.ProviderTypeACK:
		return nil
	default:
		return unsupportedReplayCloudProvider(provider)
	}
}

func premiumDataplaneDefaultsFor(provider string) func(kubernetes *cloudv1.Kubernetes) {
	switch provider {
	case consts.ProviderTypeEKS:
		return applyEKSPremiumDataplaneDefaults
	case consts.ProviderTypeACK:
		return applyACKPremiumDataplaneDefaults
	default:
		return nil
	}
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func extractKubernetesMetaFromKubeconfig(kubeconfigData []byte) (endpoint string, caData []byte, clusterName string, err error) {
	if len(kubeconfigData) == 0 {
		return "", nil, "", nil
	}

	cfg, err := clientcmd.Load(kubeconfigData)
	if err != nil {
		return "", nil, "", fmt.Errorf("failed to parse kubeconfig: %w", err)
	}

	currentContext := cfg.CurrentContext
	if currentContext == "" {
		for name := range cfg.Contexts {
			currentContext = name
			break
		}
	}

	ctxCfg, ok := cfg.Contexts[currentContext]
	if !ok || ctxCfg == nil {
		return "", nil, "", fmt.Errorf("kubeconfig context %q not found", currentContext)
	}

	clusterName = ctxCfg.Cluster
	clusterCfg, ok := cfg.Clusters[clusterName]
	if !ok || clusterCfg == nil {
		return "", nil, "", fmt.Errorf("kubeconfig cluster %q not found", clusterName)
	}

	return clusterCfg.Server, clusterCfg.CertificateAuthorityData, clusterName, nil
}

func applyACKPremiumDataplaneDefaults(kubernetes *cloudv1.Kubernetes) {
	if kubernetes.Cloud.Ack == nil {
		kubernetes.Cloud.Ack = &cloudv1.CloudAck{}
	}
}

func applyEKSPremiumDataplaneDefaults(kubernetes *cloudv1.Kubernetes) {
	if kubernetes.Cloud.AWS == nil {
		kubernetes.Cloud.AWS = &cloudv1.CloudAWS{}
	}
	ensureAWSSubnetGroups(kubernetes.Cloud.AWS, []cloudv1.AWSSubnetGroupName{
		cloudv1.AWSSubnetGroupNodes,
		cloudv1.AWSSubnetGroupPublicNLB,
		cloudv1.AWSSubnetGroupPrivateNLB,
	})
}

func ensureAWSSubnetGroups(cloudAWS *cloudv1.CloudAWS, required []cloudv1.AWSSubnetGroupName) {
	if cloudAWS == nil {
		return
	}
	existing := make(map[cloudv1.AWSSubnetGroupName]struct{}, len(cloudAWS.SubnetGroups))
	for i := range cloudAWS.SubnetGroups {
		if cloudAWS.SubnetGroups[i].Subnets == nil {
			cloudAWS.SubnetGroups[i].Subnets = []cloudv1.AWSSubnet{}
		}
		existing[cloudAWS.SubnetGroups[i].Name] = struct{}{}
	}
	for _, name := range required {
		if _, ok := existing[name]; ok {
			continue
		}
		cloudAWS.SubnetGroups = append(cloudAWS.SubnetGroups, cloudv1.AWSSubnetGroup{
			Name:    name,
			Subnets: []cloudv1.AWSSubnet{},
		})
	}
}