package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"fmt"
	"strconv"
	"strings"
	"testing"
)

// ============================ SQL 类型 ↔ VRTX 类型 ============================
//
// 这一层是迁移的"翻译表", 出错的后果不是报错而是【静默改变列的语义】:
//   - BLOB 认成 TEXT → 二进制内容进了文本列;
//   - FLOAT 在 SQLite/MySQL 上按同一档收 → 单精度列被当成双精度 (或反过来);
//   - DECIMAL(18,2) 只读类型名不读 (p,s) → 小数位被抹平, 金额变成整数。
//
// 所以这里既钉住逐条映射, 也钉住"声明出去再读回来还是同一个类型"这个往返性质。

func TestInferSQLType(t *testing.T) {
	cases := []struct {
		decl  string
		mysql bool
		ft    types.FieldType
		param byte
	}{
		// 整数: 有符号 / 无符号分档, SQLite 的 INTEGER 一律按 8 字节收
		{"TINYINT", true, types.FieldType_Int8, 0},
		{"SMALLINT", true, types.FieldType_Int16, 0},
		{"INT", true, types.FieldType_Int32, 0},
		{"MEDIUMINT", true, types.FieldType_Int32, 0},
		{"BIGINT", true, types.FieldType_Int64, 0},
		{"INTEGER", false, types.FieldType_Int64, 0},
		{"TINYINT UNSIGNED", true, types.FieldType_Uint8, 0},
		{"SMALLINT UNSIGNED", true, types.FieldType_Uint16, 0},
		{"INT UNSIGNED", true, types.FieldType_Uint32, 0},
		{"BIGINT UNSIGNED", true, types.FieldType_Uint64, 0},

		// 浮点: MySQL 的 FLOAT 就是 4 字节, SQLite 只有 REAL 亲和性 (8 字节)
		{"FLOAT", true, types.FieldType_Float32, 0},
		{"FLOAT", false, types.FieldType_Float64, 0},
		{"DOUBLE", true, types.FieldType_Float64, 0},
		{"REAL", false, types.FieldType_Float64, 0},

		// 定点: (p,s) 必须读进来 —— 只读类型名会把小数位抹平
		{"DECIMAL(18,2)", true, types.FieldType_Decimal, 2},
		{"DECIMAL(10,0)", true, types.FieldType_Decimal, 0},
		{"NUMERIC(20,6)", false, types.FieldType_Decimal, 6},
		// p 与 s 是【两个不同的上界】: p ≤ 38 (128 位系数的十进制容量),
		// s ≤ 31 (Flag 低 5 位)。以前拿 31 去卡 p, 于是 DECIMAL(38,2) 被误判成
		// "表达不了"而退成 float64 —— 那是被写错的上限限制的, 不是被能力限制的。
		{"DECIMAL(38,2)", true, types.FieldType_Decimal, 2},
		{"DECIMAL(38,31)", true, types.FieldType_Decimal, 31},
		// 真的超出系数容量 → 退回 Float64 (与改动前一致), 不假装精确; 但要报出来 (degradeNote)
		{"DECIMAL(65,30)", true, types.FieldType_Float64, 0},
		{"DECIMAL(39,0)", true, types.FieldType_Float64, 0},
		// s 超 31: 系数装得下, 但小数位无处表达
		{"DECIMAL(38,35)", true, types.FieldType_Float64, 0},

		// 其余新类型
		{"BOOLEAN", false, types.FieldType_Bool, 0},
		{"BOOL", true, types.FieldType_Bool, 0},
		{"YEAR", true, types.FieldType_Year, 0},
		{"UUID", true, types.FieldType_UUID, 0},

		// 二进制大对象 vs 文本: 这是本次新增的区分
		{"BLOB", false, types.FieldType_Blob, 0},
		{"TINYBLOB", true, types.FieldType_Blob, 0},
		{"MEDIUMBLOB", true, types.FieldType_Blob, 0},
		{"LONGBLOB", true, types.FieldType_Blob, 0},
		{"VARBINARY(64)", true, types.FieldType_Blob, 0},
		{"TEXT", false, types.FieldType_Dynamicbyte, 0},
		{"LONGTEXT", true, types.FieldType_Dynamicbyte, 0},

		// 定长串: 按字节容量【向上】取宽度档, 档位号写进列级参数 (Flag 低 5 位)。
		// 返回的 param 不再是 0 —— 它就是"这一列多宽"的载体 (见 types.FieldSize)。
		{"VARCHAR(8)", true, types.FieldType_String, types.WidthCodeFor(8)},
		{"VARCHAR(200)", true, types.FieldType_String, types.WidthCodeFor(200)},
		{"CHAR(32)", true, types.FieldType_String, types.WidthCodeFor(32)},
		// 超过旧的 256 字节天花板仍可以定长: 2048 以内都保得住"可索引"
		{"VARCHAR(1000)", true, types.FieldType_String, types.WidthCodeFor(1000)},
		// 只有超过 MaxInlineColumnWidth 才落变长列
		{"VARCHAR(4000)", true, types.FieldType_Dynamicbyte, 0},

		// 时间列落引擎的专用 datetime 类型 (Unix 毫秒), 不是动态列也不是普通整数
		{"DATETIME", true, types.FieldType_DateTime, 0},
		{"DATE", false, types.FieldType_DateTime, 0},
		{"TIMESTAMP", true, types.FieldType_DateTime, 0},
	}
	for _, c := range cases {
		ft, param := inferSQLType(c.decl, c.mysql)
		if ft != c.ft || param != c.param {
			t.Errorf("inferSQLType(%q, mysql=%v) = (%v, %d), 期望 (%v, %d)",
				c.decl, c.mysql, ft, param, c.ft, c.param)
		}
	}
}

