package scheduler

import (
	"context"
	"errors"
	"fmt"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow/action/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
)

type sharedBizChangefeedScheduler struct {
	transProvider             db.TransProvider
	bizChangefeedRepo         repo.BizChangefeedRepo
	bizChangefeedResourceRepo repo.TiCDCBizChangefeedResourceRepo
	cdcClusterModel           ticdc.ClusterModel
	instanceTypeModel         ticdc_spec.CDCInstanceTypeModel

	coreCli             grpcli.CoreGrpcClient
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient
}

func newSharedBizChangefeedScheduler(
	transProvider db.TransProvider,
	bizChangefeedRepo repo.BizChangefeedRepo,
	bizChangefeedResourceRepo repo.TiCDCBizChangefeedResourceRepo,
	cdcClusterModel ticdc.ClusterModel,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	coreCli grpcli.CoreGrpcClient,
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient,
) biz_changefeed.BizChangefeedScheduler {
	return &sharedBizChangefeedScheduler{
		transProvider:             transProvider,
		bizChangefeedRepo:         bizChangefeedRepo,
		bizChangefeedResourceRepo: bizChangefeedResourceRepo,
		cdcClusterModel:           cdcClusterModel,
		instanceTypeModel:         instanceTypeModel,
		coreCli:                   coreCli,
		tidbClusterInfraCli:       tidbClusterInfraCli,
	}
}

func (m *sharedBizChangefeedScheduler) ScheduleChangefeed(
	ctx context.Context, bizChangefeedID uint64,
) (ticdcClusterID uint64, err error) {
	bizChangefeedEntity, bizErr := m.bizChangefeedRepo.Load(ctx, bizChangefeedID)
	if bizErr != nil {
		return 0, bizErr.Chain("load biz_changefeed entity")
	}
	if bizChangefeedEntity.TiCDCClusterID != 0 {
		return bizChangefeedEntity.TiCDCClusterID.Uint64(), nil
	}

	ticdcClusterName := genSharedTiCDCClusterName(bizChangefeedEntity.ClusterID.Uint64())
	queryReq := &domain.QueryTiCDCClusterReq{
		Name: ticdcClusterName,
	}
	queryResp, bizErr := m.cdcClusterModel.Query(ctx, queryReq)
	if bizErr != nil {
		if bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return 0, bizErr.Chain("query ticdc cluster")
		}
		return m.createTiCDCCluster(ctx, ticdcClusterName, bizChangefeedEntity)
	}

	ticdcClusterEnt := queryResp.TiCDCClusterEntity
	err = m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.scaleTiCDCClusterForCreateChangefeed(ctx, bizChangefeedEntity, ticdcClusterEnt)
	})
	if err != nil {
		return 0, err
	}
	return ticdcClusterEnt.ID.Uint64(), nil
}

func (m *sharedBizChangefeedScheduler) scaleTiCDCClusterForCreateChangefeed(
	ctx context.Context, bizChangefeedEnt *entity.BizChangefeed, ticdcClusterEnt *entity.TiCDCCluster,
) error {
	bizChangefeedEnt.TiCDCClusterID = ticdcClusterEnt.ID
	if bizErr := m.bizChangefeedRepo.Update(ctx, bizChangefeedEnt); bizErr != nil {
		return bizErr.Chain("update biz changefeed entity")
	}
	return m.scaleOrDeleteTiCDCCluster(
		ctx, bizChangefeedEnt.SpecificationCode, bizChangefeedEnt, ticdcClusterEnt)
}

