package cdc

import (
	"context"
	"fmt"
	"time"

	storagev1 "k8s.io/api/storage/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"

	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cert"
	compatk8s "github.com/tidbcloud/infra-provider/pkg/compat/k8s/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/procedure"
	"github.com/tidbcloud/infra-provider/pkg/provider"
)

const defaultSystemVolumeSize = "50Gi"

var _ provider.Cdc = new(Cdc)

type Cdc struct {
	Cli       client.Client
	APIReader client.Reader
	AWSSVC    procedure.AWSService
	Config    *config.CDCConfig
}

func NewCdc(
	cli client.Client, apiReader client.Reader, awsSvc procedure.AWSService, config *config.CDCConfig) (
	provider.Cdc, error) {
	provider, exist := config.Predicate.MatchLabels[cloudv1.ProviderTypeLabel]
	if !exist {
		return nil, fmt.Errorf("empty provider type")
	}
	switch provider {
	case consts.ProviderTypeEKS:
		return NewAwsCdc(&Cdc{
			Cli:       cli,
			APIReader: apiReader,
			AWSSVC:    awsSvc,
			Config:    config,
		}), nil
	case consts.ProviderTypeGKE:
		return NewGcpCdc(&Cdc{
			Cli:       cli,
			APIReader: apiReader,
			AWSSVC:    awsSvc,
			Config:    config,
		}), nil
	default:
		return nil, fmt.Errorf("unknown provider type %s", provider)
	}
}

func (p *Cdc) GetCdcUpstreamClient(ctx context.Context, ticdc *cloudv1.TiCDC) (client.Client, error) {
	network, err := p.cdcLocated(ctx, ticdc)
	if err != nil {
		return nil, fmt.Errorf("failed find cdc located network: %v", err)
	}

	return p.GetClientByNetwork(ctx, p.Cli, network)
}

func (p *Cdc) GetCdcClient(ctx context.Context, network *cloudv1.Network) (client.Client, error) {
	return p.GetClientByNetwork(ctx, p.Cli, network)
}

func (p *Cdc) GetCdcKubernetesStatus(c *cloudv1.Cluster) (*cloudv1.KubernetesStatus, error) {
	return &cloudv1.KubernetesStatus{Namespace: fmt.Sprintf("%s%s", "tidb", c.Name)}, nil
}

func (p *Cdc) CreateCdcNodeGroup(ctx context.Context, ticdc *cloudv1.TiCDC, cls *cloudv1.Cluster, ngSet *nodegroup.Set) (map[string]string, error) {
	return nil, nil
}

func (p *Cdc) DeleteCdcNodeGroup(ctx context.Context, ticdc *cloudv1.TiCDC, ngSet *nodegroup.Set) error {
	return nil
}

func (p *Cdc) ReleaseNodes(ctx context.Context, cr *cloudv1.TiCDC, ngNamePrefix string, labelSet labels.Set) error {
	return nil
}

func (p *Cdc) UpdateNodegroupTopology(ctx context.Context, cr *cloudv1.TiCDC, topology map[string]int) error {
	return nil
}

func (p *Cdc) SetResourceLabels(ctx context.Context, cr *cloudv1.TiCDC) error {
	return nil
}

func (p *Cdc) EnsureAllNodegroupsReady(ctx context.Context, cr *cloudv1.TiCDC) error {
	return nil
}

func (p *Cdc) DeleteNodegroupsByFilter(ctx context.Context, cr *cloudv1.TiCDC, selector labels.Selector) error {
	return nil
}

func (p *Cdc) GetNodegroupsByFilter(ctx context.Context, cr *cloudv1.TiCDC, selector labels.Selector) (map[string]int, error) {
	return nil, nil
}

func (p *Cdc) Credentials(ctx context.Context, network *cloudv1.Network) ([]byte, error) {
	return nil, nil
}

func (p *Cdc) IsDPCompatMode() bool {
	return config.FeatureEnabled(p.Config.FeatureGates, config.EnableDPCompatibilityMode, "", "")
}

func (p *Cdc) BuildStorageClass() *storagev1.StorageClass {
	return nil
}

func (p *Cdc) cdcLocated(ctx context.Context, ticdc *cloudv1.TiCDC) (*cloudv1.Network, error) {
	networkName := ticdc.Spec.NetworkName
	if ticdc.Spec.UpstreamClusterName != "" {
		upstreamCluster := new(cloudv1.Cluster)
		if err := p.APIReader.Get(ctx, types.NamespacedName{
			Name:      ticdc.Spec.UpstreamClusterName,
			Namespace: ticdc.Namespace,
		}, upstreamCluster); err != nil {
			return nil, fmt.Errorf("get upstream cluster: %v", err)
		}

		networkName = upstreamCluster.Spec.NetworkName
	}
	var network cloudv1.Network
	networkNamespaceName := types.NamespacedName{
		Name:      networkName,
		Namespace: ticdc.Namespace,
	}
	err := p.APIReader.Get(ctx, networkNamespaceName, &network)
	if err != nil {
		return nil, fmt.Errorf("get upstream network: %v", err)
	}
	return &network, nil
}
func (p *Cdc) getCdcNetwork(ctx context.Context, ticdc *cloudv1.TiCDC) (*cloudv1.Network, error) {
	networkName := ticdc.Spec.NetworkName
	var network cloudv1.Network
	networkNamespaceName := types.NamespacedName{
		Name:      networkName,
		Namespace: ticdc.Namespace,
	}
	err := p.APIReader.Get(ctx, networkNamespaceName, &network)
	if err != nil {
		return nil, fmt.Errorf("get cdc network: %v", err)
	}
	return &network, nil
}

func tidbNodeGroupName(name string, ticdcUID string) string {
	return "tidbcloud-" + name + ticdcUID
}

// RequireV1CertManager return true because cert-manager version >= v1.6.1. It only support API v1
func (p *Cdc) RequireV1CertManager() bool {
	return true
}

func (p *Cdc) CertificateOptions(cluster *cloudv1.Cluster) (kind string, name string, group string, d time.Duration) {
	if cluster.FeatureEnabled(consts.ClusterFeatureUseSelfSignedCertForTiDBComponents) {
		return cert.TiDBComponentsSelfSignedIssuerKind, cert.TiDBComponentsSelfSignedIssuerName, cert.SelfSignedIssuerGroup, consts.CertificateDuration
	}
	return cert.AWSIssuerKind, cert.PCAIssuerName, consts.CertManagerGroup, consts.CertificateDuration
}

func (p *Cdc) OverrideImage(cluster *cloudv1.Cluster, imageName string) string {
	return sdk.OverrideBaseImage(cluster, imageName)
}

// GetNetworkProxyBaseURL return the network proxy base url
func (p *Cdc) GetNetworkProxyBaseURL(_ context.Context, network *cloudv1.Network) (string, error) {
	status, err := v1alpha1.EKSNetworkStatusFromNetwork(network)
	if err != nil {
		return "", err
	}
	return status.ProxyBaseUrl, nil
}

func (p *Cdc) GetClientByNetwork(ctx context.Context, cli client.Client, network *cloudv1.Network) (client.Client, error) {
	scheme, err := k8s.TiDBOperatorScheme()
	if err != nil {
		return nil, err
	}
	getter, err := compatk8s.NetworkClientGetter(ctx, cli, network, scheme)
	if err != nil {
		return nil, err
	}
	return getter.GetClient(ctx)
}