package datasource

import (
	"fmt"
	"reflect"
	"strings"
	"testing"

	"gorm.io/driver/mysql"
	"gorm.io/gorm"

	"github.com/pingcap-inc/clinic/pkg/datasource/clustersearch"
	"github.com/pingcap-inc/clinic/pkg/datasource/view"
)

func TestNormalizeDashboardDeployTypeFiltersUsesV2ForEquivalentLegacyValues(t *testing.T) {
	original := &view.ListOrganizationAndClusterFilter{
		DeployType:    []string{"dedicated", "byoc", "dedicated"},
		ClusterStatus: []string{"active"},
	}

	normalized := normalizeDashboardDeployTypeFilters(original)

	if len(normalized.DeployType) != 0 {
		t.Fatalf("expected equivalent legacy filter to be removed, got %v", normalized.DeployType)
	}
	if want := []string{"dedicated", "byoc"}; !reflect.DeepEqual(normalized.DeployTypeV2, want) {
		t.Fatalf("expected v2 filter %v, got %v", want, normalized.DeployTypeV2)
	}
	if want := []string{"dedicated", "byoc", "dedicated"}; !reflect.DeepEqual(original.DeployType, want) {
		t.Fatalf("expected caller filter to remain unchanged, got %v", original.DeployType)
	}
	if !reflect.DeepEqual(normalized.ClusterStatus, original.ClusterStatus) {
		t.Fatalf("expected unrelated filters to remain unchanged")
	}

	sql := dashboardCountSQL(t, normalized)
	if strings.Contains(sql, "c.deploy_type IN") {
		t.Fatalf("expected normalized SQL not to filter the unindexed legacy column, sql=%s", sql)
	}
	if !strings.Contains(sql, "c.deploy_type_v2 IN (?,?)") {
		t.Fatalf("expected normalized SQL to use the indexed v2 column, sql=%s", sql)
	}
}

func TestNormalizeDashboardDeployTypeFiltersKeepsAmbiguousOrCombinedFilters(t *testing.T) {
	tests := []struct {
		name   string
		filter *view.ListOrganizationAndClusterFilter
	}{
		{
			name: "ambiguous legacy value",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployType: []string{"shared"},
			},
		},
		{
			name: "mixed equivalent and ambiguous legacy values",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployType: []string{"dedicated", "premium"},
			},
		},
		{
			name: "explicit v2 filter keeps existing and semantics",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployType:   []string{"dedicated"},
				DeployTypeV2: []string{"dedicated"},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := normalizeDashboardDeployTypeFilters(tt.filter)
			if !reflect.DeepEqual(got.DeployType, tt.filter.DeployType) {
				t.Fatalf("expected legacy filter to remain %v, got %v", tt.filter.DeployType, got.DeployType)
			}
			if !reflect.DeepEqual(got.DeployTypeV2, tt.filter.DeployTypeV2) {
				t.Fatalf("expected v2 filter to remain %v, got %v", tt.filter.DeployTypeV2, got.DeployTypeV2)
			}
		})
	}
}

func TestDashboardDeployTypeV2FastListUsesOneBoundedIndexBranchPerType(t *testing.T) {
	gdb := newDryRunDB(t)
	filter := &view.ListOrganizationAndClusterFilter{
		DeployTypeV2: []string{"premium", "essential", "byoc resource pool", "premium"},
	}

	raw, err := buildDashboardDeployTypeV2FastListSQL(gdb, filter, 20, 81)
	if err != nil {
		t.Fatalf("build deploy type fast list SQL: %v", err)
	}
	if raw.SQL == "" {
		t.Fatal("expected sparse multi-type created-at list to use the bounded union")
	}
	sql := raw.SQL

	if !strings.Contains(sql, "UNION ALL") {
		t.Fatalf("expected mutually exclusive deploy types to use UNION ALL, sql=%s", sql)
	}
	if count := strings.Count(sql, "FORCE INDEX (idx_clusters_deploy_type_v2_created_at)"); count != 3 {
		t.Fatalf("expected the deploy-type/created-at index in all branches, count=%d sql=%s", count, sql)
	}
	if count := strings.Count(sql, "c.deploy_type_v2 = ?"); count != 3 {
		t.Fatalf("expected one parameterized equality per branch, count=%d sql=%s", count, sql)
	}
	if strings.Contains(sql, "c.deploy_type_v2 IN") {
		t.Fatalf("expected branch equality instead of a multi-value range, sql=%s", sql)
	}
	if count := strings.Count(sql, "ORDER BY c.created_at DESC,c.id DESC LIMIT ?"); count != 3 {
		t.Fatalf("expected each branch to stop at offset + limit, count=%d sql=%s", count, sql)
	}
	if !strings.HasSuffix(sql, "ORDER BY cluster_created_at DESC, cluster_row_id DESC LIMIT ? OFFSET ?") {
		t.Fatalf("expected outer created-at pagination, sql=%s", sql)
	}
	wantArgs := []any{"premium", 101, "essential", 101, "byoc resource pool", 101, 81, 20}
	if !reflect.DeepEqual(raw.Args, wantArgs) {
		t.Fatalf("expected parameterized branch and page args %v, got %v", wantArgs, raw.Args)
	}
}

