package repo

import (
	"context"
	"errors"
	"fmt"

	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/db"
	"gorm.io/gorm"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/infra/persistence/model"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/infra/persistence/repo/cvt"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

type clusterRepoImpl struct {
	dbProvider db.Provider
}

func NewClusterRepoImpl(dbProvider db.Provider) repository.ClusterRepo {
	return &clusterRepoImpl{
		dbProvider: dbProvider,
	}
}

func (c *clusterRepoImpl) Save(ctx context.Context, cluster *entity.Cluster) error {
	ctx, span := Tracer.Start(ctx, "clusterRepo.Save")
	defer span.End()

	wdb := c.dbProvider.WriteDB(ctx)

	clusterModel := cvt.ClusterEnToModel(cluster)

	// load region
	var re model.Region
	rdb := c.dbProvider.ReadDB(ctx)
	if err := rdb.Model(&model.Region{}).
		Joins("JOIN providers ON regions.provider_id = providers.id").
		Where("regions.name=?", cluster.Region.Id.String()).
		Where("regions.service_plan=?", cluster.Region.ServicePlan.String()).
		Where("regions.engine_type=?", cluster.Region.EngineType.String()).
		Where("providers.name=?", cluster.CloudProvider.Id.String()).
		Find(&re).Error; err != nil {
		return bizerr.WrapGormError(err, "load region")
	}
	clusterModel.RegionId = uint64(re.ID)
	clusterModel.ProviderId = re.ProviderId

	return saveWithVersioning(wdb, clusterModel)
}

func saveWithVersioning(wdb *gorm.DB, value any) bizerr.BizError {
	dst, ok := value.(model.IdAndVersioning)
	if !ok {
		return bizerr.WrapGormError(wdb.Save(value).Error, "save to db")
	}

	id := dst.GetId()
	versioning := dst.GetVersioning()

	var ver model.Versioned
	if err := wdb.Model(value).Select("versioning").Where("id=?", id).Find(&ver).Error; err != nil {
		return bizerr.WrapGormError(err, "find version")
	}

	if ver.Versioning == 0 { // new record try to create
		return bizerr.WrapGormError(wdb.Create(value).Error, "create to db")
	}

	result := wdb.Model(value).Select("*").Where("versioning=?", versioning).Updates(value)
	if err := result.Error; err != nil {
		return bizerr.WrapGormError(err, "update to db")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeOptimisticLockConflict, "version conflict,current=%d", versioning)
	}
	return nil
}

func (c *clusterRepoImpl) Count(ctx context.Context, parameter *repository.ClusterQueryParameter) (int64, error) {
	ctx, span := Tracer.Start(ctx, "clusterRepo.Count")
	defer span.End()

	var count int64
	tx, err := buildTxByFilter(c.dbProvider.ReadDB(ctx), parameter)
	if err != nil {
		return 0, err
	}
	if err := tx.Model(&model.Cluster{}).
		InnerJoins("Provider").
		InnerJoins("Region").
		Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "count cluster")
	}

	return count, nil
}

func buildTxByFilter(tx *gorm.DB, parameter *repository.ClusterQueryParameter) (*gorm.DB, bizerr.BizError) {
	if parameter.Deleted {
		tx = tx.Unscoped()
	}
	if len(parameter.TenantIds) > 0 {
		tx = tx.Where("clusters.tenant_id in (?)", parameter.TenantIds)
	}
	if len(parameter.ProjectIds) > 0 {
		tx = tx.Where("clusters.project_id in (?)", parameter.ProjectIds)
	}
	if len(parameter.ClusterIds) > 0 {
		tx = tx.Where("clusters.id in (?)", parameter.ClusterIds)
	}
	if len(parameter.EngineTypes) > 0 {
		tx = tx.Where("clusters.engine_type in (?)", parameter.EngineTypes)
	}
	if parameter.InCreatingStatus {
		tx = tx.Where("clusters.creation_finished_at is null")
	}
	if len(parameter.CloudProviderNames) > 0 {
		tx = tx.Where(
			"`Provider`.`name` in (?)",
			lo.Map(
				parameter.CloudProviderNames,
				func(i globalcommonv1.CloudProvider_Name, _ int) string { return i.String() },
			),
		)
	}
	if len(parameter.RegionNames) > 0 {
		if len(parameter.CloudProviderNames) == 0 {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"region names must be used with cloud provider names, to avoid region name conflict",
			)
		}
		tx = tx.Where(
			"Region.name in (?)",
			lo.Map(parameter.RegionNames, func(i globalcommonv1.Region_Name, _ int) string { return i.String() }),
		)
	}

	return tx, nil
}

func (c *clusterRepoImpl) Query(
	ctx context.Context,
	parameter *repository.ClusterQueryParameter,
) ([]*entity.Cluster, error) {
	ctx, span := Tracer.Start(ctx, "clusterRepo.Query")
	defer span.End()

	var clusters []*model.Cluster

	tx := c.dbProvider.ReadDB(ctx).Model(&model.Cluster{}).InnerJoins("Provider").InnerJoins("Region")

	tx, err := buildTxByFilter(tx, parameter)
	if err != nil {
		return nil, err
	}

	if err := tx.Find(&clusters).Error; err != nil {
		return nil, fmt.Errorf("query cluster failed: %w", err)
	}
	if len(clusters) == 0 {
		return nil, nil
	}

	outs := make([]*entity.Cluster, 0, len(clusters))
	for _, each := range clusters {
		out, err := cvt.Model2En(each)
		if err != nil {
			return nil, err
		}
		outs = append(outs, out)
	}

	return outs, nil
}

func (c *clusterRepoImpl) Get(ctx context.Context, id string, needDeleted bool) (*entity.Cluster, error) {
	if id == "" {
		return nil, errors.New("empty id")
	}

	ctx, span := Tracer.Start(ctx, "clusterRepo.Get")
	defer span.End()

	clusters, err := c.Query(ctx, &repository.ClusterQueryParameter{
		ClusterIds: []string{id},
		Deleted:    needDeleted,
	})
	if err != nil {
		return nil, err
	}

	if len(clusters) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found the cluster %s", id)
	}

	return clusters[0], nil
}

func (c *clusterRepoImpl) UpdateRuVersion(ctx context.Context, id string, ruVersion int32) error {
	ctx, span := Tracer.Start(ctx, "clusterRepo.UpdateRuVersion")
	defer span.End()

	wdb := c.dbProvider.WriteDB(ctx)
	result := wdb.Model(&model.Cluster{}).Where("id = ?", id).Update("ru_version", ruVersion)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "update ru_version")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster %s not found", id)
	}
	return nil
}

func (c *clusterRepoImpl) Delete(ctx context.Context, id string) error {
	ctx, span := Tracer.Start(ctx, "clusterRepo.Delete")
	defer span.End()

	wdb := c.dbProvider.WriteDB(ctx)
	return wdb.Model(&model.Cluster{}).
		Where("id=?", id).
		Update("user_status", model.ClusterStatusDeleted).
		Delete(&model.Cluster{}).
		Error
}