package slowquery

import (
	"testing"

	"github.com/stretchr/testify/require"
)

// sampleReadPoolTaskDetails is a redacted production slow-log field value.
// It intentionally excludes the surrounding query, user, address, and plan.
const sampleReadPoolTaskDetails = "{tasks:1, poll_count:{total:2, avg:2, max:2, min:2}, " +
	"dispatch_count:{total:2, max:2, min:2}, " +
	"task_wall_time:{total:611.5µs, avg:611.5µs, max:611.5µs, min:611.5µs}, " +
	"queue_wait:{total:25.8µs, avg:12.9µs, max:13.9µs, min:12µs}, " +
	"wake_wait:{total:389.7µs, avg:389.7µs, max:389.7µs, min:389.7µs}, " +
	"fair_queue:{enabled:true, waited_task_slices:{total:0, avg:0, max:0, min:0}}, " +
	"poll_cpu:{total:183.2µs, avg:91.6µs, max:165.5µs, min:17.7µs}, " +
	"poll_wall:{total:183.5µs, avg:91.8µs, max:165.7µs, min:17.9µs}}"

func allSlowlogDetailStorageColumns() map[string]struct{} {
	seen := make(map[string]struct{})
	for _, spec := range slowlogDetailColumnSpecs {
		for _, name := range spec.required {
			seen[name] = struct{}{}
		}
	}
	return seen
}

func TestBuildAvailableFieldsFromSchema(t *testing.T) {
	columns := map[string]struct{}{
		"digest":             {},
		"query":              {},
		"conn_id":            {},
		"time":               {},
		"query_time":         {},
		"mem_max":            {},
		"request_unit_read":  {},
		"request_unit_write": {},
		"tidb_cpu_time":      {},
	}

	fields := buildAvailableFieldsFromSchema(columns)
	require.Contains(t, fields, "digest")
	require.Contains(t, fields, "connection_id")
	require.Contains(t, fields, "timestamp")
	require.Contains(t, fields, "memory_max")
	require.Contains(t, fields, "ru")
	require.Contains(t, fields, "tidb_cpu_time")
	require.NotContains(t, fields, "ru_v2")
	require.NotContains(t, fields, "tikv_cpu_time")
	require.NotContains(t, fields, "ia_remote_read_segment_count")
	require.NotContains(t, fields, "resource_group")
}

func TestBuildAvailableFieldsFromSchemaAllColumns(t *testing.T) {
	columns := make(map[string]struct{})
	for _, spec := range slowlogAvailableFieldSpecs {
		for _, name := range spec.required {
			columns[name] = struct{}{}
		}
	}

	fields := buildAvailableFieldsFromSchema(columns)
	require.Equal(t, allSlowlogAvailableFieldNames(), fields)
}

func TestBuildDetailSelectColumnsFromSchema(t *testing.T) {
	columns := allSlowlogDetailStorageColumns()
	selectColumns := buildDetailSelectColumnsFromSchema(columns)
	require.Contains(t, selectColumns, "id, time, digest")
	require.Contains(t, selectColumns, "(COALESCE(request_unit_v2, 0)) AS ru_v2")
	require.Contains(t, selectColumns, "tidb_cpu_time")
	require.Contains(t, selectColumns, "tikv_cpu_time")
	require.Contains(t, selectColumns, "ia_remote_read_segment_count")
	require.Contains(t, selectColumns, "ia_remote_read_segment_size")
	require.Contains(t, selectColumns, "ia_remote_read_segment_wait_time")
	require.Contains(t, selectColumns, "binary_plan")
}

func TestBuildDetailSelectColumnsFromSchemaWithoutRUV2(t *testing.T) {
	columns := allSlowlogDetailStorageColumns()
	delete(columns, "request_unit_v2")

	selectColumns := buildDetailSelectColumnsFromSchema(columns)
	require.NotContains(t, selectColumns, "ru_v2")
	require.Contains(t, selectColumns, "query")
}

