package action

import (
	"context"
	"fmt"
	"time"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

// PLS-side bounds. AWS PrivateLink endpoint service creation typically
// completes in 30-60s; budget 5 minutes to absorb retries on rare slow
// AWS responses without burying the DAG indefinitely.
var (
	cassandraProxyPLSWaitTimeout  = 5 * time.Minute
	cassandraProxyPLSWaitInterval = 5 * time.Second
)

const ApplyCassandraProxyPLSActionName = "apply-cassandra-proxy-pls"

type ApplyCassandraProxyPLSAction struct {
	sharedPoolRepo  repo.SharedPoolRepo
	tidbClusterRepo repo.TiDBClusterRepo
	plsClient       dplanemgrv1.PrivateLinkConnectionServiceClient
}

func NewApplyCassandraProxyPLSAction(
	sharedPoolRepo repo.SharedPoolRepo,
	tidbClusterRepo repo.TiDBClusterRepo,
	plsClient dplanemgrv1.PrivateLinkConnectionServiceClient,
) actions.Action {
	return &ApplyCassandraProxyPLSAction{
		sharedPoolRepo:  sharedPoolRepo,
		tidbClusterRepo: tidbClusterRepo,
		plsClient:       plsClient,
	}
}

func (a *ApplyCassandraProxyPLSAction) Name() string {
	return ApplyCassandraProxyPLSActionName
}

// Invoke creates (or accepts a pre-existing) PrivateLinkService CR for each
// CASSANDRA_PROXY component named in `ComponentIdsKey`, then waits for it to
// reach ACTIVE.
//
// PLS uses get-or-create + wait-ACTIVE rather than declarative convergence.
// DPM `UpdatePrivateLinkService` only mutates `Spec.Connections`, not
// LoadBalancerName. If an existing PLS points to a different LB, waiting for
// ACTIVE would publish the wrong route as success, so fail fast instead.
func (a *ApplyCassandraProxyPLSAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "apply cassandra-proxy pls action")

	if _, err := args.GetString(workflow.ComponentIdsKey); err != nil {
		return fmt.Errorf("target component id is required, but not found")
	}

	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool id: %w", err)
	}

	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		return fmt.Errorf("failed to load sharedPool: %w", err)
	}

	sharedPoolStr, err := args.GetString(workflow.SharedPoolKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool: %w", err)
	}
	var tempSharedPool entity.SharedPool
	if err := protojson.Unmarshal([]byte(sharedPoolStr), &tempSharedPool); err != nil {
		return fmt.Errorf("failed to unmarshal sharedPool: %w", err)
	}
	sharedPool.ComponentGroups = tempSharedPool.GetComponentGroups()

	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != apiv1.ComponentType_CASSANDRA_PROXY {
			continue
		}
		if !workflow.CheckIsTargetComponent(ctx, comp, args) {
			continue
		}

		tidbCluster, err := a.tidbClusterRepo.Load(ctx, comp.GetId(), false)
		if err != nil {
			return fmt.Errorf("failed to load tidb cluster %s: %w", comp.GetId(), err)
		}

		desired := buildCassandraProxyPLS(comp, tidbCluster, sharedPool)
		dpId := sharedPool.GetDataplaneName()

		current, err := a.plsClient.GetPrivateLinkService(ctx, &dplanemgrv1.GetPrivateLinkServiceRequest{
			DataPlaneId: dpId,
			Id:          desired.GetId(),
		})
		switch {
		case err == nil:
			if err := validateCassandraProxyPLSLoadBalancerName(current, desired); err != nil {
				return err
			}
		case status.Code(err) == codes.NotFound:
			if _, createErr := a.plsClient.CreatePrivateLinkService(ctx, &dplanemgrv1.CreatePrivateLinkServiceRequest{
				DataPlaneId:        dpId,
				Id:                 desired.GetId(),
				PrivateLinkService: desired,
			}); createErr != nil {
				if status.Code(createErr) != codes.AlreadyExists {
					return fmt.Errorf("failed to create cassandra-proxy PLS %s: %w", desired.GetId(), createErr)
				}
				current, getErr := a.plsClient.GetPrivateLinkService(ctx, &dplanemgrv1.GetPrivateLinkServiceRequest{
					DataPlaneId: dpId,
					Id:          desired.GetId(),
				})
				if getErr != nil {
					return fmt.Errorf("failed to get cassandra-proxy PLS %s after create conflict: %w", desired.GetId(), getErr)
				}
				if err := validateCassandraProxyPLSLoadBalancerName(current, desired); err != nil {
					return err
				}
			}
		default:
			return fmt.Errorf("failed to get cassandra-proxy PLS %s: %w", desired.GetId(), err)
		}

		if err := a.waitForActive(ctx, dpId, desired); err != nil {
			return err
		}
	}

	return nil
}

