package datasource

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"gorm.io/gorm"

	error2 "github.com/pingcap-inc/clinic/pkg/api/error"
	"github.com/pingcap-inc/clinic/pkg/auth/session"
	"github.com/pingcap-inc/clinic/pkg/datasource/clustersearch"
	cloud2 "github.com/pingcap-inc/clinic/pkg/datasource/internal/cloud"
	"github.com/pingcap-inc/clinic/pkg/datasource/internal/op"
	"github.com/pingcap-inc/clinic/pkg/datasource/view"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/nextgen_client"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/pagination"
)

var clusterDatasources = map[view.DataSourceType]clusterDatasource{
	view.CloudDataSource: &cloudClusterDatasource{},
	view.OPDataSource:    &opClusterDatasource{},
}

const dashboardNumericListTimeout = 45 * time.Second

var ErrInvalidDashboardPagination = errors.New("invalid dashboard pagination")

func normalizeDashboardPagination(page, perPage int) (int, int, error) {
	if page <= 0 {
		page = 1
	}
	pageSize := pagination.GetSize(perPage)
	maxInt := int(^uint(0) >> 1)
	// Reserve one extra row for has_more without overflowing either the
	// start offset multiplication or offset + pageSize + 1.
	if pageSize >= maxInt || page > (maxInt-1)/pageSize {
		return 0, 0, ErrInvalidDashboardPagination
	}
	return page, pageSize, nil
}

func equivalentDashboardDeployTypeV2(legacy string) (string, bool) {
	switch legacy {
	case o11y.DedicatedType:
		return o11y.DeployTypeV2Dedicated, true
	case o11y.ByocType:
		return o11y.DeployTypeV2Byoc, true
	default:
		return "", false
	}
}

// normalizeDashboardDeployTypeFilters moves only exact legacy-to-v2 mappings
// onto the indexed column. Ambiguous legacy values and explicit mixed filters
// keep their existing semantics.
func normalizeDashboardDeployTypeFilters(filter *view.ListOrganizationAndClusterFilter) *view.ListOrganizationAndClusterFilter {
	if filter == nil || len(filter.DeployType) == 0 || len(filter.DeployTypeV2) > 0 {
		return filter
	}

	mapped := make([]string, 0, len(filter.DeployType))
	seen := make(map[string]struct{}, len(filter.DeployType))
	for _, legacy := range filter.DeployType {
		deployTypeV2, ok := equivalentDashboardDeployTypeV2(legacy)
		if !ok {
			return filter
		}
		if _, ok := seen[deployTypeV2]; ok {
			continue
		}
		seen[deployTypeV2] = struct{}{}
		mapped = append(mapped, deployTypeV2)
	}

	normalized := *filter
	normalized.DeployType = nil
	normalized.DeployTypeV2 = mapped
	return &normalized
}

// ListOrganizationAndClusterResult represents either an exact-total page or a has-more page.
type ListOrganizationAndClusterResult struct {
	Items   []*view.DashboardItemExternal
	Total   *int64
	HasMore bool
}

// dashboardClusterExternalSelect is the projection for dashboard list rows (must match view.DashboardItemExternal scan).
const dashboardClusterExternalSelect = `CAST(o.dbaas_org_id AS CHAR(24)) as tenant_id,
		o.name as tenant_name,
		o.plan as tenant_plan,
		c.cluster_id as cluster_id,
		c.name as cluster_name,
		c.version as cluster_version,
		CAST(UNIX_TIMESTAMP(c.created_at)*1000 as SIGNED) as cluster_created_at,
		CAST(UNIX_TIMESTAMP(c.deleted_at)*1000 as SIGNED) as cluster_deleted_at,
		TIMEDIFF(c.creation_finished_at, c.created_at) as cluster_duration,
		c.status as cluster_status,
		c.deploy_type as cluster_deploy_type,
		c.deploy_type_v2 as cluster_deploy_type_v2,
		c.region as cluster_region_name,
		c.provider as cluster_provider_name,
		c.shoot_name as cluster_shoot_name,
		c.data_plane_name as cluster_data_plane_name,
		c.k8s_namespace as cluster_k8s_namespace,
		c.network_name as cluster_network_name,
		c.cluster_type,
		c.project_id,
		c.control_plane_arch,
		c.topology,
		c.parent_id,
		CAST(c.org_id as CHAR(24)) as org_id,
		c.id as cluster_row_id`

func applyDashboardNonQueryFilters(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter) *gorm.DB {
	if len(filter.ClusterIDs) > 0 {
		tx = tx.Where("c.cluster_id IN ?", filter.ClusterIDs)
	}
	if len(filter.ClusterStatus) > 0 {
		tx = tx.Where("c.status IN ?", filter.ClusterStatus)
	}
	if len(filter.Version) > 0 {
		tx = tx.Where("c.version IN ?", filter.Version)
	}
	if len(filter.DeployType) > 0 {
		tx = tx.Where("c.deploy_type IN ?", filter.DeployType)
	}
	if len(filter.DeployTypeV2) > 0 {
		tx = tx.Where("c.deploy_type_v2 IN ?", filter.DeployTypeV2)
	}
	if len(filter.TenantPlan) > 0 {
		tx = tx.Where("o.plan IN ?", filter.TenantPlan)
	}
	if len(filter.ControlPlaneArch) > 0 {
		tx = tx.Where("c.control_plane_arch IN ?", filter.ControlPlaneArch)
	}
	if len(filter.Provider) > 0 {
		tx = tx.Where("c.provider IN ?", filter.Provider)
	}
	if len(filter.Region) > 0 {
		tx = tx.Where("c.region IN ?", filter.Region)
	}
	if filter.Hide_deleted {
		tx = tx.Where("c.status NOT IN ?", []string{"deleted", "deletecomplete", "cleared"})
	}
	if filter.CreatedAtFrom > 0 {
		tx = tx.Where("unix_timestamp(c.created_at) >= ?", filter.CreatedAtFrom)
	}
	if filter.CreatedAtTo > 0 {
		tx = tx.Where("unix_timestamp(c.created_at) <= ?", filter.CreatedAtTo)
	}
	if filter.DeletedAtFrom > 0 {
		tx = tx.Where("unix_timestamp(c.deleted_at) >= ?", filter.DeletedAtFrom)
	}
	if filter.DeletedAtTo > 0 {
		tx = tx.Where("unix_timestamp(c.deleted_at) <= ?", filter.DeletedAtTo)
	}
	return tx
}