func (m *sharedBizChangefeedScheduler) createTiCDCCluster(
	ctx context.Context, ticdcClusterName string, bizChangefeedEnt *entity.BizChangefeed,
) (uint64, error) {
	cluster, bizErr := m.coreCli.GetAvailableClusterByID(ctx, bizChangefeedEnt.ClusterID.Uint64())
	if bizErr != nil {
		return 0, bizErr.Chain("get cluster")
	}

	// azure always use 3 layer architecture, do not need to get cloud account id to call SDK
	var cloudAccountID string
	if cluster.CloudProviderName != string(entity.CloudProviderTypeAzure) {
		cloudAccountID, bizErr = m.tidbClusterInfraCli.GetCloudAccountID(ctx, &grpcli.RegionalNetworkReq{
			ProjectID:    cluster.ProjectId,
			ProviderType: cluster.InfraProviderType,
			Region:       cluster.RegionName,
		})
		if bizErr != nil {
			return 0, bizErr.Chain("get cloud account id")
		}
	}

	ticdcClusterSpec, err := m.calTiCDCClusterSpecByChangefeedSpecs(
		ctx,
		domain.CloudProviderType(cluster.CloudProviderName),
		cluster.RegionName,
		[]string{bizChangefeedEnt.SpecificationCode})
	if err != nil {
		return 0, err
	}
	createTiCDCClusterReq := m.buildCreateTiCDCClusterReq(cluster, cloudAccountID, ticdcClusterName, ticdcClusterSpec)
	var ticdcClusterID uint64
	err = m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		id, err := m.doCreateTiCDCCluster(ctx, createTiCDCClusterReq, bizChangefeedEnt)
		if err != nil {
			return err
		}
		ticdcClusterID = id
		return nil
	})
	if err != nil {
		return 0, fmt.Errorf("create ticdc cluster error: %w", err)
	}
	return ticdcClusterID, nil
}

func (m *sharedBizChangefeedScheduler) doCreateTiCDCCluster(
	ctx context.Context, createTiCDCClusterReq *domain.CreateTiCDCClusterReq, bizChangefeedEnt *entity.BizChangefeed,
) (uint64, error) {
	createTiCDCClusterResp, bizErr := m.cdcClusterModel.Create(ctx, createTiCDCClusterReq)
	if bizErr != nil {
		return 0, bizErr.Chain("create ticdc cluster")
	}
	bizChangefeedResource := &entity.TiCDCBizChangefeedResource{
		TiCDCClusterID:        entity.ID(createTiCDCClusterResp.ID),
		BizChangefeedID:       bizChangefeedEnt.ID,
		ExpectedSpecification: bizChangefeedEnt.SpecificationCode,
	}
	if bizErr = m.bizChangefeedResourceRepo.Save(ctx, bizChangefeedResource); bizErr != nil {
		return 0, bizErr.Chain("save biz changefeed resource")
	}
	bizChangefeedEnt.TiCDCClusterID = entity.ID(createTiCDCClusterResp.ID)
	if bizErr = m.bizChangefeedRepo.Update(ctx, bizChangefeedEnt); bizErr != nil {
		return 0, bizErr.Chain("update biz changefeed entity")
	}
	return createTiCDCClusterResp.ID, nil
}

func (m *sharedBizChangefeedScheduler) buildCreateTiCDCClusterReq(
	cluster *core.Cluster, cloudAccountID, ticdcClusterName, specCode string,
) *domain.CreateTiCDCClusterReq {
	return &domain.CreateTiCDCClusterReq{
		InfraProviderType: cluster.InfraProviderType,
		CloudAccountID:    cloudAccountID,
		TenantID:          cluster.TenantId,
		ProjectID:         cluster.ProjectId,
		ClusterID:         cluster.Id,
		Region:            cluster.RegionName,
		Version:           cluster.Version,
		Name:              ticdcClusterName,
		BizType:           domain.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeShared,
		SpecificationCode: specCode,
	}
}

func (m *sharedBizChangefeedScheduler) scaleOrDeleteTiCDCCluster(
	ctx context.Context,
	bizChangefeedSpec string,
	bizChangefeedEnt *entity.BizChangefeed,
	ticdcClusterEnt *entity.TiCDCCluster,
) error {
	return m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doScaleOrDeleteTiCDCCluster(ctx, bizChangefeedSpec, bizChangefeedEnt, ticdcClusterEnt)
	})
}

