package migrationcluster

import (
	"context"
	"fmt"

	"github.com/samber/lo"
	"go.opentelemetry.io/otel"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/dedicated/cluster"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/dedicated/migrationcluster"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/apimiddleware/bizerror"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

var tracer = otel.Tracer("App.Dedicated.MigrationCluster")

func NewServer(
	migrationClusterSvc migrationcluster.Service,
	dedicatedClusterSvc cluster.Service,
	accountCli account.Client,
) tidbmgmtservicev1.MigrationClusterServiceServer {
	return &serverImpl{
		migrationClusterSvc: migrationClusterSvc,
		dedicatedClusterSvc: dedicatedClusterSvc,
		accountCli:          accountCli,
	}
}

type serverImpl struct {
	migrationClusterSvc migrationcluster.Service
	dedicatedClusterSvc cluster.Service
	accountCli          account.Client
}

func (s *serverImpl) CreateMigrationCluster(
	ctx context.Context,
	req *tidbmgmtservicev1.CreateMigrationClusterRequest) (*tidbmgmtservicev1.MigrationCluster, error) {
	ctx, span := tracer.Start(ctx, "CreateMigrationCluster")
	defer span.End()

	migrationCluster := req.MigrationCluster
	if migrationCluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "migration cluster is nil")
	}
	if migrationCluster.SourceClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source_cluster_id is required")
	}
	if err := validateRestoreMode(migrationCluster.GetRestoreMode()); err != nil {
		return nil, err
	}

	// Check cluster display name format
	if !common.CheckDisplayNameFmt(migrationCluster.ClusterDisplayName) {
		err := fmt.Errorf("cluster name not allowed: %s", migrationCluster.ClusterDisplayName)
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid cluster display name")
	}

	clusters, err := s.dedicatedClusterSvc.ListClusters(ctx, &repository.ClusterFilter{
		ClusterIds: []uint64{migrationCluster.SourceClusterId},
		View:       entity.ClusterViewClusterViewBasic,
	}, nil)
	if err != nil {
		return nil, bizerror.NewInternal(err, "failed to get dedicated cluster")
	}
	if len(clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "source cluster not found")
	}
	sourceCluster := clusters[0]
	sourceGlobalCluster, err := sourceCluster.GetDedicatedCluster()
	if err != nil {
		return nil, bizerror.NewInternal(err, "failed to get dedicated cluster")
	}
	if sourceGlobalCluster.CustomerManagedEncryptionKeyId != "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source cluster with CMEK is not supported for creating migration cluster")
	}
	if sourceGlobalCluster.State != globaldedicatedv1.Cluster_AVAILABLE {
		return nil, bizerr.New(bizerr.ErrCodeParams, "only source cluster in AVAILABLE state is supported for creating migration cluster")
	}

	projectID := sourceCluster.ProjectId
	if migrationCluster.TargetProjectId != 0 && migrationCluster.TargetProjectId != projectID {
		// verify target project id
		project, err := s.accountCli.GetProject(ctx, migrationCluster.TargetProjectId)
		if err != nil {
			return nil, bizerror.NewInternal(err, "failed to get target project")
		}
		if project.TenantId != sourceCluster.OrganizationId {
			return nil, bizerr.New(bizerr.ErrCodeParams, "target project must belong to the same organization as source cluster")
		}
		if project.AwsCmekEnabled {
			return nil, bizerr.New(bizerr.ErrCodeParams, "target project with AWS CMEK enabled is not supported for migration cluster")
		}

		projectID = migrationCluster.TargetProjectId
	}

	// gen request meta and append to outgoing meta
	ctx = metadata.AppendToOutgoingMeta(ctx, &metadata.RequestMeta{
		OrgID:             sourceCluster.OrganizationId,
		DefaultProjectID:  projectID,
		ValidProjectList:  []uint64{projectID},
		UserID:            "0",
		IsCheckPermission: lo.ToPtr(true),
	})

	return s.migrationClusterSvc.CreateMigrationCluster(ctx, req, projectID, sourceCluster)
}

func validateRestoreMode(mode tidbmgmtservicev1.MigrationCluster_RestoreMode) error {
	switch mode {
	case tidbmgmtservicev1.MigrationCluster_MODE_UNSPECIFIED,
		tidbmgmtservicev1.MigrationCluster_SNAPSHOT,
		tidbmgmtservicev1.MigrationCluster_PITR:
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "invalid restore_mode: %s", mode.String())
	}
}

func (s *serverImpl) GetMigrationCluster(
	ctx context.Context,
	req *tidbmgmtservicev1.GetMigrationClusterRequest) (*tidbmgmtservicev1.MigrationCluster, error) {
	ctx, span := tracer.Start(ctx, "GetMigrationCluster")
	defer span.End()

	return s.migrationClusterSvc.GetMigrationCluster(ctx, req)
}