// TestMySQLColumnDecl 钉住"字符列的容量按字节算"这条规则 (information_schema 里
// CHARACTER_OCTET_LENGTH 那一列)。
//
// 这条是真实踩出来的: resource_info.resAlias 是 varchar(50) utf8mb4, 库里存了 73 字节
// 的中文别名。老代码拿 50 (字符数) 去取档得到 64 字节, 整表迁移失败 —— 而 50 个字符在
// utf8mb4 下最多是 200 字节, 只有拿 200 去取档才装得下。
func TestMySQLColumnDecl(t *testing.T) {
	cases := []struct {
		name     string
		dataType string
		// information_schema 给的三组长度: 字节容量 / 精度 / 小数位
		charBytes, numPrec, numScale int64
		want                         types.FieldType
		wantParam                    byte
	}{
		// varchar(50) utf8mb4 → octet 200 → 224 字节档 (老算法 50 → 64 字节 装不下中文;
		// 中间的过渡算法给 256 档, 现在能按声明精确到 224)
		{"utf8mb4 varchar(50)", "varchar", 200, 0, 0, types.FieldType_String, types.WidthCodeFor(200)},
		// varchar(12) utf8mb4 → octet 48 → 48 字节档
		{"utf8mb4 varchar(12)", "varchar", 48, 0, 0, types.FieldType_String, types.WidthCodeFor(48)},
		// 单字节字符集不背 4 倍: latin1 的 varchar(50) 就是 50 字节 (取 64 档)
		{"latin1 varchar(50)", "varchar", 50, 0, 0, types.FieldType_String, types.WidthCodeFor(50)},
		// 正好卡在旧天花板上: varchar(64) utf8mb4 = 256 字节
		{"utf8mb4 varchar(64)", "varchar", 256, 0, 0, types.FieldType_String, types.WidthCodeFor(256)},
		// 超过旧天花板不再落变长列: 320 字节仍在定长档内, 保住"可索引"
		{"utf8mb4 varchar(65)", "varchar", 260, 0, 0, types.FieldType_String, types.WidthCodeFor(260)},
		// 直到 MaxInlineColumnWidth (2048) 为止都还能定长
		{"utf8mb4 varchar(512)", "varchar", 2048, 0, 0, types.FieldType_String, types.WidthCodeFor(2048)},
		// 超出 2048 才落变长列 (变长只是没索引, 不会丢数据)
		{"utf8mb4 varchar(600)", "varchar", 2400, 0, 0, types.FieldType_Dynamicbyte, 0},
		{"utf8mb4 char(1)", "char", 4, 0, 0, types.FieldType_String, types.WidthCodeFor(4)},
		{"utf8mb4 text", "text", 65535, 0, 0, types.FieldType_Dynamicbyte, 0},
		{"blob", "blob", 65535, 0, 0, types.FieldType_Blob, 0},
		// 定点: (p,s) 要读进来, 小数位不能被抹平
		{"decimal(18,2)", "decimal", 20, 18, 2, types.FieldType_Decimal, 2},
		// 长度信息为 0 (数值/时间列没有 octet length) 时退回纯类型名
		{"int", "int", 0, 10, 0, types.FieldType_Int32, 0},
		{"datetime", "datetime", 0, 0, 0, types.FieldType_DateTime, 0},
	}
	for _, c := range cases {
		decl := mysqlColumnDecl(c.dataType, false, c.charBytes, c.numPrec, c.numScale)
		ft, param := inferSQLType(decl, true)
		if ft != c.want || param != c.wantParam {
			t.Errorf("%s: %q → (%v, %d), 期望 (%v, %d)",
				c.name, decl, ft, param, c.want, c.wantParam)
		}
	}
}

