package server

import (
	"context"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type CreateTiKVCapacityReservationInput struct {
	*apiv1.CreateTiKVCapacityReservationRequest
}

type GetTiKVCapacityReservationInput struct {
	*apiv1.GetTiKVCapacityReservationRequest
}

type DeleteTiKVCapacityReservationInput struct {
	*apiv1.DeleteTiKVCapacityReservationRequest
}

func (s *serverImpl) CreateTiKVCapacityReservation(ctx context.Context, input *CreateTiKVCapacityReservationInput) (*entity.TiKVCapacityReservation, error) {
	if err := s.guardSharedPoolMutationByID(ctx, input.GetSharedPoolId(), sharedPoolMutationBlockedWhilePaused); err != nil {
		return nil, err
	}
	decision, bizType, err := s.getScaleDecisionByReservation(ctx, input.GetSharedPoolId(), input.GetBizType(), input.GetReservationId(), true)
	if err != nil && status.Code(err) != codes.NotFound {
		return nil, err
	}
	if decision != nil {
		if decision.DeletedAt.Valid {
			return nil, status.Errorf(codes.AlreadyExists, "reservation %s/%s/%s already exists and is released, please use a new reservation_id", input.GetSharedPoolId(), input.GetBizType().String(), input.GetReservationId())
		}
		return s.refreshAndBuildTiKVCapacityReservation(ctx, input.GetSharedPoolId(), decision)
	}
	if input.GetBizType() == apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_AUTO_SCALER {
		activeDecision, err := s.getActiveAutoScalerDecision(ctx, input.GetSharedPoolId())
		if err != nil {
			return nil, err
		}
		if activeDecision != nil {
			return s.refreshAndBuildTiKVCapacityReservation(ctx, input.GetSharedPoolId(), activeDecision)
		}
	}

	err = s.scaleManager.Scale(ctx, &braction.ScaleInput{
		SharedPoolID:     input.GetSharedPoolId(),
		BizType:          bizType,
		BizID:            input.GetReservationId(),
		FullfilDataBytes: input.GetRequiredDataBytes(),
		PostScaleWriteDB: func(context.Context, uint) error { return nil },
	})
	if err != nil {
		if status.Code(err) == codes.AlreadyExists {
			// Another request may have created the same reservation after the initial existence check.
			// In that case treat this create as idempotent and return the created reservation.
			decision, _, lookupErr := s.getScaleDecisionByReservation(ctx, input.GetSharedPoolId(), input.GetBizType(), input.GetReservationId(), false)
			if lookupErr == nil && decision != nil {
				return s.refreshAndBuildTiKVCapacityReservation(ctx, input.GetSharedPoolId(), decision)
			}
		}
		return nil, fmt.Errorf("failed to create tikv capacity reservation: %w", err)
	}

	decision, _, err = s.getScaleDecisionByReservation(ctx, input.GetSharedPoolId(), input.GetBizType(), input.GetReservationId(), false)
	if err != nil {
		return nil, fmt.Errorf("failed to load created tikv capacity reservation: %w", err)
	}
	return s.refreshAndBuildTiKVCapacityReservation(ctx, input.GetSharedPoolId(), decision)
}

func (s *serverImpl) getActiveAutoScalerDecision(ctx context.Context, sharedPoolID string) (*schema.ScaleTikvDecision, error) {
	decisions, err := s.scaleTikvDecisionRepo.List(ctx, &repo.ScaleTikvDecisionFilter{
		SharedPoolID:   sharedPoolID,
		BizType:        schema.BizTypeAutoScaler,
		IncludeDeleted: false,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to list active auto scaler reservations: %w", err)
	}

	var active *schema.ScaleTikvDecision
	for _, decision := range decisions {
		if decision == nil || !decision.NeedScale || decision.Status == schema.ScaleStatusReachExpected {
			continue
		}
		if active == nil || decision.ID > active.ID {
			active = decision
		}
	}
	return active, nil
}

func (s *serverImpl) GetTiKVCapacityReservation(ctx context.Context, input *GetTiKVCapacityReservationInput) (*entity.TiKVCapacityReservation, error) {
	decision, _, err := s.getScaleDecisionByReservation(ctx, input.GetSharedPoolId(), input.GetBizType(), input.GetReservationId(), true)
	if err != nil {
		return nil, err
	}
	return s.refreshAndBuildTiKVCapacityReservation(ctx, input.GetSharedPoolId(), decision)
}

func (s *serverImpl) DeleteTiKVCapacityReservation(ctx context.Context, input *DeleteTiKVCapacityReservationInput) error {
	if err := s.guardSharedPoolMutationByID(ctx, input.GetSharedPoolId(), sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}
	decision, _, err := s.getScaleDecisionByReservation(ctx, input.GetSharedPoolId(), input.GetBizType(), input.GetReservationId(), true)
	if err != nil {
		return err
	}
	if decision.DeletedAt.Valid {
		tilog.Info(ctx, "tikv capacity reservation is already released",
			zap.String("sharedPoolId", input.GetSharedPoolId()),
			zap.String("reservationId", input.GetReservationId()),
			zap.String("bizType", input.GetBizType().String()))
		return nil
	}
	if err := s.scaleTikvDecisionRepo.Delete(ctx, decision.ID); err != nil {
		return fmt.Errorf("failed to delete tikv capacity reservation: %w", err)
	}
	return nil
}

func (s *serverImpl) getScaleDecisionByReservation(
	ctx context.Context,
	sharedPoolID string,
	bizType apiv1.TiKVCapacityReservationBizType,
	reservationID string,
	includeDeleted bool,
) (*schema.ScaleTikvDecision, string, error) {
	scaleBizType, err := toScaleDecisionBizType(bizType)
	if err != nil {
		return nil, "", err
	}

	decisions, err := s.scaleTikvDecisionRepo.List(ctx, &repo.ScaleTikvDecisionFilter{
		SharedPoolID:   sharedPoolID,
		BizType:        scaleBizType,
		BizID:          reservationID,
		IncludeDeleted: includeDeleted,
	})
	if err != nil {
		return nil, "", fmt.Errorf("failed to list tikv capacity reservations: %w", err)
	}
	if len(decisions) == 0 {
		return nil, scaleBizType, status.Errorf(codes.NotFound, "tikv capacity reservation not found")
	}
	if len(decisions) > 1 {
		return nil, scaleBizType, status.Errorf(codes.Internal, "multiple tikv capacity reservations found for shared pool %s biz type %s reservation %s", sharedPoolID, bizType.String(), reservationID)
	}
	return decisions[0], scaleBizType, nil
}

func (s *serverImpl) refreshAndBuildTiKVCapacityReservation(ctx context.Context, sharedPoolID string, decision *schema.ScaleTikvDecision) (*entity.TiKVCapacityReservation, error) {
	if !decision.DeletedAt.Valid && decision.NeedScale && decision.Status != schema.ScaleStatusReachExpected {
		if _, err := s.scaleManager.WaitAndSetScaleFinish(ctx, sharedPoolID, decision.ID); err != nil {
			return nil, fmt.Errorf("failed to refresh tikv capacity reservation status: %w", err)
		}
		reloaded, err := s.scaleTikvDecisionRepo.Load(ctx, decision.ID, true)
		if err != nil {
			return nil, fmt.Errorf("failed to reload tikv capacity reservation: %w", err)
		}
		decision = reloaded
	}

	return buildTiKVCapacityReservation(decision)
}

func buildTiKVCapacityReservation(decision *schema.ScaleTikvDecision) (*entity.TiKVCapacityReservation, error) {
	bizType, err := fromScaleDecisionBizType(decision.BizType)
	if err != nil {
		return nil, err
	}

	reservationStatus := apiv1.TiKVCapacityReservationStatus_TIKV_CAPACITY_RESERVATION_STATUS_PROVISIONING
	switch {
	case decision.DeletedAt.Valid:
		reservationStatus = apiv1.TiKVCapacityReservationStatus_TIKV_CAPACITY_RESERVATION_STATUS_RELEASED
	case decision.Status == schema.ScaleStatusReachExpected:
		reservationStatus = apiv1.TiKVCapacityReservationStatus_TIKV_CAPACITY_RESERVATION_STATUS_READY
	}

	return &entity.TiKVCapacityReservation{
		SharedPoolId:  decision.SharedPoolID,
		BizType:       bizType,
		ReservationId: decision.BizID,
		NeedScale:     decision.NeedScale,
		Status:        reservationStatus,
		From:          buildTiKVSpecUsage(decision.From),
		To:            buildTiKVSpecUsage(decision.To),
	}, nil
}

func buildTiKVSpecUsage(usage schema.TiKVSpecUsage) *entity.TiKVSpecUsage {
	return &entity.TiKVSpecUsage{
		NodeCount:           usage.NodeCount,
		TotalCapacityBytes:  usage.TotalCapacityBytes,
		TotalAvailableBytes: usage.TotalAvailableBytes,
		UsagePercentage:     usage.UsagePercentage,
	}
}

func toScaleDecisionBizType(bizType apiv1.TiKVCapacityReservationBizType) (string, error) {
	switch bizType {
	case apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_IMPORT:
		return schema.BizTypeImport, nil
	case apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_AUTO_SCALER:
		return schema.BizTypeAutoScaler, nil
	default:
		return "", status.Errorf(codes.InvalidArgument, "unsupported tikv capacity reservation biz type: %s", bizType.String())
	}
}

func fromScaleDecisionBizType(bizType string) (apiv1.TiKVCapacityReservationBizType, error) {
	switch bizType {
	case schema.BizTypeImport:
		return apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_IMPORT, nil
	case schema.BizTypeAutoScaler:
		return apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_AUTO_SCALER, nil
	default:
		return apiv1.TiKVCapacityReservationBizType_TIKV_CAPACITY_RESERVATION_BIZ_TYPE_UNSPECIFIED, status.Errorf(codes.Internal, "unknown scale decision biz type: %s", bizType)
	}
}