package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"math"
	"testing"
	"time"
)

// TestRoundFloatToIntBoundary 钉住上界的写法。
//
// float64(math.MaxInt64) 存不下 2^63-1, 会被舍入成 2^63 —— 于是 "f > MaxInt64"
// 这个看着像校验的守卫, 恰好放过了唯一越界的那一个值, 而 int64(2^63) 是越界转换
// (结果未定义, amd64 上回绕成 MinInt64): 一个超范围的值静默变成反数存进去。
// 下界不受影响: MinInt64 = -2^63 在 float64 里可以精确表示。
func TestRoundFloatToIntBoundary(t *testing.T) {
	if got, err := roundFloatToInt(0); err != nil || got != 0 {
		t.Errorf("0 应当原样通过, 实际 (%d, %v)", got, err)
	}
	if _, err := roundFloatToInt(float64(math.MaxInt32)); err != nil {
		t.Errorf("MaxInt32 远在范围内, 不该报错: %v", err)
	}
	// 下界是可以精确表示的那个值, 不该被误伤
	if got, err := roundFloatToInt(math.MinInt64); err != nil || got != math.MinInt64 {
		t.Errorf("MinInt64 应当原样通过, 实际 (%d, %v)", got, err)
	}

	// 注意别写 math.MinInt64 - 1 当越界样例: 那个数学值在 float64 里根本没有对应的
	// 格点, 会被舍入回 -2^63 这个合法值。上界那边正好相反 —— MaxInt64 舍入到 2^63,
	// 恰好越过边界, 所以 >= 才是对的。
	for _, f := range []float64{
		math.MaxInt64, // 2^63: 越界, 但 float64(MaxInt64) 也正好等于它
		math.MaxInt64 * 2,
		math.Pow(2, 63) * 4,
		math.Inf(1),
		math.Inf(-1),
		math.NaN(),
	} {
		if got, err := roundFloatToInt(f); err == nil {
			t.Errorf("%v 超出 int64, 应当报错, 实际返回 %d", f, got)
		}
	}
}

// TestCoerceForSQL8ByteByDialect uint64 绑定的 Go 类型必须跟目标列的【声明】一致,
// 而两个方言的声明本就不同 (见 sqlTypeName):
//
//   - MySQL 声明 BIGINT UNSIGNED → 绑 uint64。以前一律绑 int64(u), 高位为 1 的值
//     (包括 VRTX 从 int64 负值搬过来的位模式) 在严格模式下会被判成
//     "Out of range value" 直接报错, 非严格模式则被静默改写。
//   - SQLite 只有有符号的 INTEGER → 绑 int64, 位模式原样存。
//
// 共同的保证是【位模式不变】; 两种绑定在各自的列里都必须无损。
func TestCoerceForSQL8ByteByDialect(t *testing.T) {
	col := Column{Name: "u", Type: types.FieldType_Uint64}

	cases := []struct {
		name       string
		in         any
		wantMySQL  any
		wantSQLite any
	}{
		{"uint64 上界", uint64(math.MaxUint64), uint64(math.MaxUint64), int64(-1)},
		{"int64 负数 (位模式)", int64(-1), uint64(math.MaxUint64), int64(-1)},
		{"普通值", int64(42), uint64(42), int64(42)},
		{"2^63 (无符号档才装得下)", uint64(math.MaxInt64) + 1, uint64(math.MaxInt64) + 1, int64(math.MinInt64)},
	}
	for _, c := range cases {
		gotMySQL, err := coerceForSQL(col, c.in, true)
		if err != nil {
			t.Errorf("%s: MySQL 侧不该报错: %v", c.name, err)
		} else if gotMySQL != c.wantMySQL {
			t.Errorf("%s: MySQL 侧 = %v (%T), 期望 %v (%T)",
				c.name, gotMySQL, gotMySQL, c.wantMySQL, c.wantMySQL)
		}

		gotSQLite, err := coerceForSQL(col, c.in, false)
		if err != nil {
			t.Errorf("%s: SQLite 侧不该报错: %v", c.name, err)
		} else if gotSQLite != c.wantSQLite {
			t.Errorf("%s: SQLite 侧 = %v (%T), 期望 %v (%T)",
				c.name, gotSQLite, gotSQLite, c.wantSQLite, c.wantSQLite)
		}
	}

	// 存量旧表 (Uint64+TimeKind) 走的还是"按纳秒还原成时间文本"那条路, 与方言无关 ——
	// 别被上面的方言分支带走, 也别把它跟新 DateTime 列的毫秒路搞混。
	timeCol := Column{Name: "t", Type: types.FieldType_Uint64, TimeKind: "datetime"}
	for _, mysql := range []bool{true, false} {
		v, err := coerceForSQL(timeCol, int64(0), mysql)
		if err != nil {
			t.Fatalf("时间列 (mysql=%v) 不该报错: %v", mysql, err)
		}
		if _, ok := v.(string); !ok {
			t.Errorf("时间列 (mysql=%v) 应当还原成时间文本, 实际 %T", mysql, v)
		}
	}
}

