package sqls

import (
	"strconv"
	"testing"

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

func TestIsPureDigits(t *testing.T) {
	tests := []struct {
		name     string
		input    string
		expected bool
	}{
		{
			name:     "Empty string",
			input:    "",
			expected: false,
		},
		{
			name:     "Pure digits",
			input:    "123456",
			expected: true,
		},
		{
			name:     "Digits with letter",
			input:    "123a456",
			expected: false,
		},
		{
			name:     "Single digit",
			input:    "5",
			expected: true,
		},
		{
			name:     "String with spaces",
			input:    "123 456",
			expected: false,
		},
		{
			name:     "String with special characters",
			input:    "123-456",
			expected: false,
		},
		{
			name:     "Zero",
			input:    "0",
			expected: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := isPureDigits(tt.input)
			require.Equal(t, tt.expected, result)
		})
	}
}

func TestParseStringToUint64(t *testing.T) {
	tests := []struct {
		name     string
		input    string
		expected uint64
	}{
		{
			name:     "Valid number",
			input:    "123456",
			expected: 123456,
		},
		{
			name:     "Zero",
			input:    "0",
			expected: 0,
		},
		{
			name:     "Max uint64",
			input:    "18446744073709551615",
			expected: 18446744073709551615,
		},
		{
			name:     "Invalid string",
			input:    "abc",
			expected: 0,
		},
		{
			name:     "Empty string",
			input:    "",
			expected: 0,
		},
		{
			name:     "Negative number (invalid for uint64)",
			input:    "-123",
			expected: 0,
		},
		{
			name:     "Number with spaces",
			input:    "123 456",
			expected: 0,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := parseStringToUint64(tt.input)
			require.Equal(t, tt.expected, result)
		})
	}
}

func TestParseInt64ToUint64(t *testing.T) {
	tests := []struct {
		name     string
		input    int64
		expected uint64
	}{
		{
			name:     "Positive number",
			input:    123456,
			expected: 123456,
		},
		{
			name:     "Zero",
			input:    0,
			expected: 0,
		},
		{
			name:     "Max int64",
			input:    9223372036854775807,
			expected: 9223372036854775807,
		},
		{
			name:     "Negative number",
			input:    -123,
			expected: 0,
		},
		{
			name:     "Min int64",
			input:    -9223372036854775808,
			expected: 0,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := parseInt64ToUint64(tt.input)
			require.Equal(t, tt.expected, result)
		})
	}
}

func TestTableAlias(t *testing.T) {
	tests := []struct {
		name      string
		clusterID string
		table     string
		expected  string
	}{
		{
			name:      "Simple cluster and table",
			clusterID: "cluster1",
			table:     "table1",
			expected:  "table1_cluster1",
		},
		{
			name:      "Cluster ID with hyphens",
			clusterID: "cluster-1-2",
			table:     "table1",
			expected:  "table1_cluster_1_2",
		},
		{
			name:      "Table with hyphens",
			clusterID: "cluster1",
			table:     "table-1",
			expected:  "table_1_cluster1",
		},
		{
			name:      "Both with hyphens",
			clusterID: "cluster-1",
			table:     "table-1",
			expected:  "table_1_cluster_1",
		},
		{
			name:      "Empty cluster ID",
			clusterID: "",
			table:     "table1",
			expected:  "table1_",
		},
		{
			name:      "Empty table",
			clusterID: "cluster1",
			table:     "",
			expected:  "_cluster1",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := tableAlias(tt.clusterID, tt.table)
			require.Equal(t, tt.expected, result)
		})
	}
}

func TestFieldAdapt(t *testing.T) {
	tests := []struct {
		name     string
		field    string
		expected string
	}{
		{
			name:     "Mapped field - summary_begin_time",
			field:    "summary_begin_time",
			expected: "min(epoch(SUMMARY_BEGIN_TIME)) as summary_begin_time",
		},
		{
			name:     "Mapped field - digest_text",
			field:    "digest_text",
			expected: "any_value(digest_text) as digest_text",
		},
		{
			name:     "Unmapped field",
			field:    "unknown_field",
			expected: "unknown_field",
		},
		{
			name:     "Empty field",
			field:    "",
			expected: "",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := fieldAdapt(tt.field, "")
			require.Equal(t, tt.expected, result)
		})
	}
}

func TestConvertDeltalakeFields(t *testing.T) {
	tests := []struct {
		name     string
		fields   string
		expected string
	}{
		{
			name:     "Empty fields",
			fields:   "",
			expected: "id, conn_id, digest, time, query_time, mem_max",
		},
		{
			name:     "Fields with ru",
			fields:   "ru,digest",
			expected: "id, conn_id, digest, time, query_time, mem_max, (request_unit_read + request_unit_write) AS ru",
		},
		{
			name:     "Fields with toRemoveFieldMap fields",
			fields:   "connection_id,digest,timestamp",
			expected: "id, conn_id, digest, time, query_time, mem_max",
		},
		{
			name:     "Mixed fields",
			fields:   "ru,connection_id,query_time,test_field",
			expected: "id, conn_id, digest, time, query_time, mem_max, (request_unit_read + request_unit_write) AS ru, test_field",
		},
		{
			name:     "Single field",
			fields:   "test_field",
			expected: "id, conn_id, digest, time, query_time, mem_max, test_field",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := convertDeltalakeFields(tt.fields)
			// The result might have trailing comma, so we check if it contains expected parts
			require.Contains(t, result, "id")
			require.Contains(t, result, "conn_id")
			require.Contains(t, result, "digest")
			require.Contains(t, result, "time")
			require.Contains(t, result, "query_time")
			require.Contains(t, result, "mem_max")

			// Check specific cases
			if tt.fields != "" {
				if strconv.Quote(tt.fields) != `""` {
					// If fields contain "ru", check for the special handling
					if contains(tt.fields, "ru") {
						require.Contains(t, result, "request_unit_read + request_unit_write")
					}
				}
			}
		})
	}
}

func contains(s, substr string) bool {
	return len(s) >= len(substr) && (s == substr || len(s) > len(substr) && (s[:len(substr)] == substr || s[len(s)-len(substr):] == substr || containsMiddle(s, substr)))
}

func containsMiddle(s, substr string) bool {
	for i := 0; i <= len(s)-len(substr); i++ {
		if s[i:i+len(substr)] == substr {
			return true
		}
	}
	return false
}