func TestDashboardDeployTypeV2FastListEligibilityIsNarrow(t *testing.T) {
	tooManyDeployTypes := make([]string, maxDashboardDeployTypeV2FastBranches+1)
	for i := range tooManyDeployTypes {
		tooManyDeployTypes[i] = fmt.Sprintf("sparse-type-%d", i)
	}

	tests := []struct {
		name   string
		filter *view.ListOrganizationAndClusterFilter
	}{
		{
			name:   "single deploy type",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: []string{"dedicated"}},
		},
		{
			name:   "starter among multiple deploy types",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: []string{"dedicated", "starter"}},
		},
		{
			name:   "case variant of starter among multiple deploy types",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: []string{"dedicated", "STARTER"}},
		},
		{
			name:   "duplicates collapse to one deploy type",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: []string{"dedicated", "dedicated"}},
		},
		{
			name:   "too many deploy type branches",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: tooManyDeployTypes},
		},
		{
			name:   "non canonical deploy type value",
			filter: &view.ListOrganizationAndClusterFilter{DeployTypeV2: []string{"dedicated", " byoc"}},
		},
		{
			name: "explicit legacy and v2 filters",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployType:   []string{"dedicated"},
				DeployTypeV2: []string{"dedicated", "byoc"},
			},
		},
		{
			name: "text search",
			filter: &view.ListOrganizationAndClusterFilter{
				Query:        "tenant-name",
				DeployTypeV2: []string{"dedicated", "byoc"},
			},
		},
		{
			name: "different sort",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployTypeV2: []string{"dedicated", "byoc"},
				Sort:         "clusterDeletedAt",
			},
		},
		{
			name: "companion cluster id filter",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployTypeV2: []string{"dedicated", "byoc"},
				ClusterIDs:   []string{"1001"},
			},
		},
		{
			name: "companion status filter",
			filter: &view.ListOrganizationAndClusterFilter{
				DeployTypeV2:  []string{"dedicated", "byoc"},
				ClusterStatus: []string{"active"},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			raw, err := buildDashboardDeployTypeV2FastListSQL(newDryRunDB(t), tt.filter, 0, 11)
			if err != nil {
				t.Fatalf("build deploy type fast list SQL: %v", err)
			}
			if raw.SQL != "" {
				t.Fatal("expected generic dashboard list path")
			}
		})
	}
}

func TestDashboardDeployTypeV2FastListRejectsOverflowingBranchLimit(t *testing.T) {
	maxInt := int(^uint(0) >> 1)
	raw, err := buildDashboardDeployTypeV2FastListSQL(newDryRunDB(t), &view.ListOrganizationAndClusterFilter{
		DeployTypeV2: []string{"dedicated", "byoc"},
	}, maxInt, 2)
	if err != nil {
		t.Fatalf("build deploy type fast list SQL: %v", err)
	}
	if raw.SQL != "" {
		t.Fatalf("expected unsafe pagination bounds to use the generic path, sql=%s", raw.SQL)
	}
}