// TestCoerceForSQLTimeTwoPaths 读路径的两条时间路必须按【列的 FieldType】分岔:
//
//   - FieldType_DateTime (新表): 值按 Unix【毫秒】解释;
//   - Uint64+TimeKind (存量旧表): 值按 Unix【纳秒】解释。
//
// 同一大小的整数, 两条路解出的时间差了六个数量级 —— 走错路不是报错而是【静默给出
// 一个完全错误的时间】, 所以判别依据是列的 FieldType, 不能只看 TimeKind。
func TestCoerceForSQLTimeTwoPaths(t *testing.T) {
	// 2024-03-05 12:34:56.123 UTC = 1709642096123 毫秒 = 1709642096123000000 纳秒。
	const ms = int64(1709642096123)
	const ns = int64(1709642096123000000)

	dtCol := Column{Name: "ctime", Type: types.FieldType_DateTime, TimeKind: "datetime"}
	got, err := coerceForSQL(dtCol, uint64(ms), false)
	if err != nil {
		t.Fatalf("DateTime 列不该报错: %v", err)
	}
	if s, ok := got.(string); !ok || s != "2024-03-05 12:34:56.123" {
		t.Errorf("DateTime 列应按毫秒还原, 实际 %v (%T)", got, got)
	}

	legacyCol := Column{Name: "ctime", Type: types.FieldType_Uint64, TimeKind: "datetime"}
	got, err = coerceForSQL(legacyCol, uint64(ns), false)
	if err != nil {
		t.Fatalf("旧 Uint64 时间列不该报错: %v", err)
	}
	if s, ok := got.(string); !ok || s != "2024-03-05 12:34:56.123000" {
		t.Errorf("旧 Uint64 时间列应按纳秒还原 (微秒文本), 实际 %v (%T)", got, got)
	}

	// DATE 语义只留日期部分; time.Time 输入直接取毫秒 (MySQL parseTime 驱动给得出)。
	dayCol := Column{Name: "day", Type: types.FieldType_DateTime, TimeKind: "date"}
	got, err = coerceForSQL(dayCol, time.UnixMilli(ms).UTC(), false)
	if err != nil {
		t.Fatalf("DATE 列不该报错: %v", err)
	}
	if s, ok := got.(string); !ok || s != "2024-03-05" {
		t.Errorf("DATE 列应只留日期, 实际 %v (%T)", got, got)
	}

	// 非时间语义的普通整数列 (无 TimeKind) 不受影响, 原样按位模式走。
	plainCol := Column{Name: "u", Type: types.FieldType_Uint64}
	got, err = coerceForSQL(plainCol, uint64(ns), false)
	if err != nil {
		t.Fatalf("普通 uint64 列不该报错: %v", err)
	}
	if got != int64(ns) {
		t.Errorf("普通 uint64 列应原样绑 int64 位模式, 实际 %v (%T)", got, got)
	}
}

