package tidbcluster

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

	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"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/pkgv2/db"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type clusterRepoImp struct {
	provider            db.Provider
	clusterConfigClient clusterconfig.Client
	dataplaneManager    dataplane.Manager
}

func NewClusterRepo(provider db.Provider, clusterConfigClient clusterconfig.Client, dataplaneManager dataplane.Manager) repo.TiDBClusterRepo {
	return &clusterRepoImp{
		provider:            provider,
		clusterConfigClient: clusterConfigClient,
		dataplaneManager:    dataplaneManager,
	}
}

func (r *clusterRepoImp) Save(ctx context.Context, cluster *entity.TiDBCluster) error {
	if cluster == nil {
		return status.Errorf(codes.InvalidArgument, "cluster cannot be nil")
	}

	wdb := r.provider.WriteDB(ctx)

	clusterModel, err := r.cvtClusterToModel(ctx, cluster)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to convert cluster to model: %v", err)
	}

	// New clusters inherit the current template version. Existing clusters keep
	// their stored template_version because the upsert below never updates it.
	var maxTemplateVersion sql.NullInt64
	if err := wdb.Model(&schema.TidbCluster{}).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 {
		clusterModel.TemplateVersion = 1
	}

	if err := wdb.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "id"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"tenant_id",
			"shared_pool_id",
			"ru_min",
			"ru_max",
			"capacity_mode",
			"audit_log_config",
			"gateway_identity",
			"ru_version",
			"updated_at",
		}),
	}).Create(clusterModel).Error; err != nil {
		return status.Errorf(codes.Internal, "failed to save cluster: %v", err)
	}

	return nil
}

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

	rdb := r.provider.ReadDB(ctx)

	var clusterModel schema.TidbCluster
	if deleted {
		rdb = rdb.Unscoped() // Use Unscoped to include soft-deleted records
	}

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

	clusterEntity, err := r.cvtModelToCluster(ctx, &clusterModel)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed to convert cluster model to entity: %v", err)
	}

	return clusterEntity, nil
}

func (r *clusterRepoImp) List(ctx context.Context, filter *repo.TiDBClusterListFilter) ([]*entity.TiDBCluster, error) {
	if filter == nil {
		return nil, status.Errorf(codes.InvalidArgument, "filter cannot be nil")
	}

	rdb := r.provider.ReadDB(ctx)

	if len(filter.Ids) == 0 && len(filter.TenantIds) == 0 && len(filter.SharedPoolIds) == 0 {
		return nil, status.Errorf(codes.InvalidArgument, "at least one filter must be provided")
	}

	if filter.Deleted {
		rdb = rdb.Unscoped() // Use Unscoped to include soft-deleted records
	}

	query := rdb.Model(&schema.TidbCluster{})

	if len(filter.Ids) > 0 {
		query = query.Where("id IN ?", filter.Ids)
	}
	if len(filter.TenantIds) > 0 {
		query = query.Where("tenant_id IN ?", filter.TenantIds)
	}
	if len(filter.SharedPoolIds) > 0 {
		query = query.Where("shared_pool_id IN ?", filter.SharedPoolIds)
	}

	var clusterModels []*schema.TidbCluster
	if err := query.Find(&clusterModels).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, status.Errorf(codes.NotFound, "no Clusters found with the provided filters")
		}
		return nil, status.Errorf(codes.Internal, "failed to list Clusters: %v", err)
	}

	var clusterEntities []*entity.TiDBCluster
	for _, clusterModel := range clusterModels {
		clusterEntity, err := r.cvtModelToCluster(ctx, clusterModel)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "failed to convert Cluster model to entity: %v", err)
		}
		clusterEntities = append(clusterEntities, clusterEntity)
	}

	return clusterEntities, nil
}

func (r *clusterRepoImp) Delete(ctx context.Context, id string) error {
	if id == "" {
		return status.Errorf(codes.InvalidArgument, "Cluster 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.TidbCluster{}).Error; err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return status.Errorf(codes.Internal, "failed to delete Cluster: %v", err)
			}
		}
		return nil
	}); err != nil {
		return err
	}

	return nil
}

// getSharedPoolDataplaneName queries only the DataplaneName from SharedPool table by ID.
// This avoids loading the full SharedPool entity which requires multiple K8s calls.
func (r *clusterRepoImp) getSharedPoolDataplaneName(ctx context.Context, sharedPoolId string) (string, error) {
	rdb := r.provider.ReadDB(ctx)

	var sharedPool schema.SharedPool
	if err := rdb.Select("dataplane_name").Where("id = ?", sharedPoolId).First(&sharedPool).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return "", status.Errorf(codes.NotFound, "shared pool with id %s not found", sharedPoolId)
		}
		return "", status.Errorf(codes.Internal, "failed to get shared pool dataplane name: %v", err)
	}

	return sharedPool.DataplaneName, nil
}