func TestNormalizeDashboardPaginationRejectsOverflowBeforeOffsetCalculation(t *testing.T) {
	maxInt := int(^uint(0) >> 1)
	tests := []struct {
		name    string
		page    int
		perPage int
	}{
		{name: "page multiplication overflows", page: maxInt, perPage: 2},
		{name: "has-more extra row overflows", page: 1, perPage: maxInt},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if _, _, err := normalizeDashboardPagination(tt.page, tt.perPage); err != ErrInvalidDashboardPagination {
				t.Fatalf("expected invalid pagination error, got %v", err)
			}
		})
	}

	page, pageSize, err := normalizeDashboardPagination(0, 0)
	if err != nil {
		t.Fatalf("normalize default pagination: %v", err)
	}
	if page != 1 || pageSize != 10 {
		t.Fatalf("expected default page=1 size=10, got page=%d size=%d", page, pageSize)
	}
}

func newDryRunDB(t *testing.T) *gorm.DB {
	t.Helper()
	gdb, err := gorm.Open(mysql.New(mysql.Config{
		DSN:                       "gorm:gorm@tcp(localhost:9910)/gorm?charset=utf8&parseTime=True&loc=Local",
		SkipInitializeWithVersion: true,
	}), &gorm.Config{
		DryRun:               true,
		DisableAutomaticPing: true,
	})
	if err != nil {
		t.Fatalf("open dry-run db: %v", err)
	}
	return gdb
}

func dashboardCountSQL(t *testing.T, filter *view.ListOrganizationAndClusterFilter) string {
	t.Helper()
	gdb := newDryRunDB(t)
	var total struct {
		Total int64
	}
	stmt := dashboardCountQuery(gdb, filter).Find(&total).Statement
	return stmt.SQL.String()
}

func dashboardFuzzyCountSQL(t *testing.T, filter *view.ListOrganizationAndClusterFilter, policy dashboardTextSearchPolicy) string {
	t.Helper()
	gdb := newDryRunDB(t)
	var total struct {
		Total int64
	}
	stmt := dashboardFuzzyCountQuery(gdb, filter, policy).Find(&total).Statement
	return stmt.SQL.String()
}

func TestDashboardFuzzySearchExcludesLargeOrgsAndUsesOrgIndex(t *testing.T) {
	cleanup := clustersearch.SetSnapshotForTest(true, true, []uint{101, 202})
	defer cleanup()

	sql := dashboardFuzzyCountSQL(t, &view.ListOrganizationAndClusterFilter{Query: "abc"}, currentDashboardTextSearchPolicy())
	if !strings.Contains(sql, "FORCE INDEX (idx_clusters_org_id)") {
		t.Fatalf("expected org_id index hint, sql=%s", sql)
	}
	for _, exactBranch := range []string{"c.cluster_id = ?", "c.parent_id = ?", "c.project_id = ?", "o.dbaas_org_id = ?"} {
		if strings.Contains(sql, exactBranch) {
			t.Fatalf("expected fuzzy-only search without exact branch %q, sql=%s", exactBranch, sql)
		}
	}
	if !strings.Contains(sql, "c.name LIKE ?") || !strings.Contains(sql, "o.name LIKE ?") {
		t.Fatalf("expected text search branch, sql=%s", sql)
	}
	if !strings.Contains(sql, "c.org_id NOT IN (?,?)") {
		t.Fatalf("expected large org exclusion, sql=%s", sql)
	}
}

func TestDashboardExactNameMatchBypassesUnavailableFuzzyPolicy(t *testing.T) {
	cleanup := clustersearch.SetSnapshotForTest(true, false, nil)
	defer cleanup()

	if _, err := dashboardFuzzySearchPolicyForQuery("abc", true); err != nil {
		t.Fatalf("expected exact-name match to bypass fuzzy policy, err=%v", err)
	}
	if _, err := dashboardFuzzySearchPolicyForQuery("abc", false); err != clustersearch.ErrTextSearchPolicyUnavailable {
		t.Fatalf("expected text search policy error, err=%v", err)
	}
	policy, err := dashboardFuzzySearchPolicyForQuery("123", false)
	if err != nil {
		t.Fatalf("expected numeric query to bypass text search policy, err=%v", err)
	}
	if policy.allow {
		t.Fatal("expected unavailable policy to keep fuzzy search disabled")
	}
}