// TestMySQLUnsigned 无符号标志只能从 COLUMN_TYPE 读。
//
// MySQL 对 "bigint unsigned" 报的 DATA_TYPE 就是 "bigint" —— 只看 DATA_TYPE
// 等于把无符号这件事整个丢掉, 于是无符号列全被当有符号收:
//   - BIGINT UNSIGNED 的 2^63 读回来是负数;
//   - INT UNSIGNED 的上界只剩一半。
//
// EXTRA 也靠不住 (只在显式 zerofill 时才有值), COLUMN_TYPE 两处都覆盖。
func TestMySQLUnsigned(t *testing.T) {
	cases := []struct {
		columnType string
		want       bool
	}{
		{"bigint(20) unsigned", true},
		{"int(10) unsigned zerofill", true}, // zerofill 隐含 unsigned
		{"bigint unsigned", true},
		{"BIGINT UNSIGNED", true}, // information_schema 大小写随版本
		{"bigint(20)", false},
		{"int(11)", false},
		{"varchar(50)", false},
		{"decimal(18,2)", false},
		{"datetime", false},
	}
	for _, c := range cases {
		if got := mysqlUnsigned(c.columnType); got != c.want {
			t.Errorf("mysqlUnsigned(%q) = %v, 期望 %v", c.columnType, got, c.want)
		}
	}
}

// TestMySQLBool BOOLEAN 只能从 COLUMN_TYPE 认出来。
//
// MySQL 的 BOOLEAN 就是 tinyint(1): information_schema 报的 DATA_TYPE 是
// "tinyint"、COLUMN_TYPE 是 "tinyint(1)" —— 只看 DATA_TYPE 会走 TINYINT 分支
// 收成 Int8, 而 sqltype.go 的 "BOOLEAN"/"BOOL" 分支对 MySQL 源永远不可达。
// 宽度不是 1 的 tinyint 是真·小整数, tinyint(1) unsigned 能装 0..255,
// 都不是真假语义, 不特判。
func TestMySQLBool(t *testing.T) {
	cases := []struct {
		name                         string
		dataType, columnType         string
		charBytes, numPrec, numScale int64
		want                         types.FieldType
	}{
		// BOOLEAN 列: DATA_TYPE 只报 tinyint, 靠 COLUMN_TYPE 特判成 Bool
		{"boolean", "tinyint", "tinyint(1)", 0, 3, 0, types.FieldType_Bool},
		// 宽度不是 1 的 tinyint 是真·小整数
		{"tinyint(4)", "tinyint", "tinyint(4)", 0, 3, 0, types.FieldType_Int8},
		// tinyint(1) unsigned 能装 0..255, 不是真假语义
		{"tinyint(1) unsigned", "tinyint", "tinyint(1) unsigned", 0, 3, 0, types.FieldType_Uint8},
	}
	for _, c := range cases {
		decl := mysqlColumnDecl(c.dataType, mysqlUnsigned(c.columnType), c.charBytes, c.numPrec, c.numScale)
		if mysqlBool(c.columnType) {
			decl = "boolean"
		}
		ft, _ := inferSQLType(decl, true)
		if ft != c.want {
			t.Errorf("%s: COLUMN_TYPE=%q → decl %q → %v, 期望 %v",
				c.name, c.columnType, decl, ft, c.want)
		}
	}
}