func applyDashboardOrder(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter) *gorm.DB {
	order := "DESC"
	if filter.Order == "ascend" {
		order = "ASC"
	}
	switch filter.Sort {
	case "clusterCreatedAt":
		tx = tx.Order("c.created_at " + order).Order("c.id " + order)
	case "clusterDeletedAt":
		tx = tx.Order("c.deleted_at " + order).Order("c.id " + order)
	case "clusterDuration":
		tx = tx.Order("TIMEDIFF(c.creation_finished_at, c.created_at) " + order).Order("c.id " + order)
	default:
		tx = tx.Order("c.created_at " + order).Order("c.id " + order)
	}
	return tx
}

// queryMode: "" (no query predicate), "full" (OR of equalities + name LIKE).
func applyDashboardOrgClusterFilters(tx *gorm.DB, gormDB gorm.DB, filter *view.ListOrganizationAndClusterFilter) *gorm.DB {
	query := strings.TrimSpace(filter.Query)
	if len(query) > 0 {
		queryCond := gormDB.Where("c.cluster_id = ?", query).
			Or("c.parent_id = ?", query).
			Or("c.project_id = ?", query).
			Or("o.dbaas_org_id = ? AND o.dbaas_org_id != 0", query)

		if disabledOrgIDs, allowTextSearch := clustersearch.DisabledOrgIDsForTextSearch(); allowTextSearch {
			likeQuery := fmt.Sprintf("%%%s%%", query)
			textCond := gormDB.Where("c.name LIKE ?", likeQuery).
				Or("o.name LIKE ?", likeQuery)
			if len(disabledOrgIDs) > 0 {
				textCond = gormDB.Where("c.org_id NOT IN ?", disabledOrgIDs).Where(textCond)
			}
			queryCond = queryCond.Or(textCond)
		}
		tx = tx.Where(queryCond)
	}
	tx = applyDashboardNonQueryFilters(tx, filter)
	return applyDashboardOrder(tx, filter)
}

// dashboardFastBranch is one indexed equality predicate. Branches are
// combined with UNION so a row matching more than one exact field is deduped.
type dashboardFastBranch struct {
	table string
	apply func(*gorm.DB) *gorm.DB
}

func (b dashboardFastBranch) clusterTable() string {
	if b.table != "" {
		return b.table
	}
	return "clusters c"
}

type dashboardTextSearchPolicy struct {
	disabledOrgIDs []uint
	allow          bool
}

func currentDashboardTextSearchPolicy() dashboardTextSearchPolicy {
	disabledOrgIDs, allowTextSearch := clustersearch.DisabledOrgIDsForTextSearch()
	return dashboardTextSearchPolicy{
		disabledOrgIDs: disabledOrgIDs,
		allow:          allowTextSearch,
	}
}

// dashboardFuzzySearchPolicyForQuery decides whether an exact-name result can
// return immediately or the request needs the policy-protected fuzzy fallback.
func dashboardFuzzySearchPolicyForQuery(query string, exactNameMatch bool) (dashboardTextSearchPolicy, error) {
	if exactNameMatch {
		return dashboardTextSearchPolicy{}, nil
	}
	if err := clustersearch.RequirePolicyForQuery(query); err != nil {
		return dashboardTextSearchPolicy{}, err
	}
	return currentDashboardTextSearchPolicy(), nil
}

func dashboardClusterTable(useOrgIDIndexHint bool) string {
	if useOrgIDIndexHint {
		return "clusters c FORCE INDEX (idx_clusters_org_id)"
	}
	return "clusters c"
}

func applyDashboardFuzzySearchFilters(tx *gorm.DB, gormDB gorm.DB, filter *view.ListOrganizationAndClusterFilter, policy dashboardTextSearchPolicy) *gorm.DB {
	tx = applyDashboardFuzzySearchPredicates(tx, gormDB, filter, policy)
	return applyDashboardOrder(tx, filter)
}

func applyDashboardFuzzySearchPredicates(tx *gorm.DB, gormDB gorm.DB, filter *view.ListOrganizationAndClusterFilter, policy dashboardTextSearchPolicy) *gorm.DB {
	query := strings.TrimSpace(filter.Query)
	if query != "" && policy.allow {
		likeQuery := fmt.Sprintf("%%%s%%", query)
		textCond := gormDB.Where("c.name LIKE ?", likeQuery).
			Or("o.name LIKE ?", likeQuery)
		if len(policy.disabledOrgIDs) > 0 {
			textCond = gormDB.Where("c.org_id NOT IN ?", policy.disabledOrgIDs).Where(textCond)
		}
		tx = tx.Where(textCond)
	}
	return applyDashboardNonQueryFilters(tx, filter)
}

func dashboardOrgIDFastBranch(orgIDs []uint) (dashboardFastBranch, bool) {
	if len(orgIDs) == 0 {
		return dashboardFastBranch{}, false
	}

	branch := dashboardFastBranch{
		table: "clusters c FORCE INDEX (" + db.ClusterOrgIDCreatedAtIndex + ")",
	}
	if len(orgIDs) == 1 {
		branch.apply = func(q *gorm.DB) *gorm.DB { return q.Where("c.org_id = ?", orgIDs[0]) }
	} else {
		branch.apply = func(q *gorm.DB) *gorm.DB { return q.Where("c.org_id IN ?", orgIDs) }
	}
	return branch, true
}

