package scheduler

import (
	"context"
	"fmt"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_impl "github.com/tidbcloud/dataflow-service/app/models/ticdc/impl"
	"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/db"
)

type dedicatedBizChangefeedScheduler struct {
	transProvider          db.TransProvider
	bizChangefeedRepo      repo.BizChangefeedRepo
	cdcClusterModelFactory *ticdc_impl.TiCDCClusterModelFactory

	coreCli             grpcli.CoreGrpcClient
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient
}

func newDedicatedBizChangefeedScheduler(
	transProvider db.TransProvider,
	bizChangefeedRepo repo.BizChangefeedRepo,
	cdcClusterModelFactory *ticdc_impl.TiCDCClusterModelFactory,
	coreCli grpcli.CoreGrpcClient,
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient,
) biz_changefeed.BizChangefeedScheduler {
	return &dedicatedBizChangefeedScheduler{
		transProvider:          transProvider,
		bizChangefeedRepo:      bizChangefeedRepo,
		cdcClusterModelFactory: cdcClusterModelFactory,
		coreCli:                coreCli,
		tidbClusterInfraCli:    tidbClusterInfraCli,
	}
}

func (m *dedicatedBizChangefeedScheduler) 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
	}

	cluster, bizErr := m.coreCli.GetAvailableClusterByID(ctx, bizChangefeedEntity.ClusterID.Uint64())
	if bizErr != nil {
		return 0, bizErr.Chain("get cluster")
	}
	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")
	}

	err = m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		createdCDCClusterID, err := m.createTiCDCCluster(ctx, bizChangefeedEntity, cluster, cloudAccountID)
		if err != nil {
			return fmt.Errorf("create ticdc cluster error: %w", err)
		}
		bizChangefeedEntity.TiCDCClusterID = entity.ID(createdCDCClusterID)
		if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		ticdcClusterID = createdCDCClusterID
		return nil
	})
	if err != nil {
		return 0, err
	}
	return ticdcClusterID, nil
}

func (m *dedicatedBizChangefeedScheduler) createTiCDCCluster(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed, cluster *core.Cluster, cloudAccountID string,
) (uint64, error) {
	req := &domain.CreateTiCDCClusterReq{
		InfraProviderType: cluster.InfraProviderType,
		CloudAccountID:    cloudAccountID,
		TenantID:          cluster.TenantId,
		ProjectID:         cluster.ProjectId,
		ClusterID:         cluster.Id,
		Region:            cluster.RegionName,
		Version:           cluster.Version,
		Name:              genDedicatedTiCDCClusterName(bizChangefeedEntity.ID.Uint64()),
		BizType:           domain.TiCDCClusterBizTypeChangefeed,
		DeployType:        entity.TiCDCClusterDeployTypeDedicated,
		SpecificationCode: bizChangefeedEntity.SpecificationCode,
	}
	resp, err := m.cdcClusterModelFactory.GetModel().Create(ctx, req)
	if err != nil {
		return 0, err
	}
	return resp.ID, nil
}

func (m *dedicatedBizChangefeedScheduler) ScaleChangefeedResource(
	ctx context.Context, bizChangefeedID uint64,
) error {
	bizChangefeedEntity, cdcClusterEntity, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return err
	}
	if cdcClusterEntity.SpecificationCode == bizChangefeedEntity.SpecificationCode {
		return nil
	}

	_, bizErr := m.cdcClusterModelFactory.GetModel().
		Scale(ctx, &domain.ScaleTiCDCClusterReq{
			ID:                bizChangefeedEntity.TiCDCClusterID.Uint64(),
			SpecificationCode: bizChangefeedEntity.SpecificationCode,
		})
	if bizErr != nil {
		return bizErr.Chain("scale ticdc cluster")
	}
	return nil
}

func (m *dedicatedBizChangefeedScheduler) DeleteChangefeedResource(
	ctx context.Context, bizChangefeedID uint64,
) error {
	bizChangefeedEntity, cdcClusterEntity, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return err
	}
	if cdcClusterEntity.Status == entity.TiCDCClusterStatusDeleting ||
		cdcClusterEntity.Status == entity.TiCDCClusterStatusDeleted {
		return nil
	}

	_, bizErr := m.cdcClusterModelFactory.GetModel().
		Delete(ctx, &domain.DeleteTiCDCClusterReq{
			ID: bizChangefeedEntity.TiCDCClusterID.Uint64(),
		})
	if bizErr != nil {
		return bizErr.Chain("delete ticdc cluster")
	}
	return nil
}

func (m *dedicatedBizChangefeedScheduler) IsChangefeedResourceReady(
	ctx context.Context, bizChangefeedID uint64,
) (bool, error) {
	_, cdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return false, err
	}
	return cdcClusterEnt.Status == entity.TiCDCClusterStatusRunning, nil
}

func (m *dedicatedBizChangefeedScheduler) IsChangefeedResourceReleased(
	ctx context.Context, bizChangefeedID uint64,
) (bool, error) {
	_, cdcClusterEnt, err := m.queryBizChangefeedAndTiCDCClusterEntities(ctx, bizChangefeedID)
	if err != nil {
		return false, err
	}
	return cdcClusterEnt.Status == entity.TiCDCClusterStatusDeleted, nil
}

func (m *dedicatedBizChangefeedScheduler) 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.cdcClusterModelFactory.GetModel().
		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 genDedicatedTiCDCClusterName(bizChangefeedID uint64) string {
	return fmt.Sprintf("changefeed-%d", bizChangefeedID)
}