package impl

import (
	"context"
	"errors"
	"strconv"

	"github.com/pelletier/go-toml/v2"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

type clusterModelRegionalImpl struct {
	instanceTypeModel      ticdc_spec.CDCInstanceTypeModel
	ticdcClusterCli        regionalapi.TiCDCClusterClient
	transProvider          db.TransProvider
	ticdcClusterRepo       repo.TiCDCClusterRepo
	ticdcCustomStorageRepo repo.TiCDCCustomStorageRepo
	configManager          config_manager.TiCDCClusterConfigManager
	regionCli              grpcli.RegionGrpcClient
}

func NewTiCDCClusterRegionalModel(
	transProvider db.TransProvider,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	ticdcClusterCli regionalapi.TiCDCClusterClient,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	ticdcCustomStorageRepo repo.TiCDCCustomStorageRepo,
	configManager config_manager.TiCDCClusterConfigManager,
	regionCli grpcli.RegionGrpcClient,
) ticdc.ClusterModel {
	return &clusterModelRegionalImpl{
		instanceTypeModel:      instanceTypeModel,
		ticdcClusterCli:        ticdcClusterCli,
		transProvider:          transProvider,
		ticdcClusterRepo:       ticdcClusterRepo,
		ticdcCustomStorageRepo: ticdcCustomStorageRepo,
		configManager:          configManager,
		regionCli:              regionCli,
	}
}

func (m *clusterModelRegionalImpl) Create(
	ctx context.Context, req *domain.CreateTiCDCClusterReq,
) (*domain.CreateTiCDCClusterResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	ticdcClusterEnt, err := m.ticdcClusterRepo.Get(
		ctx, func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.NameEq(req.Name).SharedPoolIDEq(req.SharedPoolID).DeletedAtIsNull()
		})
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("get ticdc cluster entity")
	}
	if ticdcClusterEnt != nil {
		if ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleting {
			return nil, bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc cluster %d is deleting, wait for it deleted", ticdcClusterEnt.ID.Uint64())
		}
		logger.Info("ticdc cluster already exists", zap.Uint64("cluster_id", ticdcClusterEnt.ID.Uint64()))
		return &domain.CreateTiCDCClusterResp{
			ID: ticdcClusterEnt.ID.Uint64(),
		}, nil
	}

	ticdcClusterEnt = m.createReq2Entity(req)
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doCreate(ctx, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.CreateTiCDCClusterResp{
		ID: ticdcClusterEnt.ID.Uint64(),
	}, nil
}

func (m *clusterModelRegionalImpl) doCreate(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}

	if _, err := m.configManager.Create(ctx, &domain.CreateTiCDCClusterConfigReq{
		TiCDCClusterID:     ticdcClusterEnt.ID.Uint64(),
		TiDBClusterVersion: ticdcClusterEnt.Version,
		CloudProvider:      ticdcClusterEnt.CloudProvider,
	}); err != nil {
		return err.Chain("create ticdc cluster configuration")
	}
	configResp, err := m.configManager.Get(ctx, &domain.GetTiCDCClusterConfigReq{
		TiCDCClusterID: ticdcClusterEnt.ID.Uint64(),
	})
	if err != nil {
		return err.Chain("get ticdc cluster configuration")
	}
	instanceTypeResp, err := m.instanceTypeModel.CalCDCInstanceType(ctx, &domain.CalCDCInstanceTypeReq{
		ExpectedSpecification: ticdcClusterEnt.SpecificationCode,
		CloudProvider:         ticdcClusterEnt.CloudProvider,
		Region:                ticdcClusterEnt.Region,
	})
	if err != nil {
		return err.Chain("get ticdc instance type")
	}
	if err := m.createRegionalTiCDCClusterIgnoreExists(
		ctx, ticdcClusterEnt, configResp.Configuration, instanceTypeResp.InstanceType); err != nil {
		return err.Chain("create regional ticdc cluster")
	}
	return nil
}

func (m *clusterModelRegionalImpl) createRegionalTiCDCClusterIgnoreExists(
	ctx context.Context,
	ticdcClusterEnt *entity.TiCDCCluster,
	config *domain.TiCDCConfigData,
	instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	ticdcCluster, err := m.convertRegionalTiCDCCluster(ticdcClusterEnt, config, instanceType)
	if err != nil {
		return err.Chain("convert regional ticdc cluster")
	}
	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.CreateTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		TicdcCluster:   ticdcCluster,
	}
	if _, err := m.ticdcClusterCli.CreateTiCDCCluster(ctx, region, regionalReq); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordExist {
			return nil
		}
		return err
	}
	return nil
}

