package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"git.ycl.pub/yl32940/TsDB2/DB/core/util"
	"bytes"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"unicode/utf8"

	_ "modernc.org/sqlite"
)

// quiet 一份不往日志里写东西的选项, 免得测试输出被进度刷屏。
func quiet() *Options {
	o := DefaultOptions()
	o.Log = func(string, ...any) {}
	return o
}

// ============================ 样本库 ============================

type sampleRow struct {
	pk   uint64
	vals map[string]any
}

// buildVRTXUsers 用 core 原生接口建一张覆盖各宽度档位的表:
// 8/1/2/4/32 字节 + 浮点 + 变长, 外加一个 B+ 树索引与一个 Hash 索引。
func buildVRTXUsers(t *testing.T, path string) {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatalf("建样本 VRTX 库: %v", err)
	}
	defer db.Close()

	fields := []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("_id")},
		{FieldType: types.FieldType_String, FieldName: []byte("Name")},
		{FieldType: types.FieldType_Uint16, FieldName: []byte("Age")},
		{FieldType: types.FieldType_Float64, FieldName: []byte("Score")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("Blob")},
		{FieldType: types.FieldType_Uint8, FieldName: []byte("Flag")},
		{FieldType: types.FieldType_Uint32, FieldName: []byte("Count")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("Big")},
	}
	tbl, err := db.CreateTable("Users", fields)
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	for _, spec := range []struct {
		name string
		typ  page.IndexType
		col  byte
	}{
		{"pk_Users", page.IndexTypePrimary, 0},
		{"idx_Users_Name", page.IndexTypeHash, 1},
		{"idx_Users_Age", page.IndexTypeBTree, 2},
	} {
		if _, err := tbl.CreateIndex(spec.name, spec.typ, []byte{spec.col}); err != nil {
			t.Fatalf("建索引 %s: %v", spec.name, err)
		}
	}

	big := bytes.Repeat([]byte{0xAB}, 300)
	rows := []sampleRow{
		{1, map[string]any{
			"_id": uint64(1), "Name": []byte("Alice"), "Age": int16(30), "Score": 1.5,
			"Blob": []byte("hello"), "Flag": uint8(7), "Count": int32(100), "Big": uint64(42),
		}},
		{2, map[string]any{
			// uint64 的负位模式与 255 的有符号/无符号边界
			"_id": uint64(2), "Name": []byte("Bob"), "Age": int16(-5), "Score": -2.25,
			"Blob": []byte{0, 1, 2, 255}, "Flag": uint8(255), "Count": int32(-1),
			"Big": uint64(1) << 63,
		}},
		{3, map[string]any{
			// 空字符串 / 空变长列 / 零值
			"_id": uint64(3), "Name": []byte(""), "Age": int16(32767), "Score": float64(0),
			"Blob": []byte{}, "Flag": uint8(0), "Count": int32(math.MaxInt32), "Big": uint64(0),
		}},
		{4, map[string]any{
			// 定长串打满 + 多字节 UTF-8 + 变长二进制 + uint64 全 1
			"_id": uint64(4), "Name": []byte(strings.Repeat("x", 32)), "Age": int16(-32768),
			"Score": 1e-9, "Blob": big, "Flag": uint8(1), "Count": int32(math.MinInt32),
			"Big": uint64(math.MaxUint64),
		}},
		{5, map[string]any{
			"_id": uint64(5), "Name": []byte("王小明"), "Age": int16(1),
			"Score": 3.141592653589793, "Blob": []byte("中文内容"), "Flag": uint8(128),
			"Count": int32(0), "Big": uint64(7),
		}},
	}
	writeVRTXRows(t, tbl, fields, rows)
}

// buildVRTXMetrics 建一张时序表: 行存储索引是 Time 类型, 键是 _ts。
func buildVRTXMetrics(t *testing.T, path string) {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatalf("建时序样本库: %v", err)
	}
	defer db.Close()

	fields := []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("_ts")},
		{FieldType: types.FieldType_Float64, FieldName: []byte("Value")},
		{FieldType: types.FieldType_String, FieldName: []byte("Device")},
	}
	tbl, err := db.CreateTable("Metrics", fields)
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk_Metrics", page.IndexTypeTime, []byte{0}); err != nil {
		t.Fatalf("建时序索引: %v", err)
	}
	// 时间必须递增: 时序索引按时间段分段, 乱序插入会被拒。
	writeVRTXRows(t, tbl, fields, []sampleRow{
		{1, map[string]any{"_ts": uint64(1000), "Value": 1.5, "Device": []byte("s1")}},
		{2, map[string]any{"_ts": uint64(2000), "Value": 2.5, "Device": []byte("s2")}},
		{3, map[string]any{"_ts": uint64(3000), "Value": 3.5, "Device": []byte("s1")}},
	})
}

func writeVRTXRows(t *testing.T, tbl *core.Table, fields []types.FieldInfo, rows []sampleRow) {
	t.Helper()
	for _, r := range rows {
		vals := make([]types.FieldValue, len(fields))
		for i, f := range fields {
			vals[i] = types.FieldValue{FieldName: f.FieldName, Value: r.vals[util.BytesToString(f.FieldName)]}
		}
		if _, err := tbl.InsertRowWithID(r.pk, vals); err != nil {
			t.Fatalf("插入 pk=%d: %v", r.pk, err)
		}
	}
}

// ============================ 读回真实文件 ============================

// vrtxSnapshot 直接把 VRTX 文件读成 (列结构, 索引, 数据) 三份快照 —— 不走 migrate
// 自己的读端, 免得"读写两端用同一段代码"把不对称的错悄悄抹平。
type vrtxSnapshot struct {
	columns []string // "Name:32 字节"
	indexes map[string]string
	rows    map[uint64]map[string]any
}

func snapshotVRTX(t *testing.T, path, table string) vrtxSnapshot {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatalf("打开 %s: %v", path, err)
	}
	defer db.Close()
	tbl, ok := db.FindTable(table)
	if !ok {
		t.Fatalf("%s 里没有表 %s", path, table)
	}

	snap := vrtxSnapshot{indexes: map[string]string{}, rows: map[uint64]map[string]any{}}
	fields := tbl.GetFields()
	for _, f := range fields {
		if util.IsHighBitSet(byte(f.FieldType)) {
			continue
		}
		// NULL / 唯一声明也进快照: 它们和类型一样是结构的一部分, 往返丢了就是
		// 静默改结构 (列从"可空"变"非空"、约束从"唯一"变"普通")。
		snap.columns = append(snap.columns, fmt.Sprintf("%s:%s%s",
			util.BytesToString(f.FieldName), TypeName(f.FieldType), flagMarks(f.Flag)))
	}
	for _, m := range tbl.IndexMeta().GetIndexes() {
		var cols []string
		for _, fid := range m.Fields() {
			cols = append(cols, util.BytesToString(fields[fid].FieldName))
		}
		mark := ""
		if m.Unique() {
			mark = " unique"
		}
		snap.indexes[m.Name()] = fmt.Sprintf("%s(%s)%s",
			IndexTypeName(m.IndexType), strings.Join(cols, ","), mark)
	}
	if err := tbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
		m := make(map[string]any, len(vals))
		for _, fv := range vals {
			m[util.BytesToString(fv.FieldName)] = fv.Value
		}
		snap.rows[pkID] = m
		return true
	}); err != nil {
		t.Fatalf("扫描 %s.%s: %v", path, table, err)
	}
	return snap
}

func norm(v any) string {
	if b, ok := v.([]byte); ok {
		return fmt.Sprintf("bytes:%x", b)
	}
	if f, ok := v.(float64); ok {
		return "f64:" + strconv.FormatFloat(f, 'x', -1, 64)
	}
	return fmt.Sprintf("%T:%v", v, v)
}