func dashboardNumericFastBranches(query string, orgIDs []uint) []dashboardFastBranch {
	branches := []dashboardFastBranch{
		{apply: func(q *gorm.DB) *gorm.DB { return q.Where("c.cluster_id = ?", query) }},
		{apply: func(q *gorm.DB) *gorm.DB { return q.Where("c.parent_id = ?", query) }},
		{apply: func(q *gorm.DB) *gorm.DB { return q.Where("c.project_id = ?", query) }},
	}
	if orgBranch, ok := dashboardOrgIDFastBranch(orgIDs); ok {
		branches = append(branches, orgBranch)
	}
	return branches
}

func dashboardNameFastBranches(query string, orgIDs []uint) []dashboardFastBranch {
	branches := []dashboardFastBranch{
		{
			table: "clusters c FORCE INDEX (" + db.ClusterNameCreatedAtIndex + ")",
			apply: func(q *gorm.DB) *gorm.DB {
				return q.Where("c.name = ?", query)
			},
		},
	}
	if orgBranch, ok := dashboardOrgIDFastBranch(orgIDs); ok {
		branches = append(branches, orgBranch)
	}
	return branches
}

func dashboardOrgIDsQuery(tx *gorm.DB, dbaasOrgID uint64) *gorm.DB {
	return tx.Table("organizations").
		Select("id").
		Where("dbaas_org_id = ?", dbaasOrgID)
}

func resolveDashboardOrgIDs(tx *gorm.DB, dbaasOrgID uint64) ([]uint, error) {
	var orgIDs []uint
	err := dashboardOrgIDsQuery(tx, dbaasOrgID).Pluck("id", &orgIDs).Error
	return orgIDs, err
}

func dashboardOrgIDsByNameQuery(tx *gorm.DB, name string) *gorm.DB {
	return tx.Table("organizations").
		Select("id").
		Where("name = ?", name)
}

func resolveDashboardOrgIDsByName(tx *gorm.DB, name string) ([]uint, error) {
	var orgIDs []uint
	err := dashboardOrgIDsByNameQuery(tx, name).Pluck("id", &orgIDs).Error
	return orgIDs, err
}

func dashboardCountNeedsOrganization(filter *view.ListOrganizationAndClusterFilter) bool {
	return len(filter.TenantPlan) > 0
}

func dashboardCountQuery(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter) *gorm.DB {
	query := tx.Table("clusters c").Select("count(*) as total")
	if dashboardCountNeedsOrganization(filter) {
		query = query.Joins("left join organizations o on c.org_id = o.id")
	}
	return applyDashboardNonQueryFilters(query, filter)
}

func dashboardFuzzyCountQuery(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, policy dashboardTextSearchPolicy) *gorm.DB {
	query := tx.Table(dashboardClusterTable(len(policy.disabledOrgIDs) > 0)).
		Select("count(*) as total").
		Joins("left join organizations o on c.org_id = o.id")
	return applyDashboardFuzzySearchPredicates(query, *tx, filter, policy)
}

type dashboardRawSQL struct {
	SQL  string
	Args []any
}

func buildDashboardDryRunSQL(query *gorm.DB, dest any) (dashboardRawSQL, error) {
	result := query.Session(&gorm.Session{DryRun: true}).Find(dest)
	if result.Error != nil {
		return dashboardRawSQL{}, result.Error
	}
	return dashboardRawSQL{
		SQL:  strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(result.Statement.SQL.String()), ";")),
		Args: append([]any(nil), result.Statement.Vars...),
	}, nil
}

// buildDashboardFastUnionSQL builds one parameterized UNION of all branches.
// selectIDsOnly=true is for COUNT or existence checks (c.id only); false is for
// LIST. List branches are ordered and capped before UNION so the outer sort
// only processes the rows needed for the requested page.
func buildDashboardFastUnionSQL(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, branches []dashboardFastBranch, selectIDsOnly bool, branchLimit int) (dashboardRawSQL, error) {
	if len(branches) == 0 {
		return dashboardRawSQL{}, nil
	}
	parts := make([]string, 0, len(branches))
	args := make([]any, 0)
	for _, br := range branches {
		sel := dashboardClusterExternalSelect
		if selectIDsOnly {
			sel = "c.id"
		}
		query := tx.Table(br.clusterTable()).Select(sel)
		if !selectIDsOnly || dashboardCountNeedsOrganization(filter) {
			query = query.Joins("left join organizations o on c.org_id = o.id")
		}
		query = br.apply(query)
		query = applyDashboardNonQueryFilters(query, filter)

		var raw dashboardRawSQL
		var err error
		if selectIDsOnly {
			if branchLimit > 0 {
				query = query.Limit(branchLimit)
			}
			var rows []struct {
				ID uint `gorm:"column:id"`
			}
			raw, err = buildDashboardDryRunSQL(query, &rows)
		} else {
			if branchLimit > 0 {
				query = applyDashboardOrder(query, filter).Limit(branchLimit)
			}
			var items []*view.DashboardItemExternal
			raw, err = buildDashboardDryRunSQL(query, &items)
		}
		if err != nil {
			return dashboardRawSQL{}, err
		}
		parts = append(parts, "("+raw.SQL+")")
		args = append(args, raw.Args...)
	}
	return dashboardRawSQL{SQL: strings.Join(parts, " UNION "), Args: args}, nil
}

func dashboardUnionOrderByClause(filter *view.ListOrganizationAndClusterFilter) string {
	order := "DESC"
	if filter.Order == "ascend" {
		order = "ASC"
	}
	switch filter.Sort {
	case "clusterDeletedAt":
		return "ORDER BY cluster_deleted_at " + order + ", cluster_row_id " + order
	case "clusterDuration":
		return "ORDER BY cluster_duration " + order + ", cluster_row_id " + order
	default:
		return "ORDER BY cluster_created_at " + order + ", cluster_row_id " + order
	}
}

const maxDashboardDeployTypeV2FastBranches = 16