// TestMySQLColumnDeclUnsigned 无符号后缀必须真的走到 inferSQLType,
// 而且不能污染第一个括号里的容量 / 精度参数。
func TestMySQLColumnDeclUnsigned(t *testing.T) {
	cases := []struct {
		name                         string
		dataType, columnType         string
		charBytes, numPrec, numScale int64
		want                         types.FieldType
		wantParam                    byte
	}{
		// DATA_TYPE 丢了 unsigned, 补上 COLUMN_TYPE 才能收成无符号档
		{"bigint unsigned", "bigint", "bigint(20) unsigned", 0, 20, 0, types.FieldType_Uint64, 0},
		{"bigint signed", "bigint", "bigint(20)", 0, 20, 0, types.FieldType_Int64, 0},
		{"int unsigned", "int", "int(10) unsigned", 0, 10, 0, types.FieldType_Uint32, 0},
		{"int signed", "int", "int(11)", 0, 10, 0, types.FieldType_Int32, 0},
		{"smallint unsigned", "smallint", "smallint(5) unsigned", 0, 5, 0, types.FieldType_Uint16, 0},
		{"tinyint unsigned", "tinyint", "tinyint(3) unsigned", 0, 3, 0, types.FieldType_Uint8, 0},
		// 后缀在最后, 不能把 (n) / (p,s) 吃掉
		{"varchar(50) utf8mb4 unsigned", "varchar", "varchar(50) unsigned", 200, 0, 0,
			types.FieldType_String, types.WidthCodeFor(200)},
		{"decimal(18,2) unsigned", "decimal", "decimal(18,2) unsigned", 0, 18, 2,
			types.FieldType_Decimal, 2},
	}
	for _, c := range cases {
		decl := mysqlColumnDecl(c.dataType, mysqlUnsigned(c.columnType), c.charBytes, c.numPrec, c.numScale)
		ft, param := inferSQLType(decl, true)
		if ft != c.want || param != c.wantParam {
			t.Errorf("%s: COLUMN_TYPE=%q → decl %q → (%v, %d), 期望 (%v, %d)",
				c.name, c.columnType, decl, ft, param, c.want, c.wantParam)
		}
	}
}

// TestSQLTypeRoundTrip 声明出去的类型名必须能被自己读回同一个类型。
//
// 例外的几种都【不依赖 meta 就得靠宽度/语义兜底】, 这里显式列出来当文档:
//   - 窄整数在 SQLite 只有 INTEGER 一种写法, 读回一律 Int64;
//   - Decimal 在 SQLite 没有原生类型 (落 TEXT), 读回是 Dynamicbyte;
//   - UUID 在 MySQL 用 BINARY(16) 存, 读回被认成 Blob;
//   - Blob 与 Dynamicbyte 盘上同为动态列, 靠 meta 才能区分到具体哪一种;
//   - 定长串的**宽度码不是逐位往返的**: 声明成 VARCHAR(列宽) 再读回来会按宽度重新取档,
//     档位相同, 但"用类型默认档 (param 0)"的列会变成"写明档位 (param≠0)"。
//     所以定长串钉的不变量是【字节宽度】(见下面第二个循环), 不是 param 字面值。
//
// 也就是说: 需要"精确类型"的往返都依赖 __vrtx_meta (默认开启); 没有 meta 时
// 至少保证["数据不丢、语义不离谱"]。
func TestSQLTypeRoundTrip(t *testing.T) {
	type rt struct {
		ft    types.FieldType
		param byte
	}
	for _, mysql := range []bool{true, false} {
		// 这些类型在两个方言下都能原样往返
		exact := []rt{
			{types.FieldType_Blob, 0},
			{types.FieldType_Bool, 0},
			{types.FieldType_Year, 0},
			{types.FieldType_Float64, 0},
			{types.FieldType_Int64, 0},
			{types.FieldType_DateTime, 0}, // 无 TimeKind 时按 DATETIME 声明导出再读回
		}
		if mysql {
			exact = append(exact,
				rt{types.FieldType_Int8, 0},
				rt{types.FieldType_Int16, 0},
				rt{types.FieldType_Int32, 0},
				rt{types.FieldType_Uint8, 0},
				rt{types.FieldType_Uint16, 0},
				rt{types.FieldType_Uint32, 0},
				rt{types.FieldType_Uint64, 0},
				rt{types.FieldType_Float32, 0},
				rt{types.FieldType_Decimal, 2},
			)
		}
		// 定长串: 往返不变量是【字节宽度】。
		//
		// 声明成 VARCHAR(列宽) 之后再读回来会重新算宽度码: 宽度码列往返后宽度不变;
		// 用类型默认宽度 (param 0) 的列则会拿到一个显式宽度码 —— 宽度一样, 只是表示法
		// 从"默认宽度"变成"写明档位"。这正是本节要钉住的性质: 变的只是表示, 不是行内布局。
		stringCases := []rt{{types.FieldType_String, 0}}
		for _, size := range []int{4, 16, 32, 48, 100, 200, 256, 300, 1000, 2048} {
			ft, param := bucketColumn(size)
			stringCases = append(stringCases, rt{ft, param})
		}
		for _, w := range stringCases {
			decl := sqlTypeName(w.ft, w.param, mysql)
			ft, param := inferSQLType(decl, mysql)
			if ft != w.ft || types.FieldSize(ft, param) != types.FieldSize(w.ft, w.param) {
				t.Errorf("mysql=%v: %v/param%d → %q → (%v, param%d): 类型或行内宽度变了 (%d → %d 字节)",
					mysql, w.ft, w.param, decl, ft, param,
					types.FieldSize(w.ft, w.param), types.FieldSize(ft, param))
			}
		}

		for _, e := range exact {
			decl := sqlTypeName(e.ft, e.param, mysql)
			ft, param := inferSQLType(decl, mysql)
			if ft != e.ft || param != e.param {
				t.Errorf("mysql=%v: %v → %q → (%v, %d), 期望 (%v, %d)",
					mysql, e.ft, decl, ft, param, e.ft, e.param)
			}
		}
	}
}

