package action

import (
	"context"
	"fmt"
	"time"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"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"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	appsv1 "k8s.io/api/apps/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	WaitCassandraProxyReadyActionName      = "wait-cassandra-proxy-ready"
	defaultCassandraProxyReadyPollInterval = 10 * time.Second
	defaultCassandraProxyReadyTimeout      = 5 * time.Minute
)

type WaitCassandraProxyReadyAction struct {
	dataplaneManager   dataplane.Manager
	cassandraStateRepo repo.CassandraProxyStateRepo
	waitInterval       time.Duration
	timeout            time.Duration
}

func NewWaitCassandraProxyReadyAction(
	dataplaneManager dataplane.Manager,
	cassandraStateRepo repo.CassandraProxyStateRepo,
) actions.Action {
	return &WaitCassandraProxyReadyAction{
		dataplaneManager:   dataplaneManager,
		cassandraStateRepo: cassandraStateRepo,
		waitInterval:       defaultCassandraProxyReadyPollInterval,
		timeout:            defaultCassandraProxyReadyTimeout,
	}
}

func (a *WaitCassandraProxyReadyAction) Name() string {
	return WaitCassandraProxyReadyActionName
}

func (a *WaitCassandraProxyReadyAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "wait cassandra proxy ready action")

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

	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return fmt.Errorf("failed to get remote client: %w", err)
	}

	deadline := time.Now().Add(a.timeout)
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != apiv1.ComponentType_CASSANDRA_PROXY {
			continue
		}
		if !workflow.CheckIsTargetComponent(ctx, comp, args) {
			continue
		}
		if comp.GetPhase() == apiv1.Component_DELETING {
			continue
		}

		hostID := comp.GetLabels()[cassandraProxyHostIdLabel]
		if hostID == "" {
			return fmt.Errorf("cassandra-proxy component %s missing label %q", comp.GetId(), cassandraProxyHostIdLabel)
		}

		if err := a.waitComponentReady(ctx, remoteCli, &sharedPool, comp, hostID, deadline); err != nil {
			if markErr := a.cassandraStateRepo.UpdatePhase(ctx, comp.GetId(), hostID, apiv1.Component_UNAVAILABLE); markErr != nil {
				tilog.Warn(ctx, "failed to mark cassandra-proxy state unavailable after wait failure",
					zap.String("componentId", comp.GetId()),
					zap.String("hostId", hostID),
					zap.Error(markErr))
			}
			return err
		}
		if err := a.cassandraStateRepo.UpdatePhase(ctx, comp.GetId(), hostID, apiv1.Component_AVAILABLE); err != nil {
			return fmt.Errorf("failed to mark cassandra-proxy state available: %w", err)
		}
	}

	return nil
}

func (a *WaitCassandraProxyReadyAction) waitComponentReady(
	ctx context.Context,
	remoteCli client.Client,
	sharedPool *entity.SharedPool,
	comp *entity.Component,
	hostID string,
	deadline time.Time,
) error {
	key := client.ObjectKey{
		Name:      entity.GetCassandraProxyResourceName(comp.GetId()),
		Namespace: entity.GetDataPlaneNamespace(sharedPool.GetId()),
	}
	for {
		deployment := &appsv1.Deployment{}
		err := remoteCli.Get(ctx, key, deployment)
		if err == nil {
			if isCassandraProxyDeploymentReady(deployment) {
				return nil
			}
		} else if !apierrors.IsNotFound(err) {
			return fmt.Errorf("failed to get cassandra-proxy deployment %s/%s for host %s: %w", key.Namespace, key.Name, hostID, err)
		}

		if time.Now().After(deadline) {
			return fmt.Errorf("timed out waiting for cassandra-proxy deployment %s/%s to be ready for host %s", key.Namespace, key.Name, hostID)
		}

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(a.waitInterval):
		}
	}
}

func isCassandraProxyDeploymentReady(deployment *appsv1.Deployment) bool {
	desired := int32(1)
	if deployment.Spec.Replicas != nil {
		desired = *deployment.Spec.Replicas
	}
	return deployment.Status.ObservedGeneration >= deployment.Generation &&
		deployment.Status.UpdatedReplicas >= desired &&
		deployment.Status.AvailableReplicas >= desired &&
		deployment.Status.ReadyReplicas >= desired
}