package dbhubsdk

import (
	"context"
	"testing"
	"time"

	"github.com/pingcap/tidb/pkg/parser"
	_ "github.com/pingcap/tidb/pkg/parser/test_driver"
	"github.com/stretchr/testify/require"
)

func TestConvertCellValue_Bit(t *testing.T) {
	tests := []struct {
		name    string
		value   interface{}
		colType string
		want    string
	}{
		{"BIT(1) value 0 (b'0')", []byte{0x00}, "BIT", "0"},
		{"BIT(1) value 1 (b'1')", []byte{0x01}, "BIT", "1"},
		{"BIT(8) value 10", []byte{0x0A}, "BIT", "10"},
		{"BIT(8) value 255", []byte{0xFF}, "BIT", "255"},
		{"BIT(16) value 258", []byte{0x01, 0x02}, "BIT", "258"},
		{"BIT(16) value 65535", []byte{0xFF, 0xFF}, "BIT", "65535"},
		{"BIT(64) max uint64", []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}, "BIT", "18446744073709551615"},
		{"BIT empty bytes falls back to 0", []byte{}, "BIT", "0"},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := convertCellValue(tc.value, tc.colType)
			require.Equal(t, tc.want, got)
		})
	}
}

func TestConvertCellValue_Regressions(t *testing.T) {
	// nil scan target -> empty string
	require.Equal(t, "", convertCellValue(nil, "BIT"))

	// TINYINT(1) is scanned as int64 by the driver and must render as a number.
	require.Equal(t, "0", convertCellValue(int64(0), "TINYINT"))
	require.Equal(t, "1", convertCellValue(int64(1), "TINYINT"))

	// VARCHAR valid UTF-8 is preserved.
	require.Equal(t, "hello", convertCellValue([]byte("hello"), "VARCHAR"))

	// DATETIME is formatted, not passed through as a number.
	ts := time.Date(2026, 6, 18, 10, 30, 0, 0, time.UTC)
	require.Equal(t, "2026-06-18 10:30:00", convertCellValue(ts, "DATETIME"))

	// Non-UTF-8 BINARY payload is rendered as uppercase hex, not silently passed through.
	require.Equal(t, "0XFF", convertCellValue([]byte{0xFF}, "BINARY"))
}

// TestNonTransactionalDML_StmtTypeAndQueryFlag covers BATCH ON ... DRY RUN {QUERY|STATISTICS}
// statements, which the TiDB parser produces as *ast.NonTransactionalDMLStmt.
// DRY RUN variants return a result set, so they must be classified as SELECT and
// treated as a query (db.QueryContext); otherwise the rows are discarded by ExecContext.
// The non-DRY-RUN variant must classify as the underlying DML (INSERT/UPDATE/DELETE).
func TestNonTransactionalDML_StmtTypeAndQueryFlag(t *testing.T) {
	ctx := context.Background()
	cases := []struct {
		name     string
		sql      string
		wantType StmtType
		wantQry  bool
	}{
		{
			name:    "dry run query",
			sql:     `BATCH ON batch_on_test.dws_minute_sensor_data_stage.data_timestamp LIMIT 5 DRY RUN QUERY INSERT INTO batch_on_test.dws_minute_sensor_data (id) SELECT id FROM batch_on_test.dws_minute_sensor_data_stage`,
			wantType: SELECT,
			wantQry:  true,
		},
		{
			name:    "dry run split dml",
			sql:     `BATCH ON t.c LIMIT 5 DRY RUN INSERT INTO t (a) SELECT a FROM s`,
			wantType: SELECT,
			wantQry:  true,
		},
		{
			name:     "no dry run, insert",
			sql:      `BATCH ON t.c LIMIT 5 INSERT INTO t (a) SELECT a FROM s`,
			wantType: INSERT,
			wantQry:  false,
		},
		{
			// Mirrors the user-reported case: BATCH ON <db>.<table>.<col> LIMIT N
			// DRY RUN QUERY INSERT INTO ... SELECT ... USE INDEX (...).
			name:    "user reproduction: batch on qualified column with use index",
			sql:     "BATCH ON batch_on_test.dws_minute_sensor_data_stage.data_timestamp\nLIMIT 5\nDRY RUN QUERY\nINSERT INTO batch_on_test.dws_minute_sensor_data\n(id, partition_date, data_timestamp, sensor_value)\nSELECT id, partition_date, data_timestamp, sensor_value\nFROM batch_on_test.dws_minute_sensor_data_stage\nUSE INDEX (idx_data_timestamp)\nWHERE partition_date = '2026-07-29'\n  AND data_timestamp >= 1785283200000\n  AND data_timestamp < 1785369600000",
			wantType: SELECT,
			wantQry:  true,
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			stmt, err := parser.New().ParseOneStmt(tc.sql, "", "")
			require.NoError(t, err)
			require.Equal(t, tc.wantType, GetStmtType(ctx, stmt), "GetStmtType mismatch for %s", tc.name)
			require.Equal(t, tc.wantQry, IsQueryStmt(ctx, stmt), "IsQueryStmt mismatch for %s", tc.name)
		})
	}
}