func TestDashboardDeployTypeV2FilterIsAppliedWithTextSearch(t *testing.T) {
	cleanup := clustersearch.SetSnapshotForTest(true, true, nil)
	defer cleanup()

	sql := dashboardFuzzyCountSQL(t, &view.ListOrganizationAndClusterFilter{
		Query:        "abc",
		DeployTypeV2: []string{"dedicated", "premium"},
	}, currentDashboardTextSearchPolicy())
	if !strings.Contains(sql, "c.deploy_type_v2 IN (?,?)") {
		t.Fatalf("expected deploy_type_v2 filter, sql=%s", sql)
	}
	if !strings.Contains(sql, "c.name LIKE ?") || !strings.Contains(sql, "o.name LIKE ?") {
		t.Fatalf("expected text search branch, sql=%s", sql)
	}
	for _, exactBranch := range []string{"c.cluster_id = ?", "c.parent_id = ?", "c.project_id = ?", "o.dbaas_org_id = ?"} {
		if strings.Contains(sql, exactBranch) {
			t.Fatalf("expected fuzzy-only search without exact branch %q, sql=%s", exactBranch, sql)
		}
	}
}

func TestDashboardCountOmitsOrderAndUnneededOrganizationJoin(t *testing.T) {
	sql := dashboardCountSQL(t, &view.ListOrganizationAndClusterFilter{
		ClusterStatus: []string{"active"},
		Sort:          "clusterDuration",
		Order:         "ascend",
	})

	if strings.Contains(sql, "ORDER BY") {
		t.Fatalf("expected count query without order, sql=%s", sql)
	}
	if strings.Contains(strings.ToLower(sql), "join organizations") {
		t.Fatalf("expected cluster-only count query without organization join, sql=%s", sql)
	}
}

func TestDashboardCountKeepsOrganizationJoinForTenantPlan(t *testing.T) {
	sql := dashboardCountSQL(t, &view.ListOrganizationAndClusterFilter{
		TenantPlan: []string{"premium"},
		Sort:       "clusterCreatedAt",
	})

	if strings.Contains(sql, "ORDER BY") {
		t.Fatalf("expected count query without order, sql=%s", sql)
	}
	if !strings.Contains(strings.ToLower(sql), "join organizations") {
		t.Fatalf("expected tenant-plan count query to keep organization join, sql=%s", sql)
	}
}

func TestDashboardFuzzyCountKeepsOrganizationJoinWithoutOrder(t *testing.T) {
	policy := dashboardTextSearchPolicy{allow: true}
	sql := dashboardFuzzyCountSQL(t, &view.ListOrganizationAndClusterFilter{
		Query: "abc",
		Sort:  "clusterDeletedAt",
	}, policy)

	if strings.Contains(sql, "ORDER BY") {
		t.Fatalf("expected fuzzy count query without order, sql=%s", sql)
	}
	if !strings.Contains(strings.ToLower(sql), "join organizations") {
		t.Fatalf("expected tenant-name fuzzy count to keep organization join, sql=%s", sql)
	}
}

func TestDashboardNumericFastUnionAppliesDeployTypeV2FilterToEveryBranch(t *testing.T) {
	gdb := newDryRunDB(t)
	filter := &view.ListOrganizationAndClusterFilter{
		DeployTypeV2: []string{"dedicated", "premium"},
	}
	branches := dashboardNumericFastBranches("123", []uint{101})

	raw, err := buildDashboardFastUnionSQL(gdb, filter, branches, true, 0)
	if err != nil {
		t.Fatalf("build numeric fast union SQL: %v", err)
	}
	sql := raw.SQL

	if count := strings.Count(sql, "c.deploy_type_v2 IN"); count != len(branches) {
		t.Fatalf("expected deploy_type_v2 filter in every branch, count=%d sql=%s", count, sql)
	}
	if strings.Contains(sql, "LIKE") {
		t.Fatalf("expected numeric fast union to avoid text search, sql=%s", sql)
	}
	if strings.Contains(sql, "LIMIT") || strings.Contains(sql, "ORDER BY") {
		t.Fatalf("expected count union to remain unbounded and unordered, sql=%s", sql)
	}
}