func (m *sharedBizChangefeedScheduler) doScaleOrDeleteTiCDCCluster(
	ctx context.Context,
	bizChangefeedSpec string,
	bizChangefeedEnt *entity.BizChangefeed,
	ticdcClusterEnt *entity.TiCDCCluster,
) error {
	if bizErr := m.upsertBizChangefeedResource(
		ctx, bizChangefeedSpec, bizChangefeedEnt.ID.Uint64(), ticdcClusterEnt.ID.Uint64()); bizErr != nil {
		return bizErr
	}

	bizChangefeedResources, bizErr := m.bizChangefeedResourceRepo.List(ctx,
		func(qs entity.TiCDCBizChangefeedResourceQuerySet) entity.TiCDCBizChangefeedResourceQuerySet {
			return qs.TiCDCClusterIDEq(ticdcClusterEnt.ID).ExpectedSpecificationNe("")
		})
	if bizErr != nil {
		return bizErr.Chain("list biz changefeed resources")
	}
	if len(bizChangefeedResources) == 0 {
		return m.deleteTiCDCCluster(ctx, ticdcClusterEnt.ID.Uint64())
	} else {
		return m.scaleTiCDCCluster(ctx, bizChangefeedResources, ticdcClusterEnt)
	}
}

func (m *sharedBizChangefeedScheduler) upsertBizChangefeedResource(
	ctx context.Context, bizChangefeedSpec string, bizChangefeedID, ticdcClusterID uint64,
) error {
	resource, bizErr := m.bizChangefeedResourceRepo.Get(ctx,
		func(qs entity.TiCDCBizChangefeedResourceQuerySet) entity.TiCDCBizChangefeedResourceQuerySet {
			return qs.BizChangefeedIDEq(entity.ID(bizChangefeedID)).TiCDCClusterIDEq(entity.ID(ticdcClusterID))
		})
	if bizErr != nil {
		if bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return bizErr.Chain("get biz changefeed resource")
		}
		resource = &entity.TiCDCBizChangefeedResource{
			BizChangefeedID:       entity.ID(bizChangefeedID),
			TiCDCClusterID:        entity.ID(ticdcClusterID),
			ExpectedSpecification: bizChangefeedSpec,
		}
	}
	resource.ExpectedSpecification = bizChangefeedSpec
	if bizErr = m.bizChangefeedResourceRepo.Save(ctx, resource); bizErr != nil {
		return bizErr.Chain("save biz changefeed resource")
	}
	return nil
}

func (m *sharedBizChangefeedScheduler) scaleTiCDCCluster(
	ctx context.Context,
	bizChangefeedResources []entity.TiCDCBizChangefeedResource,
	ticdcClusterEnt *entity.TiCDCCluster,
) error {
	bizChangefeedSpecs := make([]string, 0, len(bizChangefeedResources))
	for i := range bizChangefeedResources {
		bizChangefeedSpecs = append(bizChangefeedSpecs, bizChangefeedResources[i].ExpectedSpecification)
	}
	ticdcClusterSpec, err := m.calTiCDCClusterSpecByChangefeedSpecs(
		ctx,
		domain.InfraProviderType2CloudProviderType(domain.InfraProviderType(ticdcClusterEnt.InfraProviderType)),
		ticdcClusterEnt.Region,
		bizChangefeedSpecs)
	if err != nil {
		return err
	}

	scaleReq := &domain.ScaleTiCDCClusterReq{
		ID:                ticdcClusterEnt.ID.Uint64(),
		SpecificationCode: ticdcClusterSpec,
	}
	if _, bizErr := m.cdcClusterModel.Scale(ctx, scaleReq); bizErr != nil {
		return bizErr.Chain("scale ticdc cluster")
	}
	return nil
}

func (m *sharedBizChangefeedScheduler) deleteTiCDCCluster(ctx context.Context, ticdcClusterID uint64) error {
	deleteReq := &domain.DeleteTiCDCClusterReq{
		ID: ticdcClusterID,
	}
	if _, bizErr := m.cdcClusterModel.Delete(ctx, deleteReq); bizErr != nil {
		return bizErr.Chain("delete ticdc cluster")
	}
	return nil
}

