package dmcluster

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
)

type dmClusterRepoImpl struct {
	provider         db.Provider
	dataplaneManager dataplane.Manager
}

func NewDMClusterRepo(provider db.Provider, dataplaneManager dataplane.Manager) repo.DMClusterRepo {
	return &dmClusterRepoImpl{
		provider:         provider,
		dataplaneManager: dataplaneManager,
	}
}

func (r *dmClusterRepoImpl) Save(ctx context.Context, cluster *entity.DMCluster) error {
	if cluster == nil {
		return status.Errorf(codes.InvalidArgument, "DMCluster is nil")
	}

	clusterModel := r.cvtDMClusterToModel(cluster)

	wdb := r.provider.WriteDB(ctx)

	// Check if the DMCluster already exists
	var existingCluster schema.DmCluster
	isNewCluster := false
	if err := wdb.Where("id = ?", clusterModel.ID).First(&existingCluster).Error; err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			return status.Errorf(codes.Internal, "failed to check existing DMCluster: %v", err)
		}
		isNewCluster = true
		// If not found, create a new DMCluster
		// get the max template version with all dm clusters
		var maxTemplateVersion sql.NullInt64
		if err := wdb.Model(&schema.DmCluster{}).Select("MAX(template_version)").Scan(&maxTemplateVersion).Error; err != nil {
			return status.Errorf(codes.Internal, "failed to get max template version: %v", err)
		}
		if maxTemplateVersion.Valid {
			clusterModel.TemplateVersion = maxTemplateVersion.Int64
		} else {
			// If no existing dm cluster, set to 1
			clusterModel.TemplateVersion = 1
		}
	} else {
		// If found, update the existing DMCluster
		clusterModel.TemplateVersion = existingCluster.TemplateVersion
		clusterModel.Backend = existingCluster.Backend
		// Preserve the original created_at value to avoid zero value issues
		clusterModel.CreatedAt = existingCluster.CreatedAt
	}

	// New DM clusters always use operator-v2. Preserve the backend value of
	// existing rows for historical data compatibility; runtime dispatch is v2.
	if isNewCluster {
		clusterModel.Backend = entity.DMBackendOperatorV2
	}

	if err := wdb.Model(&schema.DmCluster{}).Where("id = ?", clusterModel.ID).Save(clusterModel).Error; err != nil {
		return status.Errorf(codes.Internal, "failed to save DMCluster: %v", err)
	}

	return nil
}

func (r *dmClusterRepoImpl) Load(ctx context.Context, id string, deleted bool) (*entity.DMCluster, error) {
	if id == "" {
		return nil, status.Errorf(codes.InvalidArgument, "dm cluster id cannot be empty")
	}

	rdb := r.provider.ReadDB(ctx)

	var clusterModel schema.DmCluster
	if deleted {
		rdb = rdb.Unscoped() // Load deleted records
	}

	if err := rdb.Where("id = ?", id).First(&clusterModel).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Errorf(codes.NotFound, "DMCluster with id %s not found", id)
		}
		return nil, status.Errorf(codes.Internal, "failed to load DMCluster: %v", err)
	}

	en, err := r.cvtDMClusterToEn(ctx, &clusterModel)
	if err != nil {
		return nil, err
	}
	return en, nil
}

func (r *dmClusterRepoImpl) List(ctx context.Context, filter *repo.DMClusterListFilter) ([]*entity.DMCluster, error) {
	rdb := r.provider.ReadDB(ctx)

	var clusters []*schema.DmCluster

	if filter.Deleted {
		rdb = rdb.Unscoped() // Load deleted records
	}

	if len(filter.Ids) > 0 {
		rdb = rdb.Where("id IN ?", filter.Ids)
	}

	if len(filter.SharedPoolIds) > 0 {
		rdb = rdb.Where("shared_pool_id IN ?", filter.SharedPoolIds)
	}

	if err := rdb.Find(&clusters).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Errorf(codes.NotFound, "no DMClusters found")
		}
		return nil, status.Errorf(codes.Internal, "failed to list DMClusters: %v", err)
	}

	var result []*entity.DMCluster
	for _, clusterModel := range clusters {
		en, err := r.cvtDMClusterToEn(ctx, clusterModel)
		if err != nil {
			tilog.Warn(ctx, "failed to convert DMCluster to entity", zap.String("id", clusterModel.ID), zap.Error(err))
			continue
		}
		result = append(result, en)
	}

	return result, nil
}