func TestDashboardNumericFastCountOnlyJoinsOrganizationWhereNeeded(t *testing.T) {
	gdb := newDryRunDB(t)
	branches := dashboardNumericFastBranches("123", []uint{101})

	raw, err := buildDashboardFastUnionSQL(gdb, &view.ListOrganizationAndClusterFilter{}, branches, true, 0)
	if err != nil {
		t.Fatalf("build numeric fast union SQL: %v", err)
	}
	sql := raw.SQL
	if count := strings.Count(strings.ToLower(sql), "join organizations"); count != 0 {
		t.Fatalf("expected count branches to avoid the organization join, count=%d sql=%s", count, sql)
	}

	raw, err = buildDashboardFastUnionSQL(gdb, &view.ListOrganizationAndClusterFilter{
		TenantPlan: []string{"premium"},
	}, branches, true, 0)
	if err != nil {
		t.Fatalf("build numeric fast union SQL with tenant plan: %v", err)
	}
	sql = raw.SQL
	if count := strings.Count(strings.ToLower(sql), "join organizations"); count != len(branches) {
		t.Fatalf("expected tenant-plan filter to retain every organization join, count=%d sql=%s", count, sql)
	}
}

func TestDashboardNumericFastUnionKeepsUserFiltersParameterized(t *testing.T) {
	gdb := newDryRunDB(t)
	unsafeLookingValue := "active?'\\value"
	raw, err := buildDashboardFastUnionSQL(gdb, &view.ListOrganizationAndClusterFilter{
		ClusterStatus: []string{unsafeLookingValue},
	}, dashboardNumericFastBranches("123", nil), true, 0)
	if err != nil {
		t.Fatalf("build numeric fast union SQL: %v", err)
	}

	if strings.Contains(raw.SQL, unsafeLookingValue) {
		t.Fatalf("expected user value to remain outside SQL text, sql=%s", raw.SQL)
	}
	if count := strings.Count(raw.SQL, "c.status IN (?)"); count != 3 {
		t.Fatalf("expected one parameterized status filter per branch, count=%d sql=%s", count, raw.SQL)
	}
	if count := countDashboardArg(raw.Args, unsafeLookingValue); count != 3 {
		t.Fatalf("expected user value in every branch args, count=%d args=%v", count, raw.Args)
	}
}