// TestCoerceForVRTXDateTime 写路径: SQL 侧时间值收编成 FieldType_DateTime 列认得的形态。
//
// 约定与 gormvrtx / SQL driver 对齐: time.Time 直传 (引擎编码层截毫秒), 文本按 UTC
// 解析, 整数按【毫秒】; 零时间走可空列写 NULL、非空列落零毫秒。
func TestCoerceForVRTXDateTime(t *testing.T) {
	tm := time.Date(2024, 3, 5, 12, 34, 56, 123456789, time.UTC)

	// time.Time 原样直传, 由引擎 codec 自行截毫秒。
	got, err := coerceForVRTX(types.FieldType_DateTime, 0, tm, "ctime", true, false)
	if err != nil {
		t.Fatalf("time.Time 不该报错: %v", err)
	}
	if x, ok := got.(time.Time); !ok || !x.Equal(tm) {
		t.Errorf("time.Time 应原样直传, 实际 %v (%T)", got, got)
	}

	// 文本按 UTC 解析 (不用本地时区)。
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, "2024-03-05 12:34:56.123", "ctime", true, false)
	if err != nil {
		t.Fatalf("文本不该报错: %v", err)
	}
	if x, ok := got.(time.Time); !ok || x.UnixMilli() != 1709642096123 {
		t.Errorf("文本应解析成 UTC time.Time (ms=1709642096123), 实际 %v (%T)", got, got)
	}

	// 整数按毫秒原样落 (不是纳秒 —— 那是旧 Uint64 盘面的事)。
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, int64(1709642096123), "ctime", true, false)
	if err != nil {
		t.Fatalf("毫秒整数不该报错: %v", err)
	}
	if x, ok := got.(uint64); !ok || x != 1709642096123 {
		t.Errorf("毫秒整数应按毫秒落 uint64, 实际 %v (%T)", got, got)
	}

	// 零时间: 可空列 → NULL; 非空列 → 零毫秒。
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, time.Time{}, "ctime", true, true)
	if err != nil || got != nil {
		t.Errorf("零时间走可空列应写 NULL, 实际 %v (%v)", got, err)
	}
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, time.Time{}, "ctime", true, false)
	if err != nil {
		t.Fatalf("零时间走非空列不该报错: %v", err)
	}
	if x, ok := got.(uint64); !ok || x != 0 {
		t.Errorf("零时间走非空列应落零毫秒, 实际 %v (%T)", got, got)
	}

	// nil 的约定不变: 可空列 NULL, 非空列零毫秒。
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, nil, "ctime", true, true)
	if err != nil || got != nil {
		t.Errorf("nil 走可空列应写 NULL, 实际 %v (%v)", got, err)
	}
	got, err = coerceForVRTX(types.FieldType_DateTime, 0, nil, "ctime", true, false)
	if err != nil {
		t.Fatalf("nil 走非空列不该报错: %v", err)
	}
	if x, ok := got.(uint64); !ok || x != 0 {
		t.Errorf("nil 走非空列应落零毫秒 (zeroForVRTX), 实际 %v (%T)", got, got)
	}
}