func (m *clusterModelRegionalImpl) convertRegionalTiCDCCluster(
	ticdcClusterEnt *entity.TiCDCCluster, config *domain.TiCDCConfigData, instanceType *domain.CDCInstanceType,
) (*ticdc_v1.TiCDCCluster, bizerr.BizError) {
	ticdcClusterConfig, err := m.convertRegionalTiCDCClusterConfig(config)
	if err != nil {
		return nil, err
	}
	ticdcCluster := &ticdc_v1.TiCDCCluster{
		Id:           ticdcClusterEnt.Name,
		SharedPoolId: ticdcClusterEnt.SharedPoolID,
		NodeSpec: &ticdc_v1.NodeInfo{
			CpuCore:  instanceType.MachineCPUNum,
			MemoryGi: instanceType.MachineMemoryGi,
			Count:    instanceType.Replicas,
		},
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceType.StorageType,
			SizeGi:        instanceType.MachineStorageGi,
			Iops:          instanceType.Iops,
			ThroughputMbs: instanceType.ThroughputMbs,
		},
		Configuration: ticdcClusterConfig,
	}
	return ticdcCluster, nil
}

func (m *clusterModelRegionalImpl) convertRegionalTiCDCClusterConfig(
	configData *domain.TiCDCConfigData,
) (*ticdc_v1.TiCDCCluster_Configuration, bizerr.BizError) {
	configBytes, err := toml.Marshal(configData.Config)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "marshal config")
	}
	configuration := &ticdc_v1.TiCDCCluster_Configuration{
		Config: string(configBytes),
	}
	if configData.BaseImage != nil {
		configuration.BaseImage = *configData.BaseImage
	}
	if configData.Tag != nil {
		configuration.Tag = *configData.Tag
	}
	if configData.Version != nil {
		configuration.Version = *configData.Version
	}
	return configuration, nil
}

func (m *clusterModelRegionalImpl) createReq2Entity(
	req *domain.CreateTiCDCClusterReq,
) *entity.TiCDCCluster {
	return &entity.TiCDCCluster{
		SharedPoolID:  req.SharedPoolID,
		CloudProvider: req.CloudProvider,

		Name:              req.Name,
		Region:            req.Region,
		SpecificationCode: req.SpecificationCode,
		Status:            entity.TiCDCClusterStatusCreating,
		Version:           req.Version,
	}
}

func (m *clusterModelRegionalImpl) Scale(
	ctx context.Context, req *domain.ScaleTiCDCClusterReq,
) (*domain.ScaleTiCDCClusterResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.ID)
	if err != nil {
		return nil, err.Chain("load ticdc cluster entity")
	}
	instanceTypeResp, err := m.instanceTypeModel.CalCDCInstanceType(ctx, &domain.CalCDCInstanceTypeReq{
		ExpectedSpecification: req.SpecificationCode,
		CloudProvider:         ticdcClusterEnt.CloudProvider,
		Region:                ticdcClusterEnt.Region,
	})
	if err != nil {
		return nil, err.Chain("get ticdc instance type")
	}

	customStorage, err := m.ticdcCustomStorageRepo.LoadByTiCDCClusterID(ctx, req.ID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, bizerr.WrapError(err, "load ticdc custom storage from db failed")
	}
	// overwrite instance type if custom storage exist
	if customStorage != nil {
		logger := log.GetLogger(ctx)
		logger.Info("ticdc custom storage exist, overwrite!",
			zap.Uint64("ticdc cluster id", ticdcClusterEnt.ID.Uint64()),
			zap.String("ticdc cluster name", ticdcClusterEnt.Name),
			zap.Any("instance type", instanceTypeResp.InstanceType),
			zap.Any("custom storage", customStorage))
		instanceTypeResp.InstanceType.Iops = customStorage.Iops
		instanceTypeResp.InstanceType.ThroughputMbs = customStorage.ThroughputMbs
		instanceTypeResp.InstanceType.MachineStorageGi = customStorage.SizeGi
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doScale(ctx, req, ticdcClusterEnt, instanceTypeResp.InstanceType)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.ScaleTiCDCClusterResp{}, nil
}

func (m *clusterModelRegionalImpl) doScale(
	ctx context.Context,
	req *domain.ScaleTiCDCClusterReq,
	ticdcClusterEnt *entity.TiCDCCluster,
	instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	ticdcClusterEnt.SpecificationCode = req.SpecificationCode
	ticdcClusterEnt.Status = entity.TiCDCClusterStatusScaling
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	return m.scaleRegionalTiCDCCluster(ctx, ticdcClusterEnt, instanceType)
}