// dashboardDeployTypeV2FastValues recognizes sparse multi-value v2 filters
// whose mutually exclusive values can each use the (deploy_type_v2, created_at)
// index. A filter containing starter stays on the generic path because starter
// dominates the table and a reverse created-at scan reaches a page quickly.
func dashboardDeployTypeV2FastValues(filter *view.ListOrganizationAndClusterFilter) ([]string, bool) {
	if filter == nil || strings.TrimSpace(filter.Query) != "" || len(filter.DeployType) > 0 {
		return nil, false
	}
	if filter.Sort != "" && filter.Sort != "clusterCreatedAt" {
		return nil, false
	}
	if len(filter.ClusterIDs) > 0 || len(filter.ClusterStatus) > 0 || len(filter.Version) > 0 ||
		len(filter.TenantPlan) > 0 || len(filter.ControlPlaneArch) > 0 || len(filter.Provider) > 0 ||
		len(filter.Region) > 0 || filter.CreatedAtFrom > 0 || filter.CreatedAtTo > 0 ||
		filter.DeletedAtFrom > 0 || filter.DeletedAtTo > 0 {
		return nil, false
	}

	deployTypes := make([]string, 0, len(filter.DeployTypeV2))
	seen := make(map[string]struct{}, len(filter.DeployTypeV2))
	for _, deployType := range filter.DeployTypeV2 {
		// Preserve the generic path for non-canonical input. Depending on the
		// database collation, case or whitespace variants can compare equal;
		// putting them in separate UNION ALL branches could duplicate rows.
		if deployType == "" || deployType != strings.TrimSpace(deployType) || deployType != strings.ToLower(deployType) {
			return nil, false
		}
		if deployType == o11y.DeployTypeV2Starter {
			return nil, false
		}
		if _, ok := seen[deployType]; ok {
			continue
		}
		seen[deployType] = struct{}{}
		deployTypes = append(deployTypes, deployType)
		if len(deployTypes) > maxDashboardDeployTypeV2FastBranches {
			return nil, false
		}
	}
	if len(deployTypes) < 2 {
		return nil, false
	}
	return deployTypes, true
}

func boundedDashboardBranchLimit(offset, limit int) (int, bool) {
	maxInt := int(^uint(0) >> 1)
	if offset < 0 || limit <= 0 || offset > maxInt-limit {
		return 0, false
	}
	return offset + limit, true
}

// buildDashboardDeployTypeV2FastListSQL splits a sparse multi-value IN scan
// into bounded equality scans. MySQL can then read each branch in created_at
// order from the composite index, and the outer sort sees at most
// len(deployTypes)*(offset+limit) rows. UNION ALL is safe because a cluster has
// exactly one canonical deploy_type_v2 value.
func buildDashboardDeployTypeV2FastListSQL(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, offset, limit int) (dashboardRawSQL, error) {
	deployTypes, ok := dashboardDeployTypeV2FastValues(filter)
	branchLimit, bounded := boundedDashboardBranchLimit(offset, limit)
	if !ok || !bounded {
		return dashboardRawSQL{}, nil
	}

	branchFilter := *filter
	branchFilter.DeployTypeV2 = nil
	parts := make([]string, 0, len(deployTypes))
	args := make([]any, 0)
	for _, deployType := range deployTypes {
		query := tx.Table("clusters c FORCE INDEX ("+db.ClusterDeployTypeV2CreatedAtIndex+")").
			Select(dashboardClusterExternalSelect).
			Joins("left join organizations o on c.org_id = o.id").
			Where("c.deploy_type_v2 = ?", deployType)
		query = applyDashboardNonQueryFilters(query, &branchFilter)
		query = applyDashboardOrder(query, &branchFilter).Limit(branchLimit)
		var items []*view.DashboardItemExternal
		raw, err := buildDashboardDryRunSQL(query, &items)
		if err != nil {
			return dashboardRawSQL{}, err
		}
		parts = append(parts, "("+raw.SQL+")")
		args = append(args, raw.Args...)
	}

	return dashboardRawSQL{
		SQL: fmt.Sprintf(
			"SELECT * FROM (%s) AS unioned %s LIMIT ? OFFSET ?",
			strings.Join(parts, " UNION ALL "),
			dashboardUnionOrderByClause(filter),
		),
		Args: append(args, limit, offset),
	}, nil
}

func listDashboardDeployTypeV2FastUnion(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, offset, limit int, items *[]*view.DashboardItemExternal) error {
	raw, err := buildDashboardDeployTypeV2FastListSQL(tx, filter, offset, limit)
	if err != nil {
		return err
	}
	if raw.SQL == "" {
		return fmt.Errorf("deploy type v2 fast list called with unsupported filters")
	}
	return tx.Raw(raw.SQL, raw.Args...).Scan(items).Error
}

func countDashboardFastUnion(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, branches []dashboardFastBranch) (int64, error) {
	inner, err := buildDashboardFastUnionSQL(tx, filter, branches, true, 0)
	if err != nil {
		return 0, err
	}
	if inner.SQL == "" {
		return 0, nil
	}
	var total int64
	err = tx.Raw("SELECT COUNT(*) FROM ("+inner.SQL+") AS t", inner.Args...).Scan(&total).Error
	return total, err
}