func TestDashboardNumericFastListLimitsEveryBranchForRequestedPage(t *testing.T) {
	tests := []struct {
		name          string
		filter        *view.ListOrganizationAndClusterFilter
		offset        int
		limit         int
		wantBranchSQL string
		wantOuterSQL  string
	}{
		{
			name:          "first page defaults to created at descending",
			filter:        &view.ListOrganizationAndClusterFilter{},
			offset:        0,
			limit:         10,
			wantBranchSQL: "ORDER BY c.created_at DESC,c.id DESC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_created_at DESC, cluster_row_id DESC LIMIT ? OFFSET ?",
		},
		{
			name: "first page sorts created at ascending",
			filter: &view.ListOrganizationAndClusterFilter{
				Sort:  "clusterCreatedAt",
				Order: "ascend",
			},
			offset:        0,
			limit:         10,
			wantBranchSQL: "ORDER BY c.created_at ASC,c.id ASC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_created_at ASC, cluster_row_id ASC LIMIT ? OFFSET ?",
		},
		{
			name: "later page sorts deleted at ascending",
			filter: &view.ListOrganizationAndClusterFilter{
				Sort:  "clusterDeletedAt",
				Order: "ascend",
			},
			offset:        20,
			limit:         10,
			wantBranchSQL: "ORDER BY c.deleted_at ASC,c.id ASC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_deleted_at ASC, cluster_row_id ASC LIMIT ? OFFSET ?",
		},
		{
			name: "first page sorts deleted at descending",
			filter: &view.ListOrganizationAndClusterFilter{
				Sort: "clusterDeletedAt",
			},
			offset:        0,
			limit:         10,
			wantBranchSQL: "ORDER BY c.deleted_at DESC,c.id DESC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_deleted_at DESC, cluster_row_id DESC LIMIT ? OFFSET ?",
		},
		{
			name: "later page sorts duration descending",
			filter: &view.ListOrganizationAndClusterFilter{
				Sort: "clusterDuration",
			},
			offset:        5,
			limit:         5,
			wantBranchSQL: "ORDER BY TIMEDIFF(c.creation_finished_at, c.created_at) DESC,c.id DESC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_duration DESC, cluster_row_id DESC LIMIT ? OFFSET ?",
		},
		{
			name: "first page sorts duration ascending",
			filter: &view.ListOrganizationAndClusterFilter{
				Sort:  "clusterDuration",
				Order: "ascend",
			},
			offset:        0,
			limit:         10,
			wantBranchSQL: "ORDER BY TIMEDIFF(c.creation_finished_at, c.created_at) ASC,c.id ASC LIMIT ?",
			wantOuterSQL:  "ORDER BY cluster_duration ASC, cluster_row_id ASC LIMIT ? OFFSET ?",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gdb := newDryRunDB(t)
			orgIDs := []uint{101}
			raw, err := buildNumericFastListSQL(gdb, "123", orgIDs, tt.filter, tt.offset, tt.limit)
			if err != nil {
				t.Fatalf("build numeric fast list SQL: %v", err)
			}
			sql := raw.SQL

			if count := strings.Count(sql, tt.wantBranchSQL); count != len(dashboardNumericFastBranches("123", orgIDs)) {
				t.Fatalf("expected branch order and limit in every branch, count=%d sql=%s", count, sql)
			}
			if !strings.HasSuffix(sql, tt.wantOuterSQL) {
				t.Fatalf("expected outer pagination %q, sql=%s", tt.wantOuterSQL, sql)
			}
			wantCapArgs := len(dashboardNumericFastBranches("123", orgIDs))
			if tt.offset == 0 {
				wantCapArgs++ // The outer limit has the same value on the first page.
			}
			if count := countDashboardArg(raw.Args, tt.offset+tt.limit); count != wantCapArgs {
				t.Fatalf("expected branch cap arg in every branch, count=%d want=%d args=%v", count, wantCapArgs, raw.Args)
			}
		})
	}
}

func TestDashboardNumericFastExistsIsBoundedWithoutCount(t *testing.T) {
	gdb := newDryRunDB(t)
	orgIDs := []uint{101}
	raw, err := buildNumericFastExistsSQL(gdb, "123", orgIDs, &view.ListOrganizationAndClusterFilter{})
	if err != nil {
		t.Fatalf("build numeric fast exists SQL: %v", err)
	}
	sql := raw.SQL

	if strings.Contains(sql, "COUNT") {
		t.Fatalf("expected existence probe without count, sql=%s", sql)
	}
	if count := strings.Count(sql, "LIMIT ?"); count != len(dashboardNumericFastBranches("123", orgIDs)) {
		t.Fatalf("expected every exact branch to stop after one row, count=%d sql=%s", count, sql)
	}
	if count := countDashboardArg(raw.Args, 1); count != len(dashboardNumericFastBranches("123", orgIDs)) {
		t.Fatalf("expected every exact branch to bind limit one, count=%d args=%v", count, raw.Args)
	}
}

func TestDashboardNumericOrgBranchUsesResolvedOrgIDAndCompositeIndex(t *testing.T) {
	gdb := newDryRunDB(t)
	orgIDs := []uint{101}

	raw, err := buildNumericFastListSQL(gdb, "123", orgIDs, &view.ListOrganizationAndClusterFilter{}, 0, 11)
	if err != nil {
		t.Fatalf("build numeric fast list SQL: %v", err)
	}
	sql := raw.SQL

	if !strings.Contains(sql, "clusters c FORCE INDEX (idx_clusters_org_id_created_at)") {
		t.Fatalf("expected exact-org branch to use the composite index, sql=%s", sql)
	}
	if !strings.Contains(sql, "WHERE c.org_id = ?") || countDashboardArg(raw.Args, uint(101)) != 1 {
		t.Fatalf("expected exact-org branch to filter by the resolved internal ID, sql=%s", sql)
	}
	if strings.Contains(sql, "WHERE o.dbaas_org_id") {
		t.Fatalf("expected exact-org branch not to filter through the organization join, sql=%s", sql)
	}
}

