package task

import (
	"context"
	"fmt"
	"time"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
)

type Endpoint struct {
	*baseTask
}

func NewEndpoint(ctx *Context) deploy.Task {
	return &Endpoint{baseTask: &baseTask{Context: ctx}}
}

func (t *Endpoint) Name() string {
	return "Endpoint"
}

func (t *Endpoint) Create(ctx context.Context) deploy.Result {
	pc := t.getProviderConfigName()
	name := types.NamespacedName{
		Name:      t.SinkEndpoint.Name,
		Namespace: t.SinkEndpoint.Namespace,
	}
	// cloud resource tags, used for cost and billing analysis
	tags := map[string]string{
		consts.TagKeyTenant:      t.getTenantID(),
		consts.TagKeyProject:     t.getProjectID(),
		consts.TagKeyUsedBy:      consts.UsedByCustomerTiDB,
		consts.TagKeyServiceType: consts.ServiceTypeDedicated,
		consts.TagKeyEnvironment: t.Env,
		consts.TagKeyComponent:   string(cloudv1.ComponentTiCDC),
	}
	zones := t.getEndpointSelectedZones()
	endpoint := &cloud.EndpointRequest{
		Region:              t.SinkEndpoint.Labels[cloudv1.RegionLabel],
		CloudEndpointName:   t.SinkEndpoint.Name,
		EndpointServiceName: t.SinkEndpoint.Spec.ServiceName,
		Tags:                tags,
		SelectedZones:       zones,
	}
	objs, err := t.CloudProvider.BuildSinkEndpoint(pc, name, endpoint)
	if err != nil {
		return deploy.Error(err)
	}
	for _, obj := range objs {
		if err := k8s.Apply(ctx, t.Local, obj); err != nil {
			return deploy.Panic(err).WithRequeuAfter(time.Second * 5)
		}
	}
	return deploy.Done()
}

func (t *Endpoint) getEndpointSelectedZones() []string {
	var zones []string
	if len(t.SinkEndpoint.Spec.DNSRecords) > 0 {
		// private link endpoint for Kafka, the user specifies zones
		for _, record := range t.SinkEndpoint.Spec.DNSRecords {
			zones = append(zones, record.AvailabilityZone)
		}
	} else if len(t.SinkEndpoint.Spec.AvailabilityZones) > 0 {
		zones = append(zones, t.SinkEndpoint.Spec.AvailabilityZones...)
	} else {
		// private link endpoint for serverless, we should query zones of private link service by aws api
		// fill zones of created private link endpoint to reduce aws api request
		for zone := range t.SinkEndpoint.Status.ZonesAddresses {
			zones = append(zones, zone)
		}
	}
	return zones
}

func (t *Endpoint) PostCreate(ctx context.Context) deploy.Result {
	detail, err := t.CloudProvider.GetSinkEndpointDetail(ctx, t.Local, types.NamespacedName{Name: t.SinkEndpoint.Name})
	if err != nil {
		if apierrors.IsNotFound(err) {
			// resource not ready, retry
			t.Logger.Info("private endpoint not found, will retry", "name", t.SinkEndpoint.Name)
			return deploy.Error(err).WithRequeuAfter(time.Second * 5)
		}
		t.Logger.Info("get private endpoint error", "error", err)
		return deploy.Error(err).WithRequeuAfter(time.Second * 5)
	}
	t.SinkEndpoint.Status.EndpointID = detail.EndpointID
	t.SinkEndpoint.Status.CloudAccountID = t.AppBox.GetKubernetes().GetAccountId()
	t.SinkEndpoint.Status.Phase = detail.Phase
	t.SinkEndpoint.Status.EndpointAddress = detail.Address
	t.SinkEndpoint.Status.ZonesAddresses = detail.ZoneAddress
	if detail.Phase != cdcv1.EndpointResourceStateEndpointAvailable {
		t.Logger.Info("private endpoint not ready", "name", t.SinkEndpoint.Name)
		return deploy.Error(fmt.Errorf("wait private link endpoint ready")).WithRequeuAfter(time.Second * 5)
	}
	if detail.Address == "" {
		return deploy.Error(fmt.Errorf("wait private link endpoint regional address not empty")).
			WithRequeuAfter(time.Second * 5)
	}

	zoneLen := len(t.SinkEndpoint.Spec.DNSRecords)
	// gcp and azure don't have zonal address for private link endpoint
	if zoneLen == 1 && t.SinkEndpoint.Spec.DNSRecords[0].AvailabilityZone == "all zones" {
		return deploy.Done()
	}
	// this is aws specific logic, aws has zonal address for private link endpoint
	if zoneLen > 0 && zoneLen != len(detail.ZoneAddress) {
		return deploy.Error(
			fmt.Errorf("wait private link zone addresses complete, zone addresses count: %d, expected: %d",
				len(detail.ZoneAddress), zoneLen)).WithRequeuAfter(time.Second * 5)
	}
	return deploy.Done()
}

func (t *Endpoint) Delete(ctx context.Context) deploy.Result {
	name := types.NamespacedName{
		Name:      t.SinkEndpoint.Name,
		Namespace: t.SinkEndpoint.Namespace,
	}
	return deploy.ErrorOrDone(t.CloudProvider.DeleteSinkEndpoint(ctx, t.Local, name))
}