// ============================ 用例 ============================

// TestVRTXSQLiteRoundTrip VRTX → SQLite → VRTX: 结构、索引类型、主键与每一行的
// 每个字段都必须原样回来。
func TestVRTXSQLiteRoundTrip(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.db")
	midPath := filepath.Join(dir, "mid.sqlite")
	backPath := filepath.Join(dir, "back.db")

	buildVRTXUsers(t, srcPath)

	// 先取一份源快照, 后面拿它对照
	want := snapshotVRTX(t, srcPath, "Users")

	// VRTX → SQLite
	v1, err := OpenVRTX(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	s1, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(v1, s1, quiet())
	if err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	if rep.Rows != 5 {
		t.Fatalf("VRTX→SQLite 应迁 5 行, 实际 %d", rep.Rows)
	}
	_ = v1.Close()
	_ = s1.Close()

	// SQLite → VRTX
	s2, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v2, err := OpenVRTX(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(s2, v2, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = s2.Close()
	_ = v2.Close()

	got := snapshotVRTX(t, backPath, "Users")

	if strings.Join(got.columns, "|") != strings.Join(want.columns, "|") {
		t.Errorf("列结构不一致\n 源: %v\n 回: %v", want.columns, got.columns)
	}
	for name, def := range want.indexes {
		if got.indexes[name] != def {
			t.Errorf("索引 %s 不一致: 源 %q, 回 %q", name, def, got.indexes[name])
		}
	}
	if len(got.indexes) != len(want.indexes) {
		t.Errorf("索引数量不一致: 源 %v, 回 %v", want.indexes, got.indexes)
	}
	if len(got.rows) != len(want.rows) {
		t.Fatalf("行数不一致: 源 %d, 回 %d", len(want.rows), len(got.rows))
	}
	for pk, wantRow := range want.rows {
		gotRow, ok := got.rows[pk]
		if !ok {
			t.Errorf("pk=%d 的行丢了", pk)
			continue
		}
		for col, wantVal := range wantRow {
			if norm(gotRow[col]) != norm(wantVal) {
				t.Errorf("pk=%d 列 %s 不一致: 源 %s, 回 %s",
					pk, col, norm(wantVal), norm(gotRow[col]))
			}
		}
	}
}

// TestVRTXTimeSeriesRoundTrip 时序表: 行存储索引是 Time 类型, 键是 _ts 而不是行号。
// 这种表的行号由索引内部维护 (且首个分段的 pkID 并不可靠), 所以往返时按 _ts 对照;
// 要求是 Time 索引仍建在 _ts 上, 且每个时间点的数据一字不差。
func TestVRTXTimeSeriesRoundTrip(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.db")
	midPath := filepath.Join(dir, "mid.sqlite")
	backPath := filepath.Join(dir, "back.db")

	buildVRTXMetrics(t, srcPath)
	want := snapshotVRTX(t, srcPath, "Metrics")

	v1, err := OpenVRTX(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	s1, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(v1, s1, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = v1.Close()
	_ = s1.Close()

	// 中间库不该冒出 _id 之类的替代主键: 时序表在 SQL 侧本来就没有整数主键
	mid, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	midSchema, err := mid.Schema("Metrics")
	if err != nil {
		t.Fatal(err)
	}
	if midSchema.PKColumn != "" {
		t.Errorf("时序表导出到 SQL 后不该有主键列, 实际 %q", midSchema.PKColumn)
	}
	_ = mid.Close()

	s2, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v2, err := OpenVRTX(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(s2, v2, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = s2.Close()
	_ = v2.Close()

	got := snapshotVRTX(t, backPath, "Metrics")
	if got.indexes["pk_Metrics"] != "time(_ts)" {
		t.Errorf("时序索引没回来: %v", got.indexes)
	}
	if len(got.rows) != len(want.rows) {
		t.Fatalf("行数不一致: 源 %d, 回 %d", len(want.rows), len(got.rows))
	}
	// 行号不可比, 按 _ts 配对
	byTS := func(rows map[uint64]map[string]any) map[string]map[string]any {
		out := make(map[string]map[string]any, len(rows))
		for _, r := range rows {
			out[norm(r["_ts"])] = r
		}
		return out
	}
	wantTS, gotTS := byTS(want.rows), byTS(got.rows)
	for ts, wantRow := range wantTS {
		gotRow, ok := gotTS[ts]
		if !ok {
			t.Errorf("_ts=%s 的点丢了", ts)
			continue
		}
		for _, col := range []string{"_ts", "Value", "Device"} {
			if norm(gotRow[col]) != norm(wantRow[col]) {
				t.Errorf("_ts=%s 列 %s 不一致: 源 %s, 回 %s",
					ts, col, norm(wantRow[col]), norm(gotRow[col]))
			}
		}
	}
}

// TestSQLiteOriginToVRTX SQLite 里手工建的表 (无 __vrtx_meta) → VRTX → SQLite:
// 类型按 SQL 声明就近取档, 主键与二级索引都要保住。
func TestSQLiteOriginToVRTX(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	vrtxPath := filepath.Join(dir, "mid.db")
	backPath := filepath.Join(dir, "back.sqlite")

	createSQLiteSource(t, srcPath)
	want := readSQLiteRows(t, srcPath)

	s1, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v1, err := OpenVRTX(vrtxPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(s1, v1, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = s1.Close()
	_ = v1.Close()

	// VRTX 侧应该拿到 8 字节主键 + 两个索引 (primary + btree)
	vSnap := snapshotVRTX(t, vrtxPath, "Items")
	if vSnap.indexes["pk_Items"] != "primary(id)" {
		t.Errorf("VRTX 主键索引不对: %v", vSnap.indexes)
	}
	if vSnap.indexes["idx_Items_name"] != "btree(name)" {
		t.Errorf("VRTX 二级索引不对: %v", vSnap.indexes)
	}

	v2, err := OpenVRTX(vrtxPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	s2, err := OpenSQLite(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(v2, s2, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = v2.Close()
	_ = s2.Close()

	got := readSQLiteRows(t, backPath)
	if len(got) != len(want) {
		t.Fatalf("行数不一致: 源 %d, 回 %d", len(want), len(got))
	}
	for id, wantRow := range want {
		gotRow, ok := got[id]
		if !ok {
			t.Errorf("id=%s 的行丢了", id)
			continue
		}
		if norm(gotRow["name"]) != norm(wantRow["name"]) ||
			norm(gotRow["qty"]) != norm(wantRow["qty"]) ||
			norm(gotRow["price"]) != norm(wantRow["price"]) ||
			norm(gotRow["memo"]) != norm(wantRow["memo"]) {
			t.Errorf("id=%s 不一致: 源 %v, 回 %v", id, wantRow, gotRow)
		}
	}
}

// TestSQLiteTimeColumnsToVRTX 源库的时间列迁到 VRTX。
//
// 三条硬要求, 每一条都对应一个真实踩过的坑:
//
//  1. 落引擎的专用时间类型 FieldType_DateTime (行内 8 字节 Unix 毫秒, 与 gormvrtx /
//     SQL driver 同一约定) —— 迁过来的表经 ORM 读 created_at 时列类型才是 datetime
//     (扫描类型 time.Time), 不会出现"列是 uint64、却要按时间读"的类型错误;
//  2. 建在时间列上的索引必须真的建出来。时间列以前落在 dynamic 上, 引擎会拒
//     "动态列不能作为索引键", 整张表跟着废掉 (resource_online.idx_ctime 那个报错);
//  3. 值要是时间戳本身 (毫秒), 且按 UTC 解释 —— 同一份数据在不同时区的机器上
//     必须迁出同一个数。
func TestSQLiteTimeColumnsToVRTX(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	midPath := filepath.Join(dir, "mid.db")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE Events (
			id INTEGER PRIMARY KEY,
			ctime DATETIME NOT NULL,
			day DATE NOT NULL
		)`,
		`CREATE INDEX idx_Events_ctime ON Events(ctime)`,
		`INSERT INTO Events (id, ctime, day) VALUES (1, '2024-03-05 12:34:56.123', '2024-03-05')`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	s1, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v1, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(s1, v1, quiet())
	if err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	if rep.Rows != 1 {
		t.Fatalf("应迁 1 行, 实际 %d", rep.Rows)
	}
	_ = s1.Close()
	_ = v1.Close()

	snap := snapshotVRTX(t, midPath, "Events")

	// 1. 时间列是引擎专用 datetime 类型, 不是 dynamic / 普通 uint64
	cols := strings.Join(snap.columns, ",")
	for _, want := range []string{"ctime:datetime", "day:datetime"} {
		if !strings.Contains(cols, want) {
			t.Errorf("时间列应当落成 %s, 实际列结构 %v", want, snap.columns)
		}
	}

	// 1b. 直接查引擎 meta: 列的 FieldType 必须是 FieldType_DateTime —— gormvrtx 的
	// 扫描类型映射 (scanTypeOf) 正是按这个值给出 time.Time, 这是"ORM 读 created_at
	// 不报类型错误"的确切依据, 不靠 migrate 自己的读端自证。
	vdb, err := core.Open(midPath)
	if err != nil {
		t.Fatal(err)
	}
	vtbl, ok := vdb.FindTable("Events")
	if !ok {
		t.Fatal("VRTX 里没有 Events 表")
	}
	wantFT := map[string]types.FieldType{"ctime": types.FieldType_DateTime, "day": types.FieldType_DateTime}
	for _, f := range vtbl.GetFields() {
		name := util.BytesToString(f.FieldName)
		if want, ok := wantFT[name]; ok && f.FieldType != want {
			t.Errorf("列 %s 的 FieldType = %v, 期望 %v (gormvrtx 会因此给错扫描类型)", name, f.FieldType, want)
		}
		delete(wantFT, name)
	}
	if len(wantFT) != 0 {
		t.Errorf("Events 表缺列: %v", wantFT)
	}
	_ = vdb.Close()

	// 2. 索引建出来了 (以前这里整张表都废掉)
	if got := snap.indexes["idx_Events_ctime"]; got != "btree(ctime)" {
		t.Errorf("建在时间列上的索引没建出来, 实际 %v", snap.indexes)
	}

	// 3. 值是 UTC 毫秒时间戳本身。数字写死是为了不让实现自证自己:
	// 2024-03-05 12:34:56.123 UTC = 1709642096123 毫秒; 2024-03-05 = 1709596800000 毫秒。
	for col, wantMS := range map[string]uint64{
		"ctime": 1709642096123,
		"day":   1709596800000,
	} {
		got := snap.rows[1][col]
		if s := fmt.Sprintf("%v", got); s != strconv.FormatUint(wantMS, 10) {
			t.Errorf("%s 的 Unix 毫秒应为 %d, 实际 %v (%T)", col, wantMS, got, got)
		}
	}

	// 4. 再迁回 SQLite, 时间文本必须逐字符回来 (毫秒无损)。
	backPath := filepath.Join(dir, "back.sqlite")
	v2, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	s2, err := OpenSQLite(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(v2, s2, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = v2.Close()
	_ = s2.Close()

	bdb, err := sql.Open("sqlite", backPath)
	if err != nil {
		t.Fatal(err)
	}
	var ctime, day string
	// CAST 成 TEXT 读原样存储值: modernc 驱动会把 DATETIME 声明列里的时间文本
	// 读成 time.Time, 再 Scan 进 string 时又格式化回 RFC3339 —— 那是驱动的往返
	// 转换, 不是迁移写错了文本。这里要验的是"落库的文本逐字符无损"。
	if err := bdb.QueryRow(`SELECT CAST(ctime AS TEXT), CAST(day AS TEXT) FROM Events WHERE id = 1`).
		Scan(&ctime, &day); err != nil {
		t.Fatalf("读回 SQLite: %v", err)
	}
	_ = bdb.Close()
	if ctime != "2024-03-05 12:34:56.123" {
		t.Errorf("ctime 应逐字符回来, 实际 %q", ctime)
	}
	// day 走到 VRTX 再直出, DATE 的细分声明丢失 (VRTX schema 没有放 TimeKind 的位置,
	// 导出只能按 DATETIME 兜底声明, 见 types.go 的 TimeKind 注释) —— 但值必须无损:
	// 仍是那一天的零点, 不是日期被砍或时间被挪。
	if day != "2024-03-05 00:00:00" {
		t.Errorf("day 的值应无损回来 (DATETIME 兜底声明), 实际 %q", day)
	}
}

// TestLegacyUint64NanoTimeColumns 兼容路径: 存量旧表 (Uint64 列存 Unix 纳秒) 经 migrate
// 读出仍按【纳秒】解释, 不能被新逻辑当成毫秒。
//
// 场景还原: 更早版本的 migrate 把 SQL 时间列迁成 Uint64 存纳秒, 并在 __vrtx_meta 里
// 给该列打上 TimeKind 标记。这样的中间库今天再迁, 结构按 meta 还原成 Uint64+TimeKind
// (写路径照旧换算成纳秒), 值必须是原来的那个纳秒数 —— 走错路会静默给出 1970 年附近
// 的垃圾时间 (毫秒当纳秒解释) 或天文数字 (纳秒当毫秒解释)。
func TestLegacyUint64NanoTimeColumns(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE Events (
			id INTEGER PRIMARY KEY,
			ctime DATETIME NOT NULL
		)`,
		`INSERT INTO Events (id, ctime) VALUES (1, '2024-03-05 12:34:56.123456')`,
		// 旧版 migrate 写出的保真表: ctime 的类型记成 Uint64 (10) + TimeKind 标记。
		// 这正是存量中间库的样子 —— 新实现读到它, 列必须还原成 Uint64+TimeKind。
		`CREATE TABLE __vrtx_meta (tbl TEXT PRIMARY KEY, payload TEXT NOT NULL)`,
		`INSERT INTO __vrtx_meta (tbl, payload) VALUES ('Events',
			'{"v":1,"pk":"id","pkt":2,"cols":[{"n":"id","t":11},{"n":"ctime","t":10,"tm":"datetime"}]}')`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	// 结构按旧 meta 还原: ctime 是 Uint64 + TimeKind, 不是 DateTime。
	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	sc, err := src.Schema("Events")
	if err != nil {
		t.Fatalf("读结构: %v", err)
	}
	var ctimeCol *Column
	for i := range sc.Columns {
		if sc.Columns[i].Name == "ctime" {
			ctimeCol = &sc.Columns[i]
		}
	}
	if ctimeCol == nil {
		t.Fatal("Events 表没有 ctime 列")
	}
	if ctimeCol.Type != types.FieldType_Uint64 || ctimeCol.TimeKind != "datetime" {
		t.Fatalf("旧 meta 应把 ctime 还原成 Uint64+TimeKind, 实际 %v (tm=%q)", ctimeCol.Type, ctimeCol.TimeKind)
	}

	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(src, dst, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = src.Close()
	_ = dst.Close()

	// 写路径按旧盘面换算: 时间文本 → Unix 纳秒, 落进 Uint64 列。
	// 1709642096123456000 纳秒 = 2024-03-05 12:34:56.123456 UTC。
	snap := snapshotVRTX(t, dstPath, "Events")
	if got := fmt.Sprintf("%v", snap.rows[1]["ctime"]); got != "1709642096123456000" {
		t.Errorf("旧 Uint64 时间列应按纳秒落值, 实际 %v (%T)",
			snap.rows[1]["ctime"], snap.rows[1]["ctime"])
	}
	cols := strings.Join(snap.columns, ",")
	if !strings.Contains(cols, "ctime:uint64") {
		t.Errorf("按旧 meta 建的列应是 uint64, 实际列结构 %v", snap.columns)
	}

	// 读路径 (SQL 侧) 同样按纳秒还原: 从这张旧 VRTX 表经 SQL 导出到带旧 meta 的库,
	// 时间文本必须原样回来。这里直接钉 coerceForSQL 的 Uint64+TimeKind 分支:
	// 纳秒进、微秒文本出 (MySQL DATETIME(6) 的上限), 日期语义不丢。
	legacyCol := Column{Name: "ctime", Type: types.FieldType_Uint64, TimeKind: "datetime"}
	got, err := coerceForSQL(legacyCol, snap.rows[1]["ctime"], false)
	if err != nil {
		t.Fatalf("旧列导出 SQL 不该报错: %v", err)
	}
	if s, ok := got.(string); !ok || s != "2024-03-05 12:34:56.123456" {
		t.Errorf("旧 Uint64 时间列应按纳秒还原成时间文本, 实际 %v (%T)", got, got)
	}
}

// TestSkippedIndexDoesNotKillTable 目标端建不出来的索引应当只跳它自己。
//
// dynamic 列 (长文本) 不能作为索引键是引擎的硬限制, 但源库里这是条合法索引。
// 迁一张"数据都在, 只是少个索引"的表, 比让整张表陪葬合理 —— 索引可以事后补,
// 数据没迁才是真的丢。跳过的索引要写进报告, 不能静默。
func TestSkippedIndexDoesNotKillTable(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	midPath := filepath.Join(dir, "mid.db")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{
		`CREATE TABLE Blobs (
			id INTEGER PRIMARY KEY,
			body BLOB NOT NULL
		)`,
		`CREATE INDEX idx_Blobs_body ON Blobs(body)`,
		`INSERT INTO Blobs (id, body) VALUES (1, 'hello')`,
	} {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	_ = db.Close()

	s1, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v1, err := OpenVRTX(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	rep, err := Run(s1, v1, quiet())
	if err != nil {
		t.Fatalf("跳个索引不该让迁移失败: %v", err)
	}
	_ = s1.Close()
	_ = v1.Close()

	if rep.Rows != 1 {
		t.Fatalf("表的数据应当照迁, 实际 %d 行", rep.Rows)
	}
	if len(rep.Tables) != 1 || !strings.Contains(rep.Tables[0].Note, "跳过索引") {
		t.Fatalf("跳过的索引必须写进报告, 实际: %+v", rep.Tables)
	}
	snap := snapshotVRTX(t, midPath, "Blobs")
	if _, ok := snap.indexes["idx_Blobs_body"]; ok {
		t.Errorf("dynamic 列的索引不该建出来: %v", snap.indexes)
	}
	if len(snap.rows) != 1 {
		t.Errorf("数据应当照常迁过去, 实际 %d 行", len(snap.rows))
	}
}

// TestRunRejectsSelfMigration 两件事:
//  1. 锁语义: 同一路径的第二次打开必须被文件锁拒绝 (跨进程与同进程其它句柄
//     一视同仁 —— 引擎只允许一个实例操作一个库);
//  2. Run 的自我迁移闸门: 同一句柄同时作为源与目标必须被拒 (迁移前检查,
//     不做任何 I/O)。
// 旧实现靠"同进程双开同一文件"搭场景, 而双开正是锁禁止的行为 —— 闸门本身
// 用同一句柄传两遍即可验证, 不再依赖双开。
func TestRunRejectsSelfMigration(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "same.db")
	buildVRTXUsers(t, path)

	a, err := OpenVRTX(path, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer a.Close()

	// 1) 锁: 同路径第二次打开 (同进程第二个句柄) 必须报 ErrDatabaseLocked
	if _, err := OpenVRTX(path, quiet()); !errors.Is(err, core.ErrDatabaseLocked) {
		t.Fatalf("同路径第二次打开应报 ErrDatabaseLocked, 实际: %v", err)
	}

	// 2) 自我迁移: Run 在迁移前按 Identity 拦截
	if _, err := Run(a, a, quiet()); err == nil {
		t.Fatal("同一个库的自我迁移应当报错")
	}
}

// TestRunLimit 只搬前 N 行。
func TestRunLimit(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.db")
	midPath := filepath.Join(dir, "mid.sqlite")
	buildVRTXUsers(t, srcPath)

	o := quiet()
	o.Limit = 2

	v1, err := OpenVRTX(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer v1.Close()
	s1, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer s1.Close()

	rep, err := Run(v1, s1, o)
	if err != nil {
		t.Fatalf("限额迁移失败: %v", err)
	}
	if rep.Rows != 2 {
		t.Fatalf("应只搬 2 行, 实际 %d", rep.Rows)
	}
}

// createSQLiteSource 用原生 SQL 建一个外部风格的库 (没有 __vrtx_meta)。
func createSQLiteSource(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	stmts := []string{
		`CREATE TABLE Items (
			id INTEGER PRIMARY KEY,
			name VARCHAR(32) NOT NULL,
			qty INTEGER NOT NULL,
			price REAL NOT NULL,
			memo BLOB NOT NULL
		)`,
		`CREATE INDEX idx_Items_name ON Items(name)`,
	}
	for _, s := range stmts {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	rows := []struct {
		id    int64
		name  string
		qty   int64
		price float64
		memo  []byte
	}{
		{1, "apple", 10, 1.25, []byte("fresh")},
		{2, "banana", -3, 0.5, []byte{0x00, 0xFF, 0x10}},
		{3, "中文名", 0, 9.99, []byte("")},
	}
	for _, r := range rows {
		if _, err := db.Exec(`INSERT INTO Items (id, name, qty, price, memo) VALUES (?, ?, ?, ?, ?)`,
			r.id, r.name, r.qty, r.price, r.memo); err != nil {
			t.Fatalf("插入 %v: %v", r, err)
		}
	}
}

func readSQLiteRows(t *testing.T, path string) map[string]map[string]any {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	rows, err := db.Query(`SELECT id, name, qty, price, memo FROM Items`)
	if err != nil {
		t.Fatalf("查询 %s: %v", path, err)
	}
	defer rows.Close()

	out := map[string]map[string]any{}
	for rows.Next() {
		var (
			id    int64
			name  string
			qty   int64
			price float64
			memo  []byte
		)
		if err := rows.Scan(&id, &name, &qty, &price, &memo); err != nil {
			t.Fatal(err)
		}
		out[strconv.FormatInt(id, 10)] = map[string]any{
			"name": name, "qty": qty, "price": price, "memo": memo,
		}
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	return out
}

// TestMySQLDatabasesUnreachable 主机不通时列库要立刻报错 —— 连接界面上是个按钮,
// 绝不能让它挂着不动, 更不能悄悄返回空列表 (会被当成"这台机器上没有库")。
func TestMySQLDatabasesUnreachable(t *testing.T) {
	// 1 号端口基本不会有人监听: 连接立即被拒, 不依赖任何真实 MySQL
	dsn := "root:@tcp(127.0.0.1:1)/?timeout=2s"
	if _, err := MySQLDatabases(dsn); err == nil {
		t.Fatal("连不上的主机应当报错, 而不是返回空列表")
	}
}

// TestRunContinueOnError 验的是"一张表出问题不能连坐整趟":
// 没有主键的表会被 prepareForVRTX 拒掉, 但同批里能迁的表必须照常迁完 ——
// 失败原因记进报告, 最后 Run 用聚合错误点名, 而不是静默吞掉。
func TestRunContinueOnError(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	stmts := []string{
		// 没有主键: VRTX 侧建不出来, 这张表注定失败
		`CREATE TABLE OldUserDetail (id INTEGER, detail TEXT)`,
		`INSERT INTO OldUserDetail (id, detail) VALUES (1, 'x')`,
		// 正常表: 不管上面那张怎么炸, 它都该迁完
		`CREATE TABLE GoodTable (id INTEGER PRIMARY KEY, name VARCHAR(32))`,
		`INSERT INTO GoodTable (id, name) VALUES (1, 'a'), (2, 'b')`,
	}
	for _, s := range stmts {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("执行 %s: %v", s, err)
		}
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}

	// 继续档: 坏表跳过, 好表照迁
	o := quiet()
	o.ContinueOnError = true

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer src.Close()
	dst, err := OpenVRTX(filepath.Join(dir, "keep.db"), quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer dst.Close()

	rep, runErr := Run(src, dst, o)
	if runErr == nil {
		t.Fatalf("有表没能迁完, 应当返回汇总错误; 报告:\n%s", rep)
	}
	if got := rep.Failed(); got != 1 {
		t.Fatalf("应当只有 1 张表失败, 实际 %d; 报告:\n%s", got, rep)
	}
	if !strings.Contains(runErr.Error(), "OldUserDetail") {
		t.Fatalf("汇总错误应点名失败的表: %v", runErr)
	}
	for _, tr := range rep.Tables {
		if tr.Table == "OldUserDetail" && tr.Error == "" {
			t.Fatalf("失败原因没记进报告: %+v", tr)
		}
	}

	var (
		goodRows uint64
		seen     bool
	)
	for _, tr := range rep.Tables {
		if tr.Table == "GoodTable" {
			seen, goodRows = true, tr.Rows
		}
	}
	if !seen {
		t.Fatalf("坏表把后面的表一起带走了, GoodTable 根本没过一遍; 报告:\n%s", rep)
	}
	if goodRows != 2 {
		t.Fatalf("GoodTable 应当迁 2 行, 实际 %d", goodRows)
	}
	if n := rep.Failed(); n != 1 {
		t.Fatalf("报告里的失败表数应为 1, 实际 %d", n)
	}

	// 对照: 默认档遇错即停, 不该有"带了失败的报告"这种东西返回
	strict := quiet()
	src2, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer src2.Close()
	dst2, err := OpenVRTX(filepath.Join(dir, "strict.db"), quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer dst2.Close()

	if _, err := Run(src2, dst2, strict); err == nil {
		t.Fatal("默认档应当在第一张失败的表上就中止并报错")
	}
}

// ============================ NULL / 唯一约束 ============================

// flagMarks 列上的 NULL / 唯一 / decimal scale 声明, 拼成快照里的后缀。
func flagMarks(flag byte) string {
	var s string
	if flag&types.FlagNullable != 0 {
		s += " NULL"
	}
	if flag&types.FlagUnique != 0 {
		s += " UNIQUE"
	}
	if sc := flag & types.FlagScaleMask; sc != 0 {
		s += fmt.Sprintf(" scale=%d", sc)
	}
	return s
}

// buildVRTXNullUnique 建一张把 NULL 与唯一约束都压满的样本表:
//
//	_id  uint64   行号 (主键列, 非空)
//	Name 32 字节  非空 + 唯一索引 (列上也带声明位)
//	Note uint64   可空 + 唯一索引
//	Age  uint16   可空, 无索引
//
// Note 特意让两行为 NULL: 唯一索引对 NULL 是免除的 (任一分量为 NULL 的行不进索引),
// 非空唯一列测不出这条 —— 也正是"NULL 不能被写成 0"最容易出问题的地方。
func buildVRTXNullUnique(t *testing.T, path string) {
	t.Helper()
	db, err := core.Open(path)
	if err != nil {
		t.Fatalf("建样本 VRTX 库: %v", err)
	}
	defer db.Close()

	fields := []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("_id")},
		{FieldType: types.FieldType_String, FieldName: []byte("Name"), Flag: types.FlagUnique},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("Note"), Flag: types.FlagNullable | types.FlagUnique},
		{FieldType: types.FieldType_Uint16, FieldName: []byte("Age"), Flag: types.FlagNullable},
	}
	tbl, err := db.CreateTable("People", fields)
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	for _, spec := range []struct {
		name   string
		typ    page.IndexType
		col    byte
		unique bool
	}{
		{"pk_People", page.IndexTypePrimary, 0, false},
		{"idx_People_Name", page.IndexTypeBTree, 1, true},
		{"idx_People_Note", page.IndexTypeBTree, 2, true},
	} {
		if _, err := tbl.CreateIndex(spec.name, spec.typ, []byte{spec.col}, spec.unique); err != nil {
			t.Fatalf("建索引 %s: %v", spec.name, err)
		}
	}
	writeVRTXRows(t, tbl, fields, []sampleRow{
		{pk: 1, vals: map[string]any{"_id": uint64(1), "Name": []byte("Alice"), "Note": uint64(100), "Age": int16(30)}},
		// Note / Age 写 nil: 落盘就是 NULL 位图上的一位
		{pk: 2, vals: map[string]any{"_id": uint64(2), "Name": []byte("Bob"), "Note": nil, "Age": nil}},
		{pk: 3, vals: map[string]any{"_id": uint64(3), "Name": []byte("Cara"), "Note": nil, "Age": int16(7)}},
	})
}

// checkSQLiteShape 直接查中间库的 PRAGMA, 确认可空性与唯一性真的落在了 DDL 上
// (而不是只记在 __vrtx_meta 里, SQL 侧其实什么都没表达)。
func checkSQLiteShape(t *testing.T, path, table string, nullable, unique map[string]bool) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatalf("打开中间库: %v", err)
	}
	defer db.Close()

	rows, err := db.Query(`PRAGMA table_info("` + table + `")`)
	if err != nil {
		t.Fatalf("table_info: %v", err)
	}
	gotNullable := map[string]bool{}
	for rows.Next() {
		var cid, notNull, pk int
		var name, decl string
		var dflt any
		if err := rows.Scan(&cid, &name, &decl, &notNull, &dflt, &pk); err != nil {
			t.Fatalf("扫描 table_info: %v", err)
		}
		gotNullable[name] = notNull == 0
	}
	rows.Close()
	for col, want := range nullable {
		if gotNullable[col] != want {
			t.Errorf("中间库列 %s 的可空性不对: 实际 %v, 期望 %v", col, gotNullable[col], want)
		}
	}

	idxRows, err := db.Query(`PRAGMA index_list("` + table + `")`)
	if err != nil {
		t.Fatalf("index_list: %v", err)
	}
	gotUnique := map[string]bool{}
	for idxRows.Next() {
		var seq, uniq, partial int
		var name, origin string
		if err := idxRows.Scan(&seq, &name, &uniq, &origin, &partial); err != nil {
			t.Fatalf("扫描 index_list: %v", err)
		}
		if origin == "c" { // 显式 CREATE INDEX 建的才是我们要比对的
			gotUnique[name] = uniq != 0
		}
	}
	idxRows.Close()
	for name, want := range unique {
		if gotUnique[name] != want {
			t.Errorf("中间库索引 %s 的唯一性不对: 实际 %v, 期望 %v", name, gotUnique[name], want)
		}
	}
}

// TestVRTXSQLiteNullUniqueRoundTrip 可空列 + 唯一索引的 VRTX → SQLite → VRTX。
//
// 这两个属性在两端各只有一半的表达: NULL 在 VRTX 是行内位图的位、在 SQL 是列可空
// 声明加上行里的 NULL; 唯一在 VRTX 是索引上的标志位、在 SQL 是 UNIQUE INDEX。
// 少带任何一半, 往返都会静默变形 —— NULL 被写成一个 0, 或者约束悄悄退化成普通索引。
func TestVRTXSQLiteNullUniqueRoundTrip(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.db")
	midPath := filepath.Join(dir, "mid.sqlite")
	backPath := filepath.Join(dir, "back.db")

	buildVRTXNullUnique(t, srcPath)
	want := snapshotVRTX(t, srcPath, "People")

	v1, err := OpenVRTX(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	s1, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(v1, s1, quiet()); err != nil {
		t.Fatalf("VRTX→SQLite: %v", err)
	}
	_ = v1.Close()
	_ = s1.Close()

	// 中间库的结构: 可空列写 NULL (不是硬写 NOT NULL), 索引写 UNIQUE INDEX
	checkSQLiteShape(t, midPath, "People",
		map[string]bool{"_id": false, "Name": false, "Note": true, "Age": true},
		map[string]bool{"idx_People_Name": true, "idx_People_Note": true})

	s2, err := OpenSQLite(midPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	v2, err := OpenVRTX(backPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(s2, v2, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = s2.Close()
	_ = v2.Close()

	got := snapshotVRTX(t, backPath, "People")
	if strings.Join(got.columns, "|") != strings.Join(want.columns, "|") {
		t.Errorf("列结构 (含 NULL / 唯一声明) 不一致\n  源: %v\n  回: %v", want.columns, got.columns)
	}
	if len(got.indexes) != len(want.indexes) {
		t.Errorf("索引数量不一致\n  源: %v\n  回: %v", want.indexes, got.indexes)
	}
	for name, def := range want.indexes {
		if got.indexes[name] != def {
			t.Errorf("索引 %s 不一致\n  源: %q\n  回: %q", name, def, got.indexes[name])
		}
	}
	if len(got.rows) != len(want.rows) {
		t.Fatalf("行数不一致: 源 %d, 回 %d", len(want.rows), len(got.rows))
	}
	// 逐列比对, 并且把"源里有值/回里没值"也算不一致 —— NULL 往返丢了不会体现在
	// norm() 上, 只会体现为"这一列在 map 里消失了"。
	for pk, wantRow := range want.rows {
		gotRow, ok := got.rows[pk]
		if !ok {
			t.Errorf("pk=%d 的行丢了", pk)
			continue
		}
		for _, col := range []string{"Name", "Note", "Age"} {
			wantVal, wantHas := wantRow[col]
			gotVal, gotHas := gotRow[col]
			if wantHas != gotHas {
				t.Errorf("pk=%d 列 %s 的有无不一致: 源 %v (有值=%v), 回 %v (有值=%v) —— NULL 被写成零值了",
					pk, col, wantVal, wantHas, gotVal, gotHas)
				continue
			}
			if wantHas && norm(gotVal) != norm(wantVal) {
				t.Errorf("pk=%d 列 %s 不一致: 源 %s, 回 %s", pk, col, norm(wantVal), norm(gotVal))
			}
		}
	}

	// 约束回到 VRTX 后必须真的生效, 不是画在 schema 上的装饰
	back, err := core.Open(backPath)
	if err != nil {
		t.Fatal(err)
	}
	defer back.Close()
	btbl, ok := back.FindTable("People")
	if !ok {
		t.Fatal("回来的库里没有 People 表")
	}
	if _, err := btbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("Name"), Value: []byte("Alice")},
	}); !errors.Is(err, core.ErrDuplicatedKey) {
		t.Errorf("重复 Name 应当报 ErrDuplicatedKey, 实际 %v", err)
	}
	if _, err := btbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("Name"), Value: []byte("Dan")},
		{FieldName: []byte("Note"), Value: uint64(100)},
	}); !errors.Is(err, core.ErrDuplicatedKey) {
		t.Errorf("重复 Note 应当报 ErrDuplicatedKey, 实际 %v", err)
	}
	// NULL 不算重复: 再加两行 Note=NULL 都该进得去
	for _, nm := range []string{"Eve", "Fay"} {
		if _, err := btbl.InsertRow([]types.FieldValue{
			{FieldName: []byte("Name"), Value: []byte(nm)},
			{FieldName: []byte("Note"), Value: nil},
		}); err != nil {
			t.Errorf("Note=NULL 的 %s 应当能插入 (唯一索引对 NULL 免除), 实际 %v", nm, err)
		}
	}
}

// TestSQLiteUniqueConstraintToVRTX 外部库用【列级 UNIQUE 约束】时, 约束也要跟着走。
//
// 这种写法在 SQLite 里没有显式索引, 只生成一条 sqlite_autoindex_* (origin='u')。
// 原实现只收 origin='c' 并跳过它, 注释理由是"VRTX 没有唯一约束的概念" —— 现在有了,
// 跳过就等于把用户的约束悄悄删掉。名字也不能照搬 (sqlite_ 前缀是保留的), 要改。
func TestSQLiteUniqueConstraintToVRTX(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "ext.sqlite")
	dstPath := filepath.Join(dir, "ext.db")

	db, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	// code 用 INTEGER: TEXT 落到 VRTX 是动态列 (堆存), 动态列不能当索引键
	// (引擎会拒, migrate 记进 SkippedIndexes), 那样测的就不是唯一约束的迁移了。
	for _, q := range []string{
		`CREATE TABLE t (id INTEGER PRIMARY KEY, code INTEGER UNIQUE, note TEXT)`,
		`INSERT INTO t (id, code, note) VALUES (1, 100, 'x')`,
		`INSERT INTO t (id, code, note) VALUES (2, 200, NULL)`,
	} {
		if _, err := db.Exec(q); err != nil {
			t.Fatalf("%s: %v", q, err)
		}
	}
	_ = db.Close()

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	sc, err := src.Schema("t")
	if err != nil {
		t.Fatalf("读外部库结构: %v", err)
	}
	var uniq *IndexSchema
	for i := range sc.Indexes {
		if sc.Indexes[i].Unique {
			uniq = &sc.Indexes[i]
		}
	}
	if uniq == nil {
		t.Fatalf("code 的唯一约束没读出来, 索引: %v", sc.Indexes)
	}
	if len(uniq.Columns) != 1 || uniq.Columns[0] != "code" {
		t.Fatalf("唯一索引的分量不对: %v", uniq.Columns)
	}
	if strings.HasPrefix(uniq.Name, "sqlite_") {
		t.Errorf("唯一索引名不该照搬 SQLite 的保留名: %s", uniq.Name)
	}

	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Run(src, dst, quiet()); err != nil {
		t.Fatalf("SQLite→VRTX: %v", err)
	}
	_ = src.Close()
	_ = dst.Close()

	back, err := core.Open(dstPath)
	if err != nil {
		t.Fatal(err)
	}
	defer back.Close()
	tbl, ok := back.FindTable("t")
	if !ok {
		t.Fatal("VRTX 里没有 t 表")
	}

	// 1) 唯一性落到索引上, 且列上补出了"唯一列"的声明位
	var uniqueIdx core.IndexMeta
	hasUnique := false
	for _, m := range tbl.IndexMeta().GetIndexes() {
		if m.Unique() {
			uniqueIdx, hasUnique = m, true
		}
	}
	if !hasUnique {
		t.Fatalf("唯一约束没迁过来: 库里一条唯一索引都没有")
	}
	codeFid := -1
	for i, f := range tbl.GetFields() {
		if util.BytesToString(f.FieldName) == "code" {
			codeFid = i
		}
	}
	if codeFid < 0 {
		t.Fatal("找不到 code 列")
	}
	if !tbl.GetFields()[codeFid].IsUnique() {
		t.Errorf("code 列应当带上唯一声明位")
	}
	if got := uniqueIdx.Fields(); len(got) != 1 || got[0] != byte(codeFid) {
		t.Errorf("唯一索引的分量应当是 code(%d), 实际 %v", codeFid, got)
	}

	// 2) 约束真的会拦: 重复 code 被拒
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(3)},
		{FieldName: []byte("code"), Value: uint64(100)},
	}); !errors.Is(err, core.ErrDuplicatedKey) {
		t.Errorf("重复 code 应当报 ErrDuplicatedKey, 实际 %v", err)
	}

	// 3) NULL 也保住了 (note 第 2 行是 NULL)
	row, err := tbl.FindByPK(2)
	if err != nil {
		t.Fatalf("读回 pk=2: %v", err)
	}
	for _, fv := range row {
		if util.BytesToString(fv.FieldName) == "note" {
			t.Errorf("note 本该是 NULL (不出现), 实际 %v", fv.Value)
		}
	}
}

// TestCappedColumnCapacityReported 源库声明了超过单值上限的列必须写进迁移报告。
//
// 与 SkippedIndexes 是同一条原则: 静默把容量改小等于悄悄改结构 —— 源库写的是
// varchar(60000), 到 VRTX 端只能是条装不下 60000 字节的动态列, 不报出来用户永远不会知道。
//
// 但它与"丢索引"一样【不该让整张表陪葬】: 超限只说明这一列的上限比声明的小, 大部分数据照样
// 迁得动, 真有大值会在写入时报出列名并拒 —— 为了一列的容量放弃整张表的数据, 代价不成比例。
func TestCappedColumnCapacityReported(t *testing.T) {
	dir := t.TempDir()
	srcPath := filepath.Join(dir, "src.sqlite")
	dstPath := filepath.Join(dir, "dst.db")

	raw, err := sql.Open("sqlite", srcPath)
	if err != nil {
		t.Fatal(err)
	}
	// SQLite 不校验 varchar 的长度, 所以"声明了引擎兑现不了的容量"这种库在现实中真会出现。
	stmts := []string{
		`CREATE TABLE Wide (
			id  INTEGER PRIMARY KEY,
			big VARCHAR(60000),
			ok  VARCHAR(200)
		)`,
		`INSERT INTO Wide (id, big, ok) VALUES (1, 'small', 'fine')`,
	}
	for _, s := range stmts {
		if _, err := raw.Exec(s); err != nil {
			t.Fatalf("建样本 SQLite 库: %v", err)
		}
	}
	if err := raw.Close(); err != nil {
		t.Fatal(err)
	}

	src, err := OpenSQLite(srcPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer src.Close()
	dst, err := OpenVRTX(dstPath, quiet())
	if err != nil {
		t.Fatal(err)
	}
	defer dst.Close()

	rep, err := Run(src, dst, quiet())
	if err != nil {
		t.Fatalf("迁移不该因为容量封顶而失败: %v", err)
	}
	if len(rep.Tables) != 1 {
		t.Fatalf("应当只有 1 张表的报告, 实际 %d", len(rep.Tables))
	}
	note := rep.Tables[0].Note
	// 报告要点名到列, 并给出两个数: 用户写下的 60000 与引擎能给的上限。
	for _, want := range []string{"容量封顶", "big", "60000", strconv.Itoa(types.MaxStringSize)} {
		if !strings.Contains(note, want) {
			t.Errorf("报告里缺少 %q: %q", want, note)
		}
	}
	// 没超限的列不该被连坐 (ok 列只有 200 字节, 在定长档里)。
	if strings.Contains(note, "ok:") {
		t.Errorf("未超限的列不该出现在容量报告里: %q", note)
	}
	// 数据照迁: 容量封顶只是提示, 不是拒绝。
	if rep.Rows != 1 {
		t.Errorf("应照常搬走 1 行, 实际 %d", rep.Rows)
	}
}

// TestVRTXWriterNameSourceRowOnOversize 写失败时必须报出【源库里是哪一行】。
//
// 起因是真实迁移: steambox_forum.resource_detail 有一行 content 16862 字节, 越过引擎单值
// 上限 16360。这个值 VRTX 目前确实装不下 (Dynamicbyte 与 Blob 的单值上限一样, "多页分块 +
// 链式 next"在 types.go 里还只是将来计划), 所以该拒; 但只报"表 x 的 content 超限"不够用 ——
// 用户要回源库处理那一行 (把列改大 / 把值裁小 / 剔掉), 先得知道是哪一行。那张表的 id 恰好
// 属于"主键承载不了行号"这一种 (MySQL 的 int(11) 只有 4 字节, 行号换成了补出来的 _id), 所以
// 最靠得住的身份是源库主键列的值 —— 它降级之后仍然躺在行数据里。
func TestVRTXWriterNameSourceRowOnOversize(t *testing.T) {
	// 16862 是真实那行的长度, 这里照抄: 刚好越过 16360。
	big := strings.Repeat("x", 16862)

	for _, c := range []struct {
		name  string
		srcPK string // SourcePKColumn: 源库主键列名, 被降级成普通列时才非空
		hasPK bool
		want  string
	}{
		{"源库主键承载行号", "", true, "主键 77"},
		{"源库无整型主键 (auto-pk 补号)", "", false, "批内第 1 行"},
		{"源库主键被降级成普通列", "id", false, "源表主键 12345"},
	} {
		t.Run(c.name, func(t *testing.T) {
			v, err := OpenVRTX(filepath.Join(t.TempDir(), "t.vrtx"), quiet())
			if err != nil {
				t.Fatalf("建库: %v", err)
			}
			defer v.Close()

			schema := &TableSchema{
				Name:           "ResourceDetail",
				PKColumn:       "_id",
				PKIndexName:    "pk_ResourceDetail",
				PKIndexType:    page.IndexTypePrimary,
				SourcePKColumn: c.srcPK,
				Columns: []Column{
					{Name: "id", Type: types.FieldType_Int32},
					{Name: "_id", Type: types.FieldType_Uint64},
					{Name: "content", Type: types.FieldType_Dynamicbyte},
				},
			}
			if err := v.CreateTable(schema, false); err != nil {
				t.Fatalf("建表: %v", err)
			}
			w, err := v.Open("ResourceDetail", schema)
			if err != nil {
				t.Fatalf("打开写入端: %v", err)
			}

			err = w.Write([]Row{{
				PK: 77, HasPK: c.hasPK,
				Values: []any{int64(12345), int64(77), big},
			}})
			if err == nil {
				t.Fatal("16862 字节超过单值上限 16360, 应当报错而不是写进去")
			}
			// 四样都得有: 哪一行、哪一列、值多大、上限多少 —— 少一样用户就得回源库自己捞。
			for _, want := range []string{c.want, "content", "16862", "16360"} {
				if !strings.Contains(err.Error(), want) {
					t.Errorf("错误信息里该有 %q, 实际: %v", want, err)
				}
			}
		})
	}
}

// TestRuneSafeCut 截断位置必须落在 UTF-8 字符边界上。
//
// 文本列截断最难看的结果是留下半个汉字 (只读起来是无意义的替换字符)。判据是"被排除的
// 第一个字节是不是字符起始字节": 原值本身是合法文本时, 从字符边界切开得到的必然还是合法
// 文本。
func TestRuneSafeCut(t *testing.T) {
	zh := []byte(strings.Repeat("汉", 5)) // 15 字节, 每字 3 字节: 0-2 / 3-5 / 6-8 / 9-11 / 12-14

	// 10 落在第 4 个字 (9-11) 中间 → 退到 9。
	if got := runeSafeCut(zh, 10); got != 9 {
		t.Errorf("切在字符中间应退到边界 9, 实际 %d", got)
	}
	if !utf8.Valid(zh[:9]) {
		t.Errorf("退到边界后应当是合法 UTF-8: %q", zh[:9])
	}
	// 本就落在边界上不退。
	if got := runeSafeCut(zh, 9); got != 9 {
		t.Errorf("本就落在边界上不该退, 实际 %d", got)
	}
	// 不超上限原样返回。
	if got := runeSafeCut(zh, 15); got != 15 {
		t.Errorf("等于长度应原样返回, 实际 %d", got)
	}
	if got := runeSafeCut(zh, 99); got != 15 {
		t.Errorf("超过长度应原样返回, 实际 %d", got)
	}
	// 单字节字符: 按字节截。
	if got := runeSafeCut([]byte("abcdef"), 3); got != 3 {
		t.Errorf("ASCII 应按字节截到 3, 实际 %d", got)
	}
	// 不是文本 (整段都是续字节): 没有字符边界可退, 老实按字节截, 别截成空的。
	if got := runeSafeCut(bytes.Repeat([]byte{0x80}, 10), 5); got != 5 {
		t.Errorf("非文本数据应按字节截到 5, 实际 %d", got)
	}
}

// TestVRTXWriterTruncateOversize 打开 TruncateOversize 后, 超限的值按字符边界截断写入,
// 并且【每一行都报出来】。
//
// 与 TestVRTXWriterNameSourceRowOnOversize 是一对: 那个钉"默认报错、且报得出是哪一行",
// 这个钉"显式打开后不报错, 但必须留下可追查的记录"。两个都不能少 —— 默认关着是为了不静默
// 丢数据; 打开之后逐行报, 是为了用户还能回源库把内容补回来。
func TestVRTXWriterTruncateOversize(t *testing.T) {
	// 16380 字节 (3 的倍数), 上限 16360 落在第 5454 个字的中间 (16359 才是个边界),
	// 所以这一条同时验了"超限"与"退字符边界"。
	big := strings.Repeat("汉", 5460)

	var truncLogs []string
	opts := quiet()
	opts.Log = func(format string, args ...any) {
		if s := fmt.Sprintf(format, args...); strings.Contains(s, "超过单值上限") {
			truncLogs = append(truncLogs, s)
		}
	}
	opts.TruncateOversize = true

	v, err := OpenVRTX(filepath.Join(t.TempDir(), "t.vrtx"), opts)
	if err != nil {
		t.Fatalf("建库: %v", err)
	}
	defer v.Close()

	schema := &TableSchema{
		Name:        "ResourceDetail",
		PKColumn:    "_id",
		PKIndexName: "pk_ResourceDetail",
		PKIndexType: page.IndexTypePrimary,
		Columns: []Column{
			{Name: "_id", Type: types.FieldType_Uint64},
			{Name: "content", Type: types.FieldType_Dynamicbyte},
		},
	}
	if err := v.CreateTable(schema, false); err != nil {
		t.Fatalf("建表: %v", err)
	}
	w, err := v.Open("ResourceDetail", schema)
	if err != nil {
		t.Fatalf("打开写入端: %v", err)
	}
	if err := w.Write([]Row{{PK: 91, HasPK: true, Values: []any{int64(91), big}}}); err != nil {
		t.Fatalf("打开截断开关后不该报错: %v", err)
	}
	if err := v.Commit(); err != nil {
		t.Fatalf("提交: %v", err)
	}

	// 逐行报: 表 / 行 / 列 / 原长 / 上限 / 截后长 —— 缺一样用户就补不回那一行。
	if len(truncLogs) != 1 {
		t.Fatalf("应当逐行报出 1 条截断记录, 实际 %d 条: %v", len(truncLogs), truncLogs)
	}
	for _, want := range []string{"ResourceDetail", "主键 91", "content", "16380", "16360", "16359"} {
		if !strings.Contains(truncLogs[0], want) {
			t.Errorf("截断记录里该有 %q: %q", want, truncLogs[0])
		}
	}

	// 落盘的值确实是被截过的, 而且仍然是合法 UTF-8。
	var got []byte
	if err := v.Scan("ResourceDetail", schema, func(r Row) error {
		b, ok := r.Values[1].([]byte)
		if !ok {
			t.Fatalf("content 读回来应当是 []byte, 实际 %T", r.Values[1])
		}
		got = append([]byte(nil), b...)
		return nil
	}); err != nil {
		t.Fatalf("扫描: %v", err)
	}
	if want := []byte(big[:16359]); !bytes.Equal(got, want) {
		t.Errorf("落盘的值应是前 16359 字节 (字符边界), 实际 %d 字节", len(got))
	}
	if !utf8.Valid(got) {
		t.Error("截断后的值必须是合法 UTF-8 —— 不能留半个汉字")
	}
}

// TestVRTXWriterTruncateOversizeOffByDefault -truncate-oversize 默认关着: 不开就不许截。
//
// 与 StrictValue (-loose) 的分工也要钉住: 那两个开关管"值装不进这一列的宽度", 这个只管
// "超过引擎单值上限"。所以只开 -loose 时, 超限值照样要报错 —— 否则用户以为自己在"宽松
// 模式"下跑, 却没意识到内容已经被砍掉了一截。
func TestVRTXWriterTruncateOversizeOffByDefault(t *testing.T) {
	big := strings.Repeat("x", 16862)

	for _, c := range []struct {
		name     string
		loose    bool // StrictValue = !loose
		truncate bool
		wantErr  bool
	}{
		{"两个开关都关", false, false, true},
		{"只开 -loose", true, false, true},
		{"只开 -truncate-oversize", false, true, false},
	} {
		t.Run(c.name, func(t *testing.T) {
			opts := quiet()
			opts.StrictValue = !c.loose
			opts.TruncateOversize = c.truncate

			v, err := OpenVRTX(filepath.Join(t.TempDir(), "t.vrtx"), opts)
			if err != nil {
				t.Fatalf("建库: %v", err)
			}
			defer v.Close()

			schema := &TableSchema{
				Name:        "Wide",
				PKColumn:    "_id",
				PKIndexName: "pk_Wide",
				PKIndexType: page.IndexTypePrimary,
				Columns: []Column{
					{Name: "_id", Type: types.FieldType_Uint64},
					{Name: "content", Type: types.FieldType_Dynamicbyte},
				},
			}
			if err := v.CreateTable(schema, false); err != nil {
				t.Fatalf("建表: %v", err)
			}
			w, err := v.Open("Wide", schema)
			if err != nil {
				t.Fatalf("打开写入端: %v", err)
			}
			err = w.Write([]Row{{PK: 1, HasPK: true, Values: []any{int64(1), big}}})
			if c.wantErr && err == nil {
				t.Fatal("没开 -truncate-oversize 就不该动这个值, 应当报错")
			}
			if !c.wantErr && err != nil {
				t.Fatalf("开了 -truncate-oversize 不该报错: %v", err)
			}
		})
	}
}