func TestDashboardOrgIDResolutionKeepsEveryMatchingOrganization(t *testing.T) {
	gdb := newDryRunDB(t)
	var rows []struct {
		ID uint
	}
	stmt := dashboardOrgIDsQuery(gdb, 123).Find(&rows).Statement
	sql := stmt.SQL.String()

	if !strings.Contains(sql, "FROM `organizations` WHERE dbaas_org_id = ?") {
		t.Fatalf("expected indexed external-org lookup, sql=%s", sql)
	}
	if strings.Contains(sql, "LIMIT") {
		t.Fatalf("expected all matching internal org IDs without LIMIT, sql=%s", sql)
	}
	if strings.Contains(sql, "deleted_at") {
		t.Fatalf("expected the lookup to preserve the existing join's deleted-row semantics, sql=%s", sql)
	}
}

func TestDashboardExactNameOrgResolutionUsesIndexedEquality(t *testing.T) {
	gdb := newDryRunDB(t)
	var rows []struct {
		ID uint
	}
	stmt := dashboardOrgIDsByNameQuery(gdb, "tenant-name").Find(&rows).Statement
	sql := stmt.SQL.String()

	if !strings.Contains(sql, "FROM `organizations` WHERE name = ?") {
		t.Fatalf("expected exact tenant-name equality, sql=%s", sql)
	}
	if strings.Contains(sql, "LIKE") || strings.Contains(sql, "LIMIT") {
		t.Fatalf("expected all exact tenant-name matches without fuzzy search or limit, sql=%s", sql)
	}
}

func TestDashboardExactNameListUsesSeparateIndexedBranches(t *testing.T) {
	gdb := newDryRunDB(t)
	orgIDs := []uint{101, 202}

	raw, err := buildNameFastListSQL(gdb, "shared-name", orgIDs, &view.ListOrganizationAndClusterFilter{}, 0, 11)
	if err != nil {
		t.Fatalf("build exact-name list SQL: %v", err)
	}
	sql := raw.SQL

	if strings.Contains(sql, "LIKE") || strings.Contains(sql, " OR ") {
		t.Fatalf("expected equality branches without fuzzy predicates or cross-table OR, sql=%s", sql)
	}
	if !strings.Contains(sql, "clusters c FORCE INDEX (idx_clusters_name_created_at)") || !strings.Contains(sql, "WHERE c.name = ?") {
		t.Fatalf("expected cluster-name branch to use the name/created-at index, sql=%s", sql)
	}
	if !strings.Contains(sql, "clusters c FORCE INDEX (idx_clusters_org_id_created_at)") || !strings.Contains(sql, "WHERE c.org_id IN (?,?)") {
		t.Fatalf("expected tenant-name branch to use resolved org IDs and the org/created-at index, sql=%s", sql)
	}
	if strings.Contains(sql, "UNION ALL") || !strings.Contains(sql, " UNION ") {
		t.Fatalf("expected UNION to deduplicate a cluster matching both exact branches, sql=%s", sql)
	}
	if count := strings.Count(sql, "ORDER BY c.created_at DESC,c.id DESC LIMIT ?"); count != 2 {
		t.Fatalf("expected both exact branches to be bounded before union, count=%d sql=%s", count, sql)
	}
	if !strings.HasSuffix(sql, "ORDER BY cluster_created_at DESC, cluster_row_id DESC LIMIT ? OFFSET ?") {
		t.Fatalf("expected stable outer pagination, sql=%s", sql)
	}
}

func TestDashboardExactNameExistsIsBoundedWithoutFuzzySearch(t *testing.T) {
	gdb := newDryRunDB(t)
	raw, err := buildNameFastExistsSQL(gdb, "shared-name", []uint{101}, &view.ListOrganizationAndClusterFilter{})
	if err != nil {
		t.Fatalf("build exact-name exists SQL: %v", err)
	}
	sql := raw.SQL

	if strings.Contains(sql, "COUNT") || strings.Contains(sql, "LIKE") {
		t.Fatalf("expected bounded exact existence probe, sql=%s", sql)
	}
	if count := strings.Count(sql, "LIMIT ?"); count != 2 {
		t.Fatalf("expected both exact branches to stop after one row, count=%d sql=%s", count, sql)
	}
	if count := countDashboardArg(raw.Args, 1); count != 2 {
		t.Fatalf("expected limit one for both exact branches, count=%d args=%v", count, raw.Args)
	}
}