func (m *clusterModelRegionalImpl) scaleRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster, instanceType *domain.CDCInstanceType,
) bizerr.BizError {
	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.ScaleTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		NodeSpec: &ticdc_v1.NodeInfo{
			CpuCore:  instanceType.MachineCPUNum,
			MemoryGi: instanceType.MachineMemoryGi,
			Count:    instanceType.Replicas,
		},
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceType.StorageType,
			SizeGi:        instanceType.MachineStorageGi,
			Iops:          instanceType.Iops,
			ThroughputMbs: instanceType.ThroughputMbs,
		},
	}
	return m.ticdcClusterCli.ScaleTiCDCCluster(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) Query(
	ctx context.Context, req *domain.QueryTiCDCClusterReq,
) (*domain.QueryTiCDCClusterResp, bizerr.BizError) {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Get(ctx, genFilterByQueryTiCDCClusterReq(req))
	if err != nil {
		return nil, err.Chain("get ticdc cluster entity")
	}
	return &domain.QueryTiCDCClusterResp{
		TiCDCClusterEntity: ticdcClusterEnt,
	}, nil
}

func (m *clusterModelRegionalImpl) Delete(
	ctx context.Context, req *domain.DeleteTiCDCClusterReq,
) (*domain.DeleteTiCDCClusterResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.ID)
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("load ticdc cluster entity")
	}
	if ticdcClusterEnt == nil {
		logger.Info("ticdc cluster is not found", zap.Uint64("cluster_id", req.ID))
		return &domain.DeleteTiCDCClusterResp{}, nil
	}
	if ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleting ||
		ticdcClusterEnt.Status == entity.TiCDCClusterStatusDeleted {
		logger.Info("ticdc cluster is deleted",
			zap.Uint64("cluster_id", req.ID), zap.String("status", string(ticdcClusterEnt.Status)))
		return &domain.DeleteTiCDCClusterResp{}, nil
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.doDelete(ctx, ticdcClusterEnt)
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return &domain.DeleteTiCDCClusterResp{}, nil
}

func (m *clusterModelRegionalImpl) doDelete(ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster) bizerr.BizError {
	ticdcClusterEnt.Status = entity.TiCDCClusterStatusDeleting
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}
	return m.deleteRegionalTiCDCCluster(ctx, ticdcClusterEnt)
}

func (m *clusterModelRegionalImpl) deleteRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster,
) bizerr.BizError {
	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.DeleteTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	return m.ticdcClusterCli.DeleteTiCDCCluster(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) ApplyConfig(
	ctx context.Context, req *domain.ApplyTiCDCClusterConfigReq,
) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	configResp, err := m.configManager.Get(ctx, &domain.GetTiCDCClusterConfigReq{
		TiCDCClusterID: req.TiCDCClusterID,
	})
	if err != nil {
		return err.Chain("get ticdc cluster configuration")
	}

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.updateConfigRegionalTiCDCCluster(ctx, ticdcClusterEnt, configResp.Configuration)
	})
	return convertTransactionError(transErr)
}

func (m *clusterModelRegionalImpl) UpdateStorage(
	ctx context.Context, req *domain.UpdateTiCDCClusterStorageReq,
) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	instanceTypeResp, err := m.instanceTypeModel.CalCDCInstanceType(ctx, &domain.CalCDCInstanceTypeReq{
		ExpectedSpecification: ticdcClusterEnt.SpecificationCode,
		CloudProvider:         ticdcClusterEnt.CloudProvider,
		Region:                ticdcClusterEnt.Region,
	})
	if err != nil {
		return err.Chain("get ticdc instance type")
	}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return m.updateStorage(ctx, req, ticdcClusterEnt, instanceTypeResp)
	})
	return convertTransactionError(transErr)
}