func (r *dmClusterRepoImpl) GetBackend(ctx context.Context, id string, deleted bool) (string, error) {
	if id == "" {
		return "", status.Errorf(codes.InvalidArgument, "dm cluster id cannot be empty")
	}

	rdb := r.provider.ReadDB(ctx)
	if deleted {
		rdb = rdb.Unscoped()
	}

	var clusterModel schema.DmCluster
	if err := rdb.Select("backend").Where("id = ?", id).First(&clusterModel).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return "", status.Errorf(codes.NotFound, "DMCluster with id %s not found", id)
		}
		return "", status.Errorf(codes.Internal, "failed to load DMCluster backend: %v", err)
	}

	return entity.DMBackendOperatorV2, nil
}

func (r *dmClusterRepoImpl) Delete(ctx context.Context, id string) error {
	if id == "" {
		return status.Errorf(codes.InvalidArgument, "DMCluster id cannot be empty")
	}

	wdb := r.provider.WriteDB(ctx)

	if err := wdb.Transaction(func(tx *gorm.DB) error {
		if err := tx.Where("id = ?", id).Delete(&schema.DmCluster{}).Error; err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return status.Errorf(codes.Internal, "failed to delete DMCluster: %v", err)
			}
		}
		return nil
	}); err != nil {
		return err
	}

	return nil
}

func (r *dmClusterRepoImpl) cvtDMClusterToModel(cluster *entity.DMCluster) *schema.DmCluster {
	return &schema.DmCluster{
		ID:            cluster.GetId(),
		SharedPoolID:  cluster.GetSharedPoolId(),
		DataplaneName: cluster.GetDataplaneName(),
	}
}

func (r *dmClusterRepoImpl) cvtDMClusterToEn(ctx context.Context, clusterModel *schema.DmCluster) (*entity.DMCluster, error) {
	if clusterModel == nil {
		return nil, errors.New("DMCluster model cannot be nil")
	}

	// Create a basic DMCluster entity from database
	cluster := &entity.DMCluster{
		Id:            clusterModel.ID,
		SharedPoolId:  clusterModel.SharedPoolID,
		DataplaneName: clusterModel.DataplaneName,
		Backend:       entity.DMBackendOperatorV2Proto,
	}

	// If deleted, return without fetching K8s info
	if clusterModel.DeletedAt.Valid {
		cluster.Phase = apiv1.DMCluster_UNAVAILABLE
		return cluster, nil
	}

	// Get remote client to fetch K8s info
	if clusterModel.DataplaneName == "" {
		tilog.Warn(ctx, "DataplaneName is empty, skipping K8s info fetch", zap.String("id", clusterModel.ID))
		cluster.Phase = apiv1.DMCluster_UNAVAILABLE
		return cluster, nil
	}

	remoteCli, err := r.dataplaneManager.GetClient(clusterModel.DataplaneName)
	if err != nil {
		tilog.Error(ctx, "failed to get remote client for DM cluster",
			zap.String("dataplaneName", clusterModel.DataplaneName),
			zap.Error(err))
		return nil, fmt.Errorf("failed to get remote client for DM cluster: %w", err)
	}

	// Get DM cluster info from K8s.
	dmClusterInfo, err := tioperator.GetDMClusterV2(ctx, remoteCli, cluster)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Warn(ctx, "DM cluster not found in K8s, maybe not ready",
				zap.String("id", clusterModel.ID),
				zap.Error(err))
			cluster.Phase = apiv1.DMCluster_CREATING
			return cluster, nil
		}
		cluster.Phase = apiv1.DMCluster_UNAVAILABLE
		return nil, fmt.Errorf("failed to get DM cluster from K8s: %w", err)
	}

	cluster.MasterNodeQuota = dmClusterInfo.MasterNodeQuota
	cluster.WorkerNodeQuota = dmClusterInfo.WorkerNodeQuota
	cluster.Phase = dmClusterInfo.Phase

	return cluster, nil
}