func TestBuildDetailSelectColumnsFromSchemaWithoutCPUTime(t *testing.T) {
	columns := allSlowlogDetailStorageColumns()
	delete(columns, "tidb_cpu_time")
	delete(columns, "tikv_cpu_time")

	selectColumns := buildDetailSelectColumnsFromSchema(columns)
	require.NotContains(t, selectColumns, "tidb_cpu_time")
	require.NotContains(t, selectColumns, "tikv_cpu_time")
	require.Contains(t, selectColumns, "query")
}

func TestBuildDetailSelectColumnsFromSchemaWithoutIAFields(t *testing.T) {
	columns := allSlowlogDetailStorageColumns()
	delete(columns, "ia_remote_read_segment_count")
	delete(columns, "ia_remote_read_segment_size")
	delete(columns, "ia_remote_read_segment_wait_time")

	selectColumns := buildDetailSelectColumnsFromSchema(columns)
	require.NotContains(t, selectColumns, "ia_remote_read_segment_count")
	require.NotContains(t, selectColumns, "ia_remote_read_segment_size")
	require.NotContains(t, selectColumns, "ia_remote_read_segment_wait_time")
	require.Contains(t, selectColumns, "query")
}

func TestBuildDetailSQL(t *testing.T) {
	slowlogTable := "s3://bucket/deltalake/tenant/cluster/slowlogs"
	params := &DetailParams{
		Timestamp: 1678838400.5,
		ConnID:    12345,
		Digest:    "detail_digest_abc",
		ClusterID: "test_cluster",
	}

	t.Run("includes ru_v2 and cpu time when columns exist", func(t *testing.T) {
		query, aliasResult, err := buildDetailSQL(params, slowlogTable, allSlowlogDetailStorageColumns())
		require.NoError(t, err)
		require.Equal(t, tableAlias(params.ClusterID), aliasResult)
		require.Contains(t, query, "request_unit_v2")
		require.Contains(t, query, "(COALESCE(request_unit_v2, 0)) AS ru_v2")
		require.Contains(t, query, "tidb_cpu_time")
		require.Contains(t, query, "tikv_cpu_time")
		require.Contains(t, query, "binary_plan")
		require.Contains(t, query, "read_pool_task_details")
		require.NotContains(t, query, "decoded_plan")
	})

	t.Run("omits read pool task details for older schemas", func(t *testing.T) {
		columns := allSlowlogDetailStorageColumns()
		delete(columns, "read_pool_task_details")

		query, _, err := buildDetailSQL(params, slowlogTable, columns)
		require.NoError(t, err)
		require.NotContains(t, query, "read_pool_task_details")
	})

	t.Run("omits ru_v2 when column missing", func(t *testing.T) {
		columns := allSlowlogDetailStorageColumns()
		delete(columns, "request_unit_v2")

		query, _, err := buildDetailSQL(params, slowlogTable, columns)
		require.NoError(t, err)
		require.NotContains(t, query, "ru_v2")
		require.NotContains(t, query, "request_unit_v2")
		require.Contains(t, query, "COALESCE(request_unit_read, 0) + COALESCE(request_unit_write, 0)) AS ru")
	})

	t.Run("omits cpu time when columns missing", func(t *testing.T) {
		columns := allSlowlogDetailStorageColumns()
		delete(columns, "tidb_cpu_time")
		delete(columns, "tikv_cpu_time")

		query, _, err := buildDetailSQL(params, slowlogTable, columns)
		require.NoError(t, err)
		require.NotContains(t, query, "tidb_cpu_time")
		require.NotContains(t, query, "tikv_cpu_time")
	})
}

func TestSlowLogToRespItemByFieldsIncludesReadPoolTaskDetails(t *testing.T) {
	item := slowLogToRespItemByFields(
		&SlowLog{ReadPoolTaskDetails: sampleReadPoolTaskDetails},
		[]string{"read_pool_task_details"},
	)

	require.Equal(t, sampleReadPoolTaskDetails, item.ReadPoolTaskDetails)
}