// TestCoerceForSQLUnsignedSmall 2 / 4 字节无符号档出库时必须给【无符号数值】。
//
// 与 8 字节档同一个道理 (见 TestCoerceForSQL8ByteByDialect), 但 16/32 位不需要按方言分岔:
// 目标列在 MySQL 侧声明成 SMALLINT UNSIGNED / INT UNSIGNED, SQLite 侧是 INTEGER ——
// 0..65535 与 0..4294967295 两边都整个装得下, 给无符号值两个方言都对。
// 反过来把位模式当 int64 原样送出去 (0x8278 读出来是 -32136), MySQL 会判 Out of range:
// 严格模式直接报错, 非严格模式截成 0 —— 静默改数据。
func TestCoerceForSQLUnsignedSmall(t *testing.T) {
	cases := []struct {
		name string
		ft   types.FieldType
		in   any
		want int64
	}{
		{"uint16 上半区位模式", types.FieldType_Uint16, int16(-32136), 33400}, // 0x8278
		{"uint16 普通值", types.FieldType_Uint16, int16(3306), 3306},
		{"uint16 上界位模式", types.FieldType_Uint16, int16(-1), 65535},
		{"uint32 上半区位模式", types.FieldType_Uint32, int32(-1294967296), 3000000000},
		{"uint32 上界位模式", types.FieldType_Uint32, int32(-1), 4294967295},
	}
	for _, c := range cases {
		col := Column{Name: "u", Type: c.ft}
		for _, mysql := range []bool{true, false} {
			got, err := coerceForSQL(col, c.in, mysql)
			if err != nil {
				t.Errorf("%s (mysql=%v): 不该报错: %v", c.name, mysql, err)
				continue
			}
			if got != c.want {
				t.Errorf("%s (mysql=%v): = %v (%T), 期望 %d", c.name, mysql, got, got, c.want)
			}
		}
	}
}

// TestCoerceForVRTXUnsignedSmall 2 / 4 字节无符号档"装得下"的判据是【位模式】, 不是
// int16 / int32 的正数区。
//
// 真实踩到: MySQL 的 SMALLINT UNSIGNED 存了端口号 33400, 迁进 VRTX 的 uint16 列时被判
// "值 33400 超出 int16" —— 可 33400 只占 16 位 (0x8278), 位模式完全装得下。按有符号区间判
// 等于给无符号列凭空加了一条"上界只到 32767"的约束, 把 32768..65535 这半个无符号区整片
// 判成脏数据; 而 -32136 只是同一个位模式的另一种读法, 两个方向都得收。
func TestCoerceForVRTXUnsignedSmall(t *testing.T) {
	const col = "port"
	cases := []struct {
		name string
		ft   types.FieldType
		in   any
		want any
	}{
		// 无符号上半区: 合法, 落成负的位模式载体 (读取侧 DecodeValue 也回 int16/int32)
		{"uint16 上半区", types.FieldType_Uint16, int64(33400), int16(-32136)},
		{"uint16 上界", types.FieldType_Uint16, int64(math.MaxUint16), int16(-1)},
		{"uint16 负位模式 (回读 / SQLite INTEGER)", types.FieldType_Uint16, int16(-1), int16(-1)},
		{"uint16 普通值", types.FieldType_Uint16, int64(3306), int16(3306)},
		{"uint32 上半区", types.FieldType_Uint32, int64(3000000000), int32(-1294967296)},
		{"uint32 上界", types.FieldType_Uint32, int64(math.MaxUint32), int32(-1)},
		{"uint32 普通值", types.FieldType_Uint32, int64(42), int32(42)},
	}
	for _, c := range cases {
		got, err := coerceForVRTX(c.ft, 0, c.in, col, true, false)
		if err != nil {
			t.Errorf("%s: 不该报错: %v", c.name, err)
			continue
		}
		if got != c.want {
			t.Errorf("%s: = %v (%T), 期望 %v (%T)", c.name, got, got, c.want, c.want)
		}
	}

	// 放不下位模式才是真放不下 —— 这条闸门不能因为放宽上界就整个拆掉。
	for _, c := range []struct {
		name string
		ft   types.FieldType
		in   any
	}{
		{"uint16 超出 16 位", types.FieldType_Uint16, int64(math.MaxUint16) + 1},
		{"uint16 低于 -32768", types.FieldType_Uint16, int64(math.MinInt16) - 1},
		{"uint32 超出 32 位", types.FieldType_Uint32, int64(math.MaxUint32) + 1},
	} {
		if got, err := coerceForVRTX(c.ft, 0, c.in, col, true, false); err == nil {
			t.Errorf("%s: 应当报错, 实际返回 %v", c.name, got)
		}
	}
}