func (m *sharedBizChangefeedScheduler) calTiCDCClusterSpecByChangefeedSpecs(
	ctx context.Context, cloudProvider domain.CloudProviderType, region string, specs []string,
) (string, error) {
	if len(specs) == 0 {
		return "", errors.New("specs can't be empty")
	}

	changefeedInsTypesResp, err := m.instanceTypeModel.ListChangefeedInstanceTypes(ctx)
	if err != nil {
		return "", err.Chain("list changefeed instance types")
	}
	name2ChangefeedInsTypes := make(
		map[string]*entity.ChangefeedInstanceType, len(changefeedInsTypesResp.InstanceTypes))
	for _, insType := range changefeedInsTypesResp.InstanceTypes {
		name2ChangefeedInsTypes[insType.Name] = insType
	}

	var totalRCU uint32
	for _, spec := range specs {
		insType, ok := name2ChangefeedInsTypes[spec]
		if !ok {
			return "", fmt.Errorf("spec code %s is not found", spec)
		}
		totalRCU += insType.RCUNum
	}

	calCDCInsTypeReq := &domain.CalCDCInstanceTypeReq{
		ExpectedRCUNum: totalRCU,
		CloudProvider:  cloudProvider,
		Region:         region,
	}
	cdcInstanceType, err := m.instanceTypeModel.CalCDCInstanceType(ctx, calCDCInsTypeReq)
	if err != nil {
		return "", err.Chain("calculate ticdc cluster instance type")
	}
	return cdcInstanceType.InstanceType.Name, nil
}

func (m *sharedBizChangefeedScheduler) ScaleChangefeedResource(ctx context.Context, bizChangefeedID uint64) error {
	bizChangefeedEnt, ticdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return err
	}
	return m.scaleOrDeleteTiCDCCluster(ctx, bizChangefeedEnt.SpecificationCode, bizChangefeedEnt, ticdcClusterEnt)
}

func (m *sharedBizChangefeedScheduler) DeleteChangefeedResource(ctx context.Context, bizChangefeedID uint64) error {
	bizChangefeedEnt, ticdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return err
	}
	return m.scaleOrDeleteTiCDCCluster(ctx, "", bizChangefeedEnt, ticdcClusterEnt)
}

func (m *sharedBizChangefeedScheduler) IsChangefeedResourceReady(
	ctx context.Context, bizChangefeedID uint64,
) (bool, error) {
	bizChangefeedEnt, ticdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return false, err
	}

	bizChangefeedResource, bizErr := m.bizChangefeedResourceRepo.Get(ctx,
		func(qs entity.TiCDCBizChangefeedResourceQuerySet) entity.TiCDCBizChangefeedResourceQuerySet {
			return qs.TiCDCClusterIDEq(ticdcClusterEnt.ID).BizChangefeedIDEq(bizChangefeedEnt.ID)
		})
	if bizErr != nil {
		return false, bizErr.Chain("get biz changefeed resource")
	}
	return bizChangefeedResource.ExpectedSpecification == bizChangefeedResource.ActualSpecification, nil
}

func (m *sharedBizChangefeedScheduler) IsChangefeedResourceReleased(
	ctx context.Context, bizChangefeedID uint64,
) (bool, error) {
	bizChangefeedEnt, ticdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return false, err
	}

	bizChangefeedResource, bizErr := m.bizChangefeedResourceRepo.Get(ctx,
		func(qs entity.TiCDCBizChangefeedResourceQuerySet) entity.TiCDCBizChangefeedResourceQuerySet {
			return qs.TiCDCClusterIDEq(ticdcClusterEnt.ID).BizChangefeedIDEq(bizChangefeedEnt.ID)
		})
	if bizErr != nil {
		return false, bizErr.Chain("get biz changefeed resource")
	}
	return bizChangefeedResource.ActualSpecification == "", nil
}

func (m *sharedBizChangefeedScheduler) queryBizChangefeedAndTiCDCClusterEntities(
	ctx context.Context, bizChangefeedID uint64,
) (*entity.BizChangefeed, *entity.TiCDCCluster, error) {
	bizChangefeedEnt, bizErr := m.bizChangefeedRepo.Load(ctx, bizChangefeedID)
	if bizErr != nil {
		return nil, nil, bizErr.Chain("load biz_changefeed entity")
	}
	resp, bizErr := m.cdcClusterModel.Query(ctx, &domain.QueryTiCDCClusterReq{
		ID: bizChangefeedEnt.TiCDCClusterID.Uint64(),
	})
	if bizErr != nil {
		return nil, nil, bizErr.Chain("query ticdc cluster")
	}
	return bizChangefeedEnt, resp.TiCDCClusterEntity, nil
}

func genSharedTiCDCClusterName(tidbClusterID uint64) string {
	return fmt.Sprintf("ticdc-%d", tidbClusterID)
}