// TestDeclaredCapacityNote 声明容量超过单值上限的列必须【报出来】, 且不许误报。
//
// 落动态列不等于"声明多大都行": 动态列的内容整体落在一个堆页里, 单值上限是
// types.MaxStringSize。源库里 varchar(60000) 这种声明在 VRTX 端兑现不了, 而读端是唯一看得到
// 声明原文的地方 —— 在这里漏掉, 用户只会看到一条"容量比声明小得多"的列而没有任何提示。
//
// 反过来误报同样有害: TEXT / BLOB 家族在 MySQL 的 information_schema 里带着【类型自身的】
// 上界 (text → CHARACTER_OCTET_LENGTH = 65535), 那个数不是用户的声明, 按它报"兑现不了"会让
// 每一张带 TEXT 列的表都挂上一条无意义的警告, 真正的容量问题反而被淹掉。
//
// 容量报告只影响提示, 不影响类型: 同一个入口推出来的类型必须与 inferSQLType 逐字节一致,
// 否则"报了一句"就悄悄改了列的类型, 比不报更糟。
func TestDeclaredCapacityNote(t *testing.T) {
	atCap := fmt.Sprintf("VARCHAR(%d)", types.MaxStringSize)
	overCap := fmt.Sprintf("VARCHAR(%d)", types.MaxStringSize+1)

	cases := []struct {
		decl  string
		mysql bool
		ft    types.FieldType
		param byte
		note  bool // 是否应当报"容量封顶"
	}{
		// 还在定长档: 容量本来就远小于单值上限, 不报
		{"VARCHAR(2048)", true, types.FieldType_String, types.WidthCodeFor(2048), false},
		// 已落动态列但声明没超限: 声明多少就能装多少, 不报
		{atCap, false, types.FieldType_Dynamicbyte, 0, false},
		// 超过一个字节就兑现不了: 报
		{overCap, false, types.FieldType_Dynamicbyte, 0, true},
		{"VARCHAR(60000)", true, types.FieldType_Dynamicbyte, 0, true},
		// BINARY / VARBINARY 落 blob 列 (行内 8 字节引用 + 可跨页的链, 上限约 4GB):
		// 声明多大的容量都兑现得了, 报"封顶"就是假消息。
		{"BINARY(16)", true, types.FieldType_Blob, 0, false},
		{"VARBINARY(60000)", true, types.FieldType_Blob, 0, false},
		// TEXT / BLOB 家族: 括号里那个数是类型自身的上界, 不是用户的声明 → 不报
		{"TEXT(65535)", true, types.FieldType_Dynamicbyte, 0, false},
		{"LONGTEXT(4294967295)", true, types.FieldType_Dynamicbyte, 0, false},
		{"LONGBLOB(4294967295)", true, types.FieldType_Blob, 0, false},
		// 其它带括号的类型: 括号里是精度 / 小数位, 与容量无关
		{"DECIMAL(18,2)", true, types.FieldType_Decimal, 2, false},
	}
	for _, c := range cases {
		ft, param, note := inferSQLTypeWithNote(c.decl, c.mysql)
		if ft != c.ft || param != c.param {
			t.Errorf("inferSQLTypeWithNote(%q, mysql=%v) = (%v, %d), 期望 (%v, %d)",
				c.decl, c.mysql, ft, param, c.ft, c.param)
		}
		if got := note != ""; got != c.note {
			t.Errorf("%q: 报了容量=%v (%q), 期望 =%v", c.decl, got, note, c.note)
		}
	}

	// 报出来的话要能直接定位问题: 用户写下的数与引擎能给的上限, 两个都得有。
	_, _, note := inferSQLTypeWithNote(overCap, false)
	for _, want := range []string{strconv.Itoa(types.MaxStringSize + 1), strconv.Itoa(types.MaxStringSize)} {
		if !strings.Contains(note, want) {
			t.Errorf("容量报告里缺少 %q: %q", want, note)
		}
	}
}