// waitForActive polls until the PLS reports ACTIVE or the bounded budget is
// exhausted. Each successful poll revalidates the LB binding before accepting
// ACTIVE, so an out-of-band mutation cannot publish the wrong route as success.
// The DAG framework retries the whole action if we return error, so a timeout
// here is recoverable - we just want to avoid blocking the caller indefinitely
// on an AWS-side stall.
func (a *ApplyCassandraProxyPLSAction) waitForActive(ctx context.Context, dataplaneId string, desired *dplanemgrv1.PrivateLinkService) error {
	deadline := time.Now().Add(cassandraProxyPLSWaitTimeout)
	ticker := time.NewTicker(cassandraProxyPLSWaitInterval)
	defer ticker.Stop()

	for {
		resp, err := a.plsClient.GetPrivateLinkService(ctx, &dplanemgrv1.GetPrivateLinkServiceRequest{
			DataPlaneId: dataplaneId,
			Id:          desired.GetId(),
		})
		if err == nil {
			if err := validateCassandraProxyPLSLoadBalancerName(resp, desired); err != nil {
				return err
			}
			if resp.GetStatus() == dplanemgrv1.PrivateLinkService_ACTIVE {
				return nil
			}
		}
		if err != nil && status.Code(err) != codes.NotFound {
			return fmt.Errorf("failed to poll cassandra-proxy PLS %s: %w", desired.GetId(), err)
		}
		if time.Now().After(deadline) {
			currentStatus := dplanemgrv1.PrivateLinkService_STATE_UNSPECIFIED
			if resp != nil {
				currentStatus = resp.GetStatus()
			}
			return status.Errorf(codes.FailedPrecondition,
				"cassandra-proxy PLS %s did not reach ACTIVE within %v (last status %s)",
				desired.GetId(), cassandraProxyPLSWaitTimeout, currentStatus.String())
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

// buildCassandraProxyPLS returns the desired PLS CR. The 1:1 LB-PLS pair
// reuses the LB name to keep ops mental model simple.
func buildCassandraProxyPLS(
	comp *entity.Component,
	tidbCluster *entity.TiDBCluster,
	sharedPool *entity.SharedPool,
) *dplanemgrv1.PrivateLinkService {
	lbName := entity.GetCassandraProxyLbName(comp.GetId())
	return &dplanemgrv1.PrivateLinkService{
		Id:               lbName,
		LoadBalancerName: lbName,
		Tags:             entity.BuildCassandraProxyTags(comp, tidbCluster, sharedPool),
	}
}

func validateCassandraProxyPLSLoadBalancerName(current, desired *dplanemgrv1.PrivateLinkService) error {
	if current.GetLoadBalancerName() != desired.GetLoadBalancerName() {
		return status.Errorf(codes.FailedPrecondition,
			"cassandra-proxy PLS %s points to loadbalancer %s, want %s",
			desired.GetId(), current.GetLoadBalancerName(), desired.GetLoadBalancerName())
	}
	return nil
}