func TestDashboardNumericOrgBranchKeepsEveryResolvedOrgID(t *testing.T) {
	gdb := newDryRunDB(t)
	orgIDs := []uint{101, 202}

	raw, err := buildNumericFastListSQL(gdb, "123", orgIDs, &view.ListOrganizationAndClusterFilter{}, 20, 10)
	if err != nil {
		t.Fatalf("build numeric fast list SQL: %v", err)
	}
	sql := raw.SQL

	if !strings.Contains(sql, "WHERE c.org_id IN (?,?)") || countDashboardArg(raw.Args, uint(101)) != 1 || countDashboardArg(raw.Args, uint(202)) != 1 {
		t.Fatalf("expected all resolved internal IDs in the exact-org branch, sql=%s", sql)
	}
	if strings.Contains(sql, "WHERE o.dbaas_org_id") {
		t.Fatalf("expected exact-org branch not to fall back to the slow organization predicate, sql=%s", sql)
	}
	if count := strings.Count(sql, "ORDER BY c.created_at DESC,c.id DESC LIMIT ?"); count != 4 {
		t.Fatalf("expected second-page branch cap to remain offset + limit, count=%d sql=%s", count, sql)
	}
	if count := countDashboardArg(raw.Args, 30); count != 4 {
		t.Fatalf("expected second-page branch cap arg in every branch, count=%d args=%v", count, raw.Args)
	}
}

func TestDashboardNumericSearchOmitsOrgBranchWhenNoOrgMatches(t *testing.T) {
	gdb := newDryRunDB(t)

	raw, err := buildNumericFastListSQL(gdb, "123", nil, &view.ListOrganizationAndClusterFilter{}, 0, 10)
	if err != nil {
		t.Fatalf("build numeric fast list SQL: %v", err)
	}
	sql := raw.SQL

	if strings.Contains(sql, "FORCE INDEX (idx_clusters_org_id_created_at)") || strings.Contains(sql, "WHERE c.org_id") {
		t.Fatalf("expected no exact-org branch without resolved internal IDs, sql=%s", sql)
	}
	if count := strings.Count(sql, "ORDER BY c.created_at DESC,c.id DESC LIMIT ?"); count != 3 {
		t.Fatalf("expected only cluster, parent, and project branches, count=%d sql=%s", count, sql)
	}
}

func countDashboardArg(args []any, want any) int {
	count := 0
	for _, arg := range args {
		if reflect.DeepEqual(arg, want) {
			count++
		}
	}
	return count
}

func TestFetchDashboardPageUsesOneLimitPlusOneList(t *testing.T) {
	tests := []struct {
		name        string
		fetched     int
		wantItems   int
		wantHasMore bool
	}{
		{name: "extra row reports more", fetched: 3, wantItems: 2, wantHasMore: true},
		{name: "full page reports no more without an extra row", fetched: 2, wantItems: 2, wantHasMore: false},
		{name: "short page reports no more", fetched: 1, wantItems: 1, wantHasMore: false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			calls := 0
			gotLimit := 0
			result, err := fetchDashboardPage(2, func(limit int, items *[]*view.DashboardItemExternal) error {
				calls++
				gotLimit = limit
				*items = make([]*view.DashboardItemExternal, tt.fetched)
				return nil
			})
			if err != nil {
				t.Fatalf("fetch dashboard page: %v", err)
			}

			if calls != 1 {
				t.Fatalf("expected one list call, got %d", calls)
			}
			if gotLimit != 3 {
				t.Fatalf("expected limit + 1 rows, got limit %d", gotLimit)
			}
			if len(result.Items) != tt.wantItems {
				t.Fatalf("expected %d returned items, got %d", tt.wantItems, len(result.Items))
			}
			if result.HasMore != tt.wantHasMore {
				t.Fatalf("expected has_more=%t, got %t", tt.wantHasMore, result.HasMore)
			}
			if result.Total != nil {
				t.Fatalf("expected total to remain unset, got %d", *result.Total)
			}
		})
	}
}