func (m *clusterModelRegionalImpl) updateStorage(ctx context.Context,
	req *domain.UpdateTiCDCClusterStorageReq,
	ticdcClusterEnt *entity.TiCDCCluster,
	instanceTypeResp *domain.CalCDCInstanceTypeResp,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if req.SizeGi == 0 && req.Iops == 0 && req.ThroughputMbs == 0 {
		logger.Info("no storage update", zap.Uint64("cluster_id", req.TiCDCClusterID))
		return nil
	}
	if req.SizeGi <= 0 {
		logger.Info("no storage size update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("size_gi", req.SizeGi))
		req.SizeGi = instanceTypeResp.InstanceType.MachineStorageGi
	}
	if req.Iops <= 0 {
		logger.Info("no storage iops update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("iops", req.Iops))
		req.Iops = instanceTypeResp.InstanceType.Iops
	}
	if req.ThroughputMbs <= 0 {
		logger.Info("no storage throughput update",
			zap.Uint64("cluster_id", req.TiCDCClusterID),
			zap.Uint32("throughput_mbs", req.ThroughputMbs))
		req.ThroughputMbs = instanceTypeResp.InstanceType.ThroughputMbs
	}
	customStorage := &entity.TiCDCCustomStorage{
		TiCDCClusterID: entity.ID(req.TiCDCClusterID),
		SizeGi:         req.SizeGi,
		Iops:           req.Iops,
		ThroughputMbs:  req.ThroughputMbs,
	}

	if err := m.ticdcCustomStorageRepo.Save(ctx, customStorage); err != nil {
		return err.Chain("save ticdc custom storage entity")
	}

	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.UpdateTiCDCClusterStorageRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		Storage: &ticdc_v1.TiCDCCluster_Storage{
			Type:          instanceTypeResp.InstanceType.StorageType,
			SizeGi:        customStorage.SizeGi,
			Iops:          customStorage.Iops,
			ThroughputMbs: customStorage.ThroughputMbs,
		},
	}
	return m.ticdcClusterCli.UpdateTiCDCClusterStorage(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) updateConfigRegionalTiCDCCluster(
	ctx context.Context, ticdcClusterEnt *entity.TiCDCCluster, config *domain.TiCDCConfigData,
) bizerr.BizError {
	ticdcClusterConfig, err := m.convertRegionalTiCDCClusterConfig(config)
	if err != nil {
		return err.Chain("convert regional ticdc cluster config")
	}

	ticdcClusterEnt.Status = entity.TiCDCClusterStatusUpgrading
	if err := m.ticdcClusterRepo.Save(ctx, ticdcClusterEnt); err != nil {
		return err.Chain("save ticdc cluster entity")
	}

	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.UpdateTiCDCClusterConfigRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
		Configuration:  ticdcClusterConfig,
	}
	return m.ticdcClusterCli.UpdateTiCDCClusterConfig(ctx, region, regionalReq)
}

func (m *clusterModelRegionalImpl) List(
	ctx context.Context, req *domain.ListTiCDCClustersReq,
) (*domain.ListTiCDCClustersResp, bizerr.BizError) {
	ticdcClusterEnts, err := m.ticdcClusterRepo.List(
		ctx, func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
			return qs.SharedPoolIDEq(req.SharedPoolID).DeletedAtIsNull()
		})
	if err != nil {
		return nil, err.Chain("list ticdc clusters")
	}
	return &domain.ListTiCDCClustersResp{
		Items: ticdcClusterEnts,
	}, nil
}

func (m *clusterModelRegionalImpl) Restart(ctx context.Context, req *domain.RestartTiCDCClusterReq) bizerr.BizError {
	ticdcClusterEnt, err := m.ticdcClusterRepo.Load(ctx, req.TiCDCClusterID)
	if err != nil {
		return err.Chain("load ticdc cluster entity")
	}
	region := &regionalapi.Region{
		SharedPoolID: ticdcClusterEnt.SharedPoolID,
	}
	regionalReq := &ticdc_v1.RestartTiCDCClusterRequest{
		TicdcClusterId: ticdcClusterEnt.Name,
	}
	if err := m.ticdcClusterCli.RestartTiCDCCluster(ctx, region, regionalReq); err != nil {
		return err.Chain("restart ticdc cluster")
	}
	log.GetLogger(ctx).Info(
		"restart ticdc cluster successfully", zap.Uint64("ticdc_cluster_id", req.TiCDCClusterID))
	return nil
}

func (m *clusterModelRegionalImpl) GetRole(
	ctx context.Context, req *domain.GetTiCDCClusterRoleReq,
) (*domain.GetTiCDCClusterRoleResp, bizerr.BizError) {
	region := &regionalapi.Region{
		SharedPoolID: req.SharedPoolID,
	}
	getRoleReq := &ticdc_v1.GetTiCDCClusterRoleRequest{
		TidbClusterId: strconv.FormatUint(req.TiDBClusterID, 10),
	}
	roleInfo, err := m.ticdcClusterCli.GetTiCDCClusterRole(ctx, region, getRoleReq)
	if err != nil {
		return nil, err.Chain("get ticdc cluster role")
	}
	return &domain.GetTiCDCClusterRoleResp{
		CloudAccountID:   roleInfo.CloudAccountId,
		CloudIAMIdentity: roleInfo.CloudIamIdentity,
		AWSExternalID:    roleInfo.AwsExternalId,
	}, nil
}

func convertTransactionError(err error) bizerr.BizError {
	if err == nil {
		return nil
	}

	var bizErr bizerr.BizError
	if errors.As(err, &bizErr) {
		return bizErr
	}
	return bizerr.New(bizerr.ErrCodeDB, "execute transaction error: %s", err.Error())
}

func genFilterByQueryTiCDCClusterReq(req *domain.QueryTiCDCClusterReq) repo.TiCDCClusterFilterOp {
	return func(qs entity.TiCDCClusterQuerySet) entity.TiCDCClusterQuerySet {
		if req.ID != 0 {
			qs = qs.IDEq(entity.ID(req.ID))
		}
		if req.Name != "" {
			qs = qs.NameEq(req.Name).DeletedAtIsNull()
		}
		return qs
	}
}