func countNumericFastUnion(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (int64, error) {
	return countDashboardFastUnion(tx, filter, dashboardNumericFastBranches(query, orgIDs))
}

func countNameFastUnion(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (int64, error) {
	return countDashboardFastUnion(tx, filter, dashboardNameFastBranches(query, orgIDs))
}

func buildDashboardFastExistsSQL(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, branches []dashboardFastBranch) (dashboardRawSQL, error) {
	inner, err := buildDashboardFastUnionSQL(tx, filter, branches, true, 1)
	if err != nil {
		return dashboardRawSQL{}, err
	}
	if inner.SQL == "" {
		return dashboardRawSQL{}, nil
	}
	return dashboardRawSQL{SQL: "SELECT EXISTS(SELECT 1 FROM (" + inner.SQL + ") AS t) AS found", Args: inner.Args}, nil
}

func buildNumericFastExistsSQL(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (dashboardRawSQL, error) {
	return buildDashboardFastExistsSQL(tx, filter, dashboardNumericFastBranches(query, orgIDs))
}

func buildNameFastExistsSQL(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (dashboardRawSQL, error) {
	return buildDashboardFastExistsSQL(tx, filter, dashboardNameFastBranches(query, orgIDs))
}

func dashboardFastUnionExists(tx *gorm.DB, raw dashboardRawSQL, err error) (bool, error) {
	if err != nil || raw.SQL == "" {
		return false, err
	}
	var result struct {
		Found bool `gorm:"column:found"`
	}
	err = tx.Raw(raw.SQL, raw.Args...).Scan(&result).Error
	return result.Found, err
}

func numericFastUnionExists(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (bool, error) {
	raw, err := buildNumericFastExistsSQL(tx, query, orgIDs, filter)
	return dashboardFastUnionExists(tx, raw, err)
}

func nameFastUnionExists(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter) (bool, error) {
	raw, err := buildNameFastExistsSQL(tx, query, orgIDs, filter)
	return dashboardFastUnionExists(tx, raw, err)
}

func buildDashboardFastListSQL(tx *gorm.DB, filter *view.ListOrganizationAndClusterFilter, branches []dashboardFastBranch, offset, limit int) (dashboardRawSQL, error) {
	branchLimit, ok := boundedDashboardBranchLimit(offset, limit)
	if !ok {
		return dashboardRawSQL{}, fmt.Errorf("invalid dashboard pagination bounds")
	}
	inner, err := buildDashboardFastUnionSQL(tx, filter, branches, false, branchLimit)
	if err != nil {
		return dashboardRawSQL{}, err
	}
	if inner.SQL == "" {
		return dashboardRawSQL{}, nil
	}
	orderClause := dashboardUnionOrderByClause(filter)
	return dashboardRawSQL{
		SQL:  fmt.Sprintf("SELECT * FROM (%s) AS unioned %s LIMIT ? OFFSET ?", inner.SQL, orderClause),
		Args: append(inner.Args, limit, offset),
	}, nil
}

func buildNumericFastListSQL(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter, offset, limit int) (dashboardRawSQL, error) {
	return buildDashboardFastListSQL(tx, filter, dashboardNumericFastBranches(query, orgIDs), offset, limit)
}

func buildNameFastListSQL(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter, offset, limit int) (dashboardRawSQL, error) {
	return buildDashboardFastListSQL(tx, filter, dashboardNameFastBranches(query, orgIDs), offset, limit)
}

func listNumericFastUnion(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter, offset, limit int, items *[]*view.DashboardItemExternal) error {
	raw, err := buildNumericFastListSQL(tx, query, orgIDs, filter, offset, limit)
	if err != nil {
		return err
	}
	if raw.SQL == "" {
		return nil
	}
	return tx.Raw(raw.SQL, raw.Args...).Scan(items).Error
}

func listNameFastUnion(tx *gorm.DB, query string, orgIDs []uint, filter *view.ListOrganizationAndClusterFilter, offset, limit int, items *[]*view.DashboardItemExternal) error {
	raw, err := buildNameFastListSQL(tx, query, orgIDs, filter, offset, limit)
	if err != nil {
		return err
	}
	if raw.SQL == "" {
		return nil
	}
	return tx.Raw(raw.SQL, raw.Args...).Scan(items).Error
}

func ListClusters(ctx *session.CtxUser, query string, orgID, projectID uint,
	page, perPage int,
	hideServerlessTier, hideDeleted bool,
) ([]db.Cluster, int64, error) {
	org, err := GetOrganization(ctx, orgID)
	if err != nil {
		return nil, 0, err
	}

	ds, ok := clusterDatasources[view.DataSourceType(org.OrgType)]
	if !ok {
		return nil, 0, fmt.Errorf("unsupport org type, type=%s", org.OrgType)
	}

	return ds.listClusters(ctx, query, orgID, projectID, page, perPage, hideServerlessTier, hideDeleted)
}

func GetCluster(ctx *session.CtxUser, id string, orgID uint) (*db.Cluster, error) {
	org, err := GetOrganization(ctx, orgID)
	if err != nil {
		return nil, err
	}

	ds, ok := clusterDatasources[view.DataSourceType(org.OrgType)]
	if !ok {
		return nil, fmt.Errorf("unsupport org type, type=%s", org.OrgType)
	}

	return ds.getCluster(ctx, id, org)
}

func RealtimeGetCloudCluster(ctx *session.CtxUser, clusterID uint64) (*core.Cluster, string, error) {
	return cloud2.RealtimeGetCluster(ctx, clusterID)
}

func RealtimeGetMgmtCluster(ctx *session.CtxUser, clusterID uint64) (*mgmtv1.Cluster, error) {
	return cloud2.RealtimeGetMgmtCluster(ctx, clusterID)
}

func RealtimeGetMgmtClusters(ctx *session.CtxUser, clusterIDs []uint64) ([]*mgmtv1.Cluster, error) {
	return cloud2.RealtimeGetMgmtClusters(ctx, clusterIDs)
}

func GetCloudCluster(ctx *session.CtxUser, clusterID string, orgID uint) (*db.Cluster, error) {
	return cloud2.GetCluster(ctx, clusterID, orgID)
}

func ListRegions(ctx *session.CtxUser, provider []string) ([]string, error) {
	return cloud2.ListRegions(ctx, provider)
}

func ListOrganizationAndCluster(ctx *session.CtxUser,
	extraFilter *view.ListOrganizationAndClusterFilter,
	page, perPage int,
) ([]*view.DashboardItemExternal, int64, error) {
	result, err := ListOrganizationAndClusterPage(ctx, extraFilter, page, perPage, true)
	if err != nil {
		return nil, 0, err
	}
	return result.Items, *result.Total, nil
}

// ListOrganizationAndClusterPage lists dashboard rows and optionally skips the exact total query.
func ListOrganizationAndClusterPage(ctx *session.CtxUser,
	extraFilter *view.ListOrganizationAndClusterFilter,
	page, perPage int,
	includeTotal bool,
) (*ListOrganizationAndClusterResult, error) {
	if !ctx.Internal {
		return nil, error2.ErrForbidden.New("permission deny")
	}
	var err error
	page, perPage, err = normalizeDashboardPagination(page, perPage)
	if err != nil {
		return nil, err
	}
	extraFilter = normalizeDashboardDeployTypeFilters(extraFilter)

	query := strings.TrimSpace(extraFilter.Query)
	if !includeTotal {
		return listOrganizationAndClusterWithoutTotal(query, extraFilter, page, perPage)
	}

	items, total, err := listOrganizationAndClusterWithTotal(query, extraFilter, page, perPage)
	if err != nil {
		return nil, err
	}
	return &ListOrganizationAndClusterResult{Items: items, Total: &total}, nil
}

func listOrganizationAndClusterWithTotal(query string,
	extraFilter *view.ListOrganizationAndClusterFilter,
	page, perPage int,
) ([]*view.DashboardItemExternal, int64, error) {
	count, err := countDashboardClusters(query, extraFilter)
	if err != nil {
		return nil, 0, err
	}
	items, err := listDashboardClusters(query, extraFilter, page, perPage, count)
	return items, count.total, err
}

type dashboardClusterCountResult struct {
	total           int64
	useNumericUnion bool
	numericOrgIDs   []uint
	useNameUnion    bool
	nameOrgIDs      []uint
	fuzzyPolicy     dashboardTextSearchPolicy
}

func countDashboardClusters(query string, extraFilter *view.ListOrganizationAndClusterFilter) (dashboardClusterCountResult, error) {
	result := dashboardClusterCountResult{}
	// Numeric path: single UNION of four indexed equalities; COUNT uses only c.id subqueries.
	if query != "" {
		if id, err := strconv.ParseUint(query, 10, 64); err == nil && id > 0 {
			err := db.Transaction(func(tx *gorm.DB) error {
				var e error
				result.numericOrgIDs, e = resolveDashboardOrgIDs(tx, id)
				if e != nil {
					return e
				}
				result.total, e = countNumericFastUnion(tx, query, result.numericOrgIDs, extraFilter)
				return e
			})
			if err != nil {
				return result, err
			}
			if result.total > 0 {
				result.useNumericUnion = true
				return result, nil
			}
		}
	}

	// Exact-name path: resolve tenant names through organizations.name, then
	// combine that indexed org_id branch with the indexed cluster-name branch.
	if query != "" {
		err := db.Transaction(func(tx *gorm.DB) error {
			var e error
			result.nameOrgIDs, e = resolveDashboardOrgIDsByName(tx, query)
			if e != nil {
				return e
			}
			result.total, e = countNameFastUnion(tx, query, result.nameOrgIDs, extraFilter)
			return e
		})
		if err != nil {
			return result, err
		}
		result.fuzzyPolicy, err = dashboardFuzzySearchPolicyForQuery(query, result.total > 0)
		if err != nil {
			return result, err
		}
		if result.total > 0 {
			result.useNameUnion = true
			return result, nil
		}
	}

	if query != "" {
		if !result.fuzzyPolicy.allow {
			return result, nil
		}
		err := db.Transaction(func(tx *gorm.DB) error {
			var count struct {
				Total int64
			}
			if err := dashboardFuzzyCountQuery(tx, extraFilter, result.fuzzyPolicy).Scan(&count).Error; err != nil {
				return err
			}
			result.total = count.Total
			return nil
		})
		return result, err
	}

	err := db.Transaction(func(tx *gorm.DB) error {
		var count struct {
			Total int64
		}
		if err := dashboardCountQuery(tx, extraFilter).Scan(&count).Error; err != nil {
			return err
		}
		result.total = count.Total
		return nil
	})
	return result, err
}

func listDashboardClusters(query string,
	extraFilter *view.ListOrganizationAndClusterFilter,
	page, perPage int,
	count dashboardClusterCountResult,
) ([]*view.DashboardItemExternal, error) {
	items := make([]*view.DashboardItemExternal, 0)
	startIndex := pagination.GetStartIndex(page, perPage)
	if count.useNumericUnion {
		err := db.TransactionWithTimeout(dashboardNumericListTimeout, func(tx *gorm.DB) error {
			return listNumericFastUnion(tx, query, count.numericOrgIDs, extraFilter, startIndex, perPage, &items)
		})
		return items, err
	}
	if count.useNameUnion {
		err := db.TransactionWithTimeout(dashboardNumericListTimeout, func(tx *gorm.DB) error {
			return listNameFastUnion(tx, query, count.nameOrgIDs, extraFilter, startIndex, perPage, &items)
		})
		return items, err
	}
	if _, ok := dashboardDeployTypeV2FastValues(extraFilter); ok {
		pageSize := pagination.GetSize(perPage)
		if _, bounded := boundedDashboardBranchLimit(startIndex, pageSize); bounded {
			err := db.Transaction(func(tx *gorm.DB) error {
				return listDashboardDeployTypeV2FastUnion(tx, extraFilter, startIndex, pageSize, &items)
			})
			return items, err
		}
	}

	if query != "" {
		if !count.fuzzyPolicy.allow {
			return items, nil
		}
		useOrgIDIndexHint := len(count.fuzzyPolicy.disabledOrgIDs) > 0
		err := db.Transaction(func(tx *gorm.DB) error {
			return applyDashboardFuzzySearchFilters(tx.Table(dashboardClusterTable(useOrgIDIndexHint)).
				Select(dashboardClusterExternalSelect).
				Joins("left join organizations o on c.org_id = o.id"), *tx, extraFilter, count.fuzzyPolicy).
				Limit(perPage).Offset(startIndex).Scan(&items).Error
		})
		return items, err
	}

	err := db.Transaction(func(tx *gorm.DB) error {
		return applyDashboardOrgClusterFilters(tx.Table("clusters c").Select(dashboardClusterExternalSelect).
			Joins("left join organizations o on c.org_id = o.id"), *tx, extraFilter).
			Limit(perPage).Offset(startIndex).Scan(&items).Error
	})
	return items, err
}

func listOrganizationAndClusterWithoutTotal(query string,
	extraFilter *view.ListOrganizationAndClusterFilter,
	page, perPage int,
) (*ListOrganizationAndClusterResult, error) {
	useNumericUnion := false
	var numericOrgIDs []uint
	if query != "" {
		if id, err := strconv.ParseUint(query, 10, 64); err == nil && id > 0 {
			err := db.Transaction(func(tx *gorm.DB) error {
				var e error
				numericOrgIDs, e = resolveDashboardOrgIDs(tx, id)
				if e != nil {
					return e
				}
				useNumericUnion, e = numericFastUnionExists(tx, query, numericOrgIDs, extraFilter)
				return e
			})
			if err != nil {
				return nil, err
			}
		}
	}
	useNameUnion := false
	var nameOrgIDs []uint
	if query != "" && !useNumericUnion {
		err := db.Transaction(func(tx *gorm.DB) error {
			var e error
			nameOrgIDs, e = resolveDashboardOrgIDsByName(tx, query)
			if e != nil {
				return e
			}
			useNameUnion, e = nameFastUnionExists(tx, query, nameOrgIDs, extraFilter)
			return e
		})
		if err != nil {
			return nil, err
		}
	}
	var fuzzyPolicy dashboardTextSearchPolicy
	if query != "" && !useNumericUnion {
		var err error
		fuzzyPolicy, err = dashboardFuzzySearchPolicyForQuery(query, useNameUnion)
		if err != nil {
			return nil, err
		}
	}

	startIndex := pagination.GetStartIndex(page, perPage)
	if useNumericUnion {
		return fetchDashboardPage(perPage, func(limit int, items *[]*view.DashboardItemExternal) error {
			return db.TransactionWithTimeout(dashboardNumericListTimeout, func(tx *gorm.DB) error {
				return listNumericFastUnion(tx, query, numericOrgIDs, extraFilter, startIndex, limit, items)
			})
		})
	}
	if useNameUnion {
		return fetchDashboardPage(perPage, func(limit int, items *[]*view.DashboardItemExternal) error {
			return db.TransactionWithTimeout(dashboardNumericListTimeout, func(tx *gorm.DB) error {
				return listNameFastUnion(tx, query, nameOrgIDs, extraFilter, startIndex, limit, items)
			})
		})
	}
	if _, ok := dashboardDeployTypeV2FastValues(extraFilter); ok {
		pageSize := pagination.GetSize(perPage)
		maxInt := int(^uint(0) >> 1)
		if pageSize < maxInt {
			if _, bounded := boundedDashboardBranchLimit(startIndex, pageSize+1); bounded {
				return fetchDashboardPage(perPage, func(limit int, items *[]*view.DashboardItemExternal) error {
					return db.Transaction(func(tx *gorm.DB) error {
						return listDashboardDeployTypeV2FastUnion(tx, extraFilter, startIndex, limit, items)
					})
				})
			}
		}
	}

	if query != "" {
		if !fuzzyPolicy.allow {
			return &ListOrganizationAndClusterResult{Items: make([]*view.DashboardItemExternal, 0)}, nil
		}
		useOrgIDIndexHint := len(fuzzyPolicy.disabledOrgIDs) > 0
		return fetchDashboardPage(perPage, func(limit int, items *[]*view.DashboardItemExternal) error {
			return db.Transaction(func(tx *gorm.DB) error {
				return applyDashboardFuzzySearchFilters(tx.Table(dashboardClusterTable(useOrgIDIndexHint)).
					Select(dashboardClusterExternalSelect).
					Joins("left join organizations o on c.org_id = o.id"), *tx, extraFilter, fuzzyPolicy).
					Limit(limit).Offset(startIndex).Scan(items).Error
			})
		})
	}

	return fetchDashboardPage(perPage, func(limit int, items *[]*view.DashboardItemExternal) error {
		return db.Transaction(func(tx *gorm.DB) error {
			return applyDashboardOrgClusterFilters(tx.Table("clusters c").Select(dashboardClusterExternalSelect).
				Joins("left join organizations o on c.org_id = o.id"), *tx, extraFilter).
				Limit(limit).Offset(startIndex).Scan(items).Error
		})
	})
}

func fetchDashboardPage(perPage int, list func(int, *[]*view.DashboardItemExternal) error) (*ListOrganizationAndClusterResult, error) {
	pageSize := pagination.GetSize(perPage)
	items := make([]*view.DashboardItemExternal, 0, pageSize+1)
	if err := list(pageSize+1, &items); err != nil {
		return nil, err
	}
	hasMore := len(items) > pageSize
	if hasMore {
		items = items[:pageSize]
	}
	return &ListOrganizationAndClusterResult{Items: items, HasMore: hasMore}, nil
}

func CloudClusterIsUserDebugAuthEnable(ctx *session.CtxUser, orgID string) (bool, error) {
	return cloud2.IsUserDebugAuthEnable(ctx, orgID)
}

func CloudClusterIsAccessDashboardAuthorized(ctx *session.CtxUser, orgID, clusterID, user string) (bool, error) {
	return cloud2.IsAccessDashboardAuthorized(ctx, orgID, clusterID, user)
}

type clusterDatasource interface {
	listClusters(ctxUser *session.CtxUser, query string, orgID, projectID uint,
		page, perPage int, hideServerlessTier, hideDeleted bool) ([]db.Cluster, int64, error)
	getCluster(ctxUser *session.CtxUser, id string, org *db.Organization) (*db.Cluster, error)
}

type cloudClusterDatasource struct{}

func (cd *cloudClusterDatasource) listClusters(ctxUser *session.CtxUser, query string,
	orgID, projectID uint,
	page, perPage int, hideServerlessTier, hideDeleted bool,
) ([]db.Cluster, int64, error) {
	return cloud2.ListClusters(ctxUser, view.QueryFilter{
		Page:  page,
		Limit: perPage,
		ExtraFilter: &view.ListClusterFilter{
			OrgID:              orgID,
			ProjectID:          projectID,
			Query:              query,
			HideDeleted:        hideDeleted,
			HideServerlessTier: hideServerlessTier,
		},
	})
}

func (cd *cloudClusterDatasource) getCluster(ctxUser *session.CtxUser, idStr string, org *db.Organization) (*db.Cluster, error) {
	c, err := cloud2.RealtimeGetFullCluster(ctxUser, idStr, org)
	if err != nil {
		log.Info("realtime get cloud cluster failed, try read from db")
		return cloud2.GetCluster(ctxUser, idStr, org.ID)
	}
	return c, nil
}

type opClusterDatasource struct{}

func (cd *opClusterDatasource) listClusters(ctxUser *session.CtxUser, query string,
	orgID, projectID uint,
	page, perPage int, hideServerlessTier, hideDeleted bool,
) ([]db.Cluster, int64, error) {
	return op.ListClusters(view.QueryFilter{
		Page:  page,
		Limit: perPage,
		ExtraFilter: &view.ListClusterFilter{
			OrgID:       orgID,
			Query:       query,
			HideDeleted: false,
		},
	})
}

func (cd *opClusterDatasource) getCluster(ctxUser *session.CtxUser, idStr string, org *db.Organization) (*db.Cluster, error) {
	return op.GetCluster(idStr, org)
}

func RealtimeCountAllDedicatedClusters(ctx *session.CtxUser) (int64, error) {
	return cloud2.CountDedicatedClusters(ctx)
}

func RealtimeCountAllServerlessClusters(ctx *session.CtxUser) (int64, error) {
	return cloud2.CountServerlessClusters(ctx)
}

// func RealtimeListDedicatedClusters(ctx *session.CtxUser, pager *base.Pager) (*core.ListClustersResp, error) {
// 	return cloud2.RealtimeListDedicatedClusters(ctx, pager)
// }

func RealtimeListDedicatedClustersByOrg(ctx *session.CtxUser, tenantID uint64) ([]*core.Cluster, []string, error) {
	return cloud2.RealtimeListDedicatedClustersByOrg(ctx, tenantID)
}

func RealtimeListServerlessClusters(ctx *session.CtxUser, tenantID uint64, includeDeleted bool) ([]*mgmtv1.Cluster, error) {
	return cloud2.RealtimeListServerlessClusters(ctx, tenantID, includeDeleted)
}

func RealtimeListServerlessBranches(ctx *session.CtxUser, clusterID uint64) ([]*rpc.BasicBranch, error) {
	return cloud2.RealtimeListServerlessBranches(ctx, clusterID)
}

func RealtimeListMgmtClustersByOrg(ctx *session.CtxUser, tenantID uint64) ([]*mgmtv1.Cluster, error) {
	return cloud2.RealtimeListMgmtClustersByOrg(ctx, tenantID)
}

func CountCloudClusters(ctx *session.CtxUser) (int64, error) {
	return cloud2.CountCloudClusters(ctx)
}

func ListCloudClusters(ctx *session.CtxUser, page, perPage int) ([]db.Cluster, error) {
	return cloud2.ListCloudClusters(ctx, page, perPage)
}

func GetCloudClusterComponents(ctx *session.CtxUser, clusterID uint64) ([]*dedicatedv1.Cluster_ClusterComponent, string, string, error) {
	return cloud2.GetCloudClusterComponents(ctx, clusterID)
}

func GetResourcePoolComponents(ctx *session.CtxUser, clusterID uint64) ([]*view.ResourcePoolComponent, error) {
	return cloud2.GetResourcePoolComponents(ctx, clusterID)
}

func GetDedicatedClusterComponents(ctx *session.CtxUser, clusterID uint64) ([]*view.ResourcePoolComponent, error) {
	return cloud2.GetDedicatedClusterComponents(ctx, clusterID)
}

func GetPremiumClusterComponents(ctx *session.CtxUser, clusterID, sharedPoolID uint64) ([]*view.ResourcePoolComponent, error) {
	return cloud2.GetPremiumClusterComponents(ctx, clusterID, sharedPoolID)
}

func GetResourcePoolWithProfile(ctx *session.CtxUser, clusterID uint64) (*nextgenv1.ResourcePoolProfile, *nextgen_client.SharedPool, error) {
	return cloud2.GetResourcePoolWithProfile(ctx, clusterID)
}

func ListPlanReplayerFiles(ctx context.Context, clusterID uint64, pageSize int, pageToken string) (*nextgen_client.ListPlanReplayerFilesResponse, error) {
	return cloud2.ListPlanReplayerFiles(ctx, clusterID, pageSize, pageToken)
}

func GeneratePlanReplayerFileDownloadURL(
	ctx context.Context,
	clusterID uint64,
	req *nextgen_client.GeneratePlanReplayerFileDownloadURLRequest,
) (*nextgen_client.GeneratePlanReplayerFileDownloadURLResponse, error) {
	return cloud2.GeneratePlanReplayerFileDownloadURL(ctx, clusterID, req)
}

func CountResourcePoolClusters(ctx *session.CtxUser) (int64, error) {
	return cloud2.CountResourcePoolClusters(ctx)
}

func ListResourcePoolClusters(ctx *session.CtxUser, page, perPage int) ([]db.Cluster, error) {
	return cloud2.ListResourcePoolClusters(ctx, page, perPage)
}

func ListResourcePoolWithProfile(ctx *session.CtxUser, resourcePoolIds []uint64) ([]*nextgenv1.ResourcePoolProfile, error) {
	return cloud2.ListResourcePoolProfile(ctx, resourcePoolIds)
}