package dedicatedtopremiumprecheck

import (
	"context"
	"fmt"
	"net/mail"

	accountsdk "github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/pkgv2/db"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dedicatedglobal"
	"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/precheckcatalog"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow/dag"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	defaultListPageSize = 10
	maxListPageSize     = 1000
)

type Service interface {
	Create(
		ctx context.Context,
		sourceClusterID uint64,
		migrationAccountEmail string,
		targetProjectID *uint64,
		applicant string,
	) (*entity.DedicatedToPremiumPrecheck, error)
	Get(ctx context.Context, id uint64) (*entity.DedicatedToPremiumPrecheck, error)
	List(
		ctx context.Context,
		sourceClusterID uint64,
		pageSize uint32,
		page uint32,
	) ([]*entity.DedicatedToPremiumPrecheck, uint32, error)
}

type service struct {
	transaction  db.Transactioner
	precheckRepo repository.DedicatedToPremiumPrecheckRepo
	clusterRepo  repository.ClusterRepository
	accountCli   account.Client
	dedicatedCli dedicatedglobal.DedicatedGlobal
	workflow     workflow.Workflow
}

func NewService(
	transaction db.Transactioner,
	precheckRepo repository.DedicatedToPremiumPrecheckRepo,
	clusterRepo repository.ClusterRepository,
	accountCli account.Client,
	dedicatedCli dedicatedglobal.DedicatedGlobal,
	wflow workflow.Workflow,
) Service {
	return &service{
		transaction:  transaction,
		precheckRepo: precheckRepo,
		clusterRepo:  clusterRepo,
		accountCli:   accountCli,
		dedicatedCli: dedicatedCli,
		workflow:     wflow,
	}
}

func (s *service) Create(
	ctx context.Context,
	sourceClusterID uint64,
	migrationAccountEmail string,
	targetProjectID *uint64,
	applicant string,
) (*entity.DedicatedToPremiumPrecheck, error) {
	if sourceClusterID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "source_cluster_id is required")
	}
	if migrationAccountEmail != "" {
		address, err := mail.ParseAddress(migrationAccountEmail)
		if err != nil || address.Address != migrationAccountEmail {
			return nil, bizerr.New(bizerr.ErrCodeParams, "migration_account_email is invalid")
		}
	}
	address, err := mail.ParseAddress(applicant)
	if err != nil || address.Address != applicant {
		return nil, bizerr.New(bizerr.ErrCodeParams, "applicant is invalid")
	}
	cluster, err := s.clusterRepo.GetCluster(ctx, sourceClusterID, false)
	if err != nil {
		return nil, err
	}
	if cluster.DeployType != entity.DeployTypeDedicated || cluster.DeletedAt != nil {
		return nil, status.Error(codes.FailedPrecondition, "source cluster must be an active Dedicated cluster")
	}
	if migrationAccountEmail != "" {
		if err := s.validateMigrationAccountOrganization(ctx, migrationAccountEmail, cluster.OrganizationId); err != nil {
			return nil, err
		}
	}
	if targetProjectID != nil {
		if err := s.validateTargetProject(ctx, *targetProjectID, cluster.OrganizationId); err != nil {
			return nil, err
		}
	}
	globalCluster, err := s.dedicatedCli.GetCluster(ctx, &globaldedicatedv1.GetClusterRequest{ClusterId: sourceClusterID})
	if err != nil {
		return nil, status.Error(codes.FailedPrecondition, "source Dedicated cluster is not reachable")
	}
	if globalCluster.GetState() != globaldedicatedv1.Cluster_AVAILABLE {
		return nil, status.Errorf(
			codes.FailedPrecondition,
			"source Dedicated cluster must be AVAILABLE, current state is %s",
			globalCluster.GetState(),
		)
	}

	precheck := &entity.DedicatedToPremiumPrecheck{
		SourceClusterID: sourceClusterID, MigrationAccountEmail: migrationAccountEmail,
		TargetPremiumProjectID: targetProjectID, Applicant: applicant,
		State: entity.DedicatedToPremiumPrecheckStatePending,
		Items: precheckcatalog.NewItems(),
	}
	if err := s.transaction.Transaction(ctx, func(txCtx context.Context) error {
		if err := s.precheckRepo.Create(txCtx, precheck); err != nil {
			return err
		}
		if err := dag.RunDedicatedToPremiumPrecheckDag(txCtx, s.workflow, &dag.DedicatedToPremiumPrecheckDagParams{
			PrecheckID: precheck.ID, OrgID: cluster.OrganizationId, ProjectID: cluster.ProjectId,
		}); err != nil {
			return fmt.Errorf("register dedicated to premium precheck DAG: %w", err)
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return s.precheckRepo.Get(ctx, precheck.ID)
}

func (s *service) validateMigrationAccountOrganization(
	ctx context.Context,
	migrationAccountEmail string,
	sourceOrgID uint64,
) error {
	inSourceOrg, err := s.accountCli.IsUserInOrg(ctx, migrationAccountEmail, sourceOrgID)
	if err != nil {
		return bizerr.WrapGrpcError(err, "failed to verify migration account organization")
	}
	if !inSourceOrg {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"migration account %s does not belong to source cluster organization %d",
			migrationAccountEmail,
			sourceOrgID,
		)
	}
	return nil
}

func (s *service) validateTargetProject(ctx context.Context, projectID, sourceOrgID uint64) error {
	project, err := s.accountCli.GetProject(ctx, projectID)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to get target project")
	}
	if project.GetType() != accountsdk.ProjectType_PROJECT_TYPE_TIDBX {
		return bizerr.New(bizerr.ErrCodeParams, "target project %d must be a TiDB X project", projectID)
	}
	if project.GetTenantId() != sourceOrgID {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"target project %d does not belong to source cluster organization %d",
			projectID,
			sourceOrgID,
		)
	}
	return nil
}

func (s *service) Get(ctx context.Context, id uint64) (*entity.DedicatedToPremiumPrecheck, error) {
	if id == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "precheck id is required")
	}
	return s.precheckRepo.Get(ctx, id)
}

func (s *service) List(
	ctx context.Context,
	sourceClusterID uint64,
	pageSize uint32,
	page uint32,
) ([]*entity.DedicatedToPremiumPrecheck, uint32, error) {
	normalizedPageSize, err := normalizePageSize(pageSize)
	if err != nil {
		return nil, 0, err
	}
	if page == 0 {
		page = 1
	}
	filter := &repository.DedicatedToPremiumPrecheckFilter{
		SourceClusterID: sourceClusterID,
		Limit:           normalizedPageSize,
		Offset:          int(page-1) * normalizedPageSize,
	}
	total, err := s.precheckRepo.Count(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	if total == 0 {
		return []*entity.DedicatedToPremiumPrecheck{}, 0, nil
	}
	prechecks, err := s.precheckRepo.List(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	return prechecks, total, nil
}

func normalizePageSize(pageSize uint32) (int, error) {
	if pageSize == 0 {
		return defaultListPageSize, nil
	}
	if pageSize > maxListPageSize {
		return 0, bizerr.New(bizerr.ErrCodeParams, "page size is too large")
	}
	return int(pageSize), nil
}