TsDB2/DB/db.go

package DB

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

// DB ORM 层数据库入口
type DB struct {
	coreDB *core.DB
}

// Open 打开或创建数据库
func Open(path string) (*DB, error) {
	cdb, err := core.Open(path)
	if err != nil {
		return nil, fmt.Errorf("vrtx: open %s: %w", path, err)
	}
	return &DB{coreDB: cdb}, nil
}

// Close 关闭数据库
func (db *DB) Close() error {
	return db.coreDB.Close()
}

// Commit 显式提交 — 通过 WAL 保证原子性
func (db *DB) Commit() error {
	return db.coreDB.Commit()
}

// SetAutoCommitDelay 设置防抖自动提交延迟
// delay=0 表示禁用自动提交 (仅手动 Commit)
// 默认值为 1 秒: 每次写操作重置计时器, 停止操作 1 秒后自动 Commit
func (db *DB) SetAutoCommitDelay(d time.Duration) {
	db.coreDB.SetAutoCommitDelay(d)
}

// SetDirtyThreshold 设置脏页阈值（页数），超过此值自动触发 Commit
func (db *DB) SetDirtyThreshold(threshold int) {
	db.coreDB.SetDirtyThreshold(threshold)
}

// SetCommitPreset 设置提交策略预设
// preset: "batch" | "online" | "safe" | "ultra-safe"
//   - batch:      离线批量导入 (500页, 1s超时) - 高吞吐，可接受少量数据丢失风险
//   - online:     在线业务默认 (100页, 500ms超时) - 平衡性能和安全性
//   - safe:       金融/医疗关键数据 (20页, 100ms超时) - 最高安全性，性能较低
//   - ultra-safe: 极端安全场景 (40页, 50ms超时) - 极致安全，性能很低（慢5-8倍）
//
// 注意: SetCommitPreset 之后**不要再调用 SetAutoCommitDelay(0)** —— delay=0 会
// 完全禁用防抖, 让阈值自动 commit 也无效, 退化成"每行 fsync"。如需调整防抖时长,
// 用 SetAutoCommitDelay(time.Millisecond*200) 这种非零值。
func (db *DB) SetCommitPreset(preset string) {
	switch preset {
	case "batch":
		// 离线批量导入：高吞吐
		db.SetDirtyThreshold(500)          // 8MB
		db.SetAutoCommitDelay(time.Second) // 1s
	case "online":
		// 在线业务默认：平衡
		db.SetDirtyThreshold(100)                     // 1.6MB
		db.SetAutoCommitDelay(500 * time.Millisecond) // 500ms
	case "safe":
		// 关键数据：最高安全
		db.SetDirtyThreshold(20)                      // 320KB
		db.SetAutoCommitDelay(100 * time.Millisecond) // 100ms
	case "ultra-safe":
		// 极端安全：每40页 Commit，断电最多丢失 ~800行
		db.SetDirtyThreshold(40)                     // 640KB
		db.SetAutoCommitDelay(50 * time.Millisecond) // 50ms
	default:
		// 未知预设，使用默认值
		db.SetDirtyThreshold(500)
		db.SetAutoCommitDelay(time.Second)
	}
}

// LockWrites 锁定写入 —— 锁的是"页落盘"(WritePage), 不是写入动作本身。
//
// 锁定期 Insert/Update/Delete 照常改内存中的页并标记脏, 只有真正要写磁盘时才会阻塞在锁上;
// 没有任何"写入被拒绝"的错误码。Commit/Flush 期间引擎内部会调用它, 目的是拿一个不再变化的
// 磁盘视图。适用场景: 批量导入前、维护操作期间、备份时 (期望这几步之间磁盘内容保持不变)。
func (db *DB) LockWrites() {
	db.coreDB.LockWrites()
}

// UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
func (db *DB) UnlockWrites() {
	db.coreDB.UnlockWrites()
}

// SetBufferPoolCapacity 设置 BufferPool 容量（页数）
func (db *DB) SetBufferPoolCapacity(capacity int) {
	db.coreDB.SetBufferPoolCapacity(capacity)
}

// BeginTransaction 开启新事务
// 返回 Transaction 对象，用于 Commit/Rollback
//
// 使用示例:
//
//	txn := db.BeginTransaction()
//	defer func() {
//	  if err := recover(); err != nil {
//	    txn.Rollback() // 异常时回滚
//	    panic(err)
//	  }
//	}()
//
//	tbl.Insert(&Record{...})
//	tbl.Update(...)
//
//	if err := txn.Commit(); err != nil {
//	  return err // 提交失败
//	}
func (db *DB) BeginTransaction() *core.Transaction {
	return db.coreDB.BeginTransaction()
}

// DebugDBState 打印数据库调试信息 (调试用)
func (db *DB) DebugDBState() {
	db.coreDB.DebugDBState()
}

// DebugTableState 打印表调试信息 (调试用)
func (db *DB) DebugTableState(tableName string) {
	db.coreDB.DebugTableState(tableName)
}

// checkNameBudget 在建表 / 建索引之前, 把自动命名额外吃掉的字节预先算清楚。
//
// 为什么必须先算:
//   - 表名和索引名各有各的定长上限: 表名 TableNameSize(48)、索引名 IndexNameSize(64);
//     而自动命名要在索引名里再加前缀 —— pk_<表名>、idx_<表名>_<列名> —— 索引名会额外多吃几个字节;
//   - 不预检的话, 表会先建成功, 失败发生在紧接着的 CreateIndex 上: 库里留下半成品的表,
//     而错误信息只有"索引名过长", 完全看不出根因其实是表名 / 列名太长。
//
// 报错里把「是哪个名字、多少字节、上限多少」一次讲清楚。
func checkNameBudget(tableName string, indexes []IndexDef) error {
	if len(tableName) > core.TableNameSize {
		return fmt.Errorf("%w: 表名 %q 有 %d 字节, 上限 %d",
			core.ErrTableNameTooLong, tableName, len(tableName), core.TableNameSize)
	}
	// 主键索引名 pk_<表名> 必须塞进 IndexNameSize。
	// 当前取值下 64-48-3 = 13 > 0, 所以这条不会被触发 (文件下方的编译期不变量保证了它);
	// 留着是为了将来收紧 IndexNameSize 时, 拦截点仍然落在建表之前,
	// 而不是回退成"表建好了、紧跟着建 pk 索引才失败"的半成品状态。
	if name := pkIndexPrefix + tableName; len(name) > core.IndexNameSize {
		return fmt.Errorf("%w: 自动主键索引名 %q 有 %d 字节 (表名 %q 的 %d 字节 + 前缀 %q), 上限 %d",
			core.ErrIndexNameTooLong, name, len(name), tableName, len(tableName), pkIndexPrefix, core.IndexNameSize)
	}
	for _, idx := range indexes {
		if name := idx.indexName(tableName); len(name) > core.IndexNameSize {
			return fmt.Errorf("%w: 自动二级索引名 %q 有 %d 字节 (上限 %d) —— 用 `vrtx:\"index:<名字>\"` 显式命名, 或缩短表名 / 首列名",
				core.ErrIndexNameTooLong, name, len(name), core.IndexNameSize)
		}
	}
	return nil
}

// 编译期不变量: 表名合法 (<= TableNameSize) 时, 自动主键索引名 pk_<表名> 必须仍然合法,
// 即 IndexNameSize >= TableNameSize + len(pkIndexPrefix)。差值为负时 uint 转换直接编译失败,
// 把这层依赖钉在编译期, 而不是靠注释和自觉。
const _ = uint(core.IndexNameSize - core.TableNameSize - len(pkIndexPrefix))

// UseTable 获取或创建泛型表
// 如果表不存在，会自动创建表和索引
// 如果表已存在，会自动同步 schema：struct 多的字段自动添加，少的字段自动删除
//
// T 必须内嵌 DB.BaseModel / DB.PrimaryModel (主键表) 或 DB.TimeSeriesModel (时序表):
// 内嵌基类既是"这是一张本框架的表"的标记, 也是对象头部增删改查的来源, 所以没有内嵌
// 基类的 T 会被直接拒绝 (ErrNotModel), 不会留下一张半成品表。主键列可由每张表用
// `vrtx:"pk"` 覆盖基类的默认键列 (_id / _ts)。
//
// 用法1: 自动推导表名（推荐）
//
//	tbl, _ := DB.UseTable[Employee](db)  // 表名自动为 "employee"
//
// 用法2: 手动指定表名
//
//	tbl, _ := DB.UseTable[Employee](db, "emp_table")  // 自定义表名
func UseTable[T any](db *DB, name ...string) (*Table[T], error) {
	var zero T
	model, err := ParseModel(&zero)
	if err != nil {
		return nil, fmt.Errorf("vrtx: parse model: %w", err)
	}

	// 自动推导表名：从类型名转为小写（如 Employee -> employee）
	tableName := typeNameToTableName(zero)
	if len(name) > 0 && name[0] != "" {
		tableName = name[0] // 用户指定的表名优先
	}

	// 建表前先把「自动命名要吃掉多少字节」算清楚。pk_ / idx_ 前缀都占用
	// core.TableNameSize 的额度, 不提前拦就会出现"表建好了、紧跟着建 pk 索引才失败"
	// 的半成品状态, 而且报错只会喊"索引名过长", 把根因 (表名太长) 指偏。
	// 表已存在的分支也要查: 新加一个自动命名的二级索引会撞同一个坑。
	if err := checkNameBudget(tableName, model.Indexes); err != nil {
		return nil, err
	}

	// 尝试查找已有表
	coreTbl, ok := db.coreDB.FindTable(tableName)
	if !ok {
		// 表不存在，创建新表
		coreTbl, err = db.coreDB.CreateTable(tableName, model.CoreFields)
		if err != nil {
			return nil, fmt.Errorf("vrtx: create table %s: %w", tableName, err)
		}
		// 创建一级索引
		idxName := pkIndexPrefix + tableName
		_, err = coreTbl.CreateIndex(idxName, model.GetIndexType(), []byte{model.CorePKField})
		if err != nil {
			return nil, fmt.Errorf("vrtx: create index for %s: %w", tableName, err)
		}
		// 自动创建标签声明的二级索引 (同名字段已合并为复合索引)
		for _, idx := range model.Indexes {
			name := idx.indexName(tableName)
			if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
				return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
			}
		}
	} else {
		// 表已存在，自动同步 schema
		if err := syncSchema(coreTbl, model); err != nil {
			return nil, err
		}
		// 检查并补充缺失的二级索引
		for _, idx := range model.Indexes {
			name := idx.indexName(tableName)
			m, found := coreTbl.IndexMeta().FindIndex(name)
			if !found {
				if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
					return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
				}
				continue
			}
			// 已存在: 校验分量列表与模型一致。
			// 不一致说明索引是按旧结构建的 (例如后来才把同名字段改成复合索引),
			// 静默沿用会一直用一个"少一列"的索引 —— 查询结果对但索引形同虚设,
			// 这里直接报错, 让人决定是重建还是改名。
			if !sameIndexFields(m, idx.coreFieldIDs()) {
				return nil, fmt.Errorf("vrtx: 索引 %s 已存在但分量列表不同 (已有 %d 列, 模型期望 %d 列), 请先删除该索引",
					name, m.FieldCnt, len(idx.coreFieldIDs()))
			}
			// 模型声明了 unique 而库里那个索引不是唯一的 → 约束根本不会生效。
			// 静默沿用等于"模型说唯一、写入却允许重复", 是最危险的一类不一致
			// (只在数据脏了以后才发现), 直接报错要求重建。
			if idx.Unique && !m.Unique() {
				return nil, fmt.Errorf("vrtx: 索引 %s 已存在但不是唯一索引 (模型声明了 unique), 请先删除该索引后重建", name)
			}
		}
	}

	tbl := &Table[T]{
		db:        db,
		coreTable: coreTbl,
		model:     model,
		name:      tableName,
	}
	tbl.refreshActiveColumns()
	return tbl, nil
}

// sameIndexFields 比较已有索引的分量列表与模型期望是否一致 (顺序敏感)
func sameIndexFields(m *core.IndexMeta, want []byte) bool {
	n := int(m.FieldCnt)
	if n != len(want) {
		return false
	}
	for i := 0; i < n; i++ {
		if m.FieldIDs[i] != want[i] {
			return false
		}
	}
	return true
}

// isPKColumn 判断列名是否是主键列。
// 主键列不参与 schema 同步: 它在建表时按模型建好, 一旦被 DropField 删掉,
// 行里的主键值就再也读不回来 —— 引擎的 pkID 只存在于行的 8 字节前缀里,
// 结构体侧的 pk 字段没有任何别的途径反查。
// 主键列既可能是内嵌基类带来的 _id / _ts, 也可能是用户 vrtx:"pk" 指定的列。
func isPKColumn(model *ModelInfo, name string) bool {
	if name == pkColumnName || name == tsColumnName {
		return true
	}
	return model.PKInStruct && name == model.PKField
}

// syncSchema 将 DB schema 同步为与 struct 一致
// struct 多 → AddField；struct 少 → DropField
//
// 除了增删列, 还会校验【已存在列的列级参数】: decimal 的 scale 与字符串列的宽度码都必须
// 与模型一致。两者都是"行布局参数":
//   - scale 不一致 → "行内字节序 == 数值序"不再成立 (见 types.FieldType_Decimal), 排序静默出错;
//   - 宽度不一致 → 模型以为的容量与库里的实际容量不同 (闸门只认库里的档位), 声明 maxlen:16
//     的列实际能写满 1024 字节, 或反之合法值被拒。
//
// 对不上时直接报错, 而不是让新写入的行用另一套参数 —— 那会让同一列里两种布局混排。
func syncSchema(coreTbl *core.Table, model *ModelInfo) error {
	// 构建 struct 字段名集合 (不含主键)
	structCols := make(map[string]FieldDef, len(model.Fields))
	for _, fd := range model.Fields {
		if isPKColumn(model, fd.ColName) {
			continue
		}
		structCols[fd.ColName] = fd
	}

	// 获取 DB 当前活跃列 (排除主键列)
	allDBCols := coreTbl.GetActiveFieldNames()
	dbCols := make([]string, 0, len(allDBCols))
	dbColSet := make(map[string]bool, len(allDBCols))
	for _, name := range allDBCols {
		if isPKColumn(model, name) {
			continue
		}
		dbCols = append(dbCols, name)
		dbColSet[name] = true
	}

	// 1. struct 有但 DB 没有 → 添加列 (主键列除外: 它由建表时确定, AddField 造不出主键)
	for _, fd := range model.Fields {
		if isPKColumn(model, fd.ColName) {
			continue
		}
		if !dbColSet[fd.ColName] {
			if err := coreTbl.AddField(types.FieldInfo{
				FieldType: fd.FieldType,
				FieldName: []byte(fd.ColName),
				Flag:      fd.Flag,
			}); err != nil {
				return fmt.Errorf("vrtx: add field %s: %w", fd.ColName, err)
			}
			continue
		}
		// 已存在的列: 校验列级参数 (decimal 的 scale / 字符串列的宽度码) 是否与模型一致。
		dbFT, _ := coreTbl.GetActiveFieldType(fd.ColName)
		dbFlag, _ := coreTbl.GetActiveFieldFlag(fd.ColName)
		dbParam := dbFlag & types.FlagScaleMask

		if fd.FieldType == types.FieldType_Decimal || dbFT == types.FieldType_Decimal {
			if fd.FieldType == types.FieldType_Decimal && dbFT == types.FieldType_Decimal &&
				dbParam != fd.Param() {
				return fmt.Errorf(
					"vrtx: 列 %s 的 decimal scale 与库中不一致 (库中 %d, 模型 %d); "+
						"同一列只能有一个 scale, 请对齐模型或迁移/重建该列",
					fd.ColName, dbParam, fd.Param())
			}
			continue
		}

		// 字符串列 (含动态列): 行内宽度必须与模型声明一致。
		//
		// 宽度决定"这一列占行内哪几字节", 库里的档位就是真正的闸门 (见 types.FieldSize):
		// 库里是 1024 字节档而模型声明 16 字节档时, 模型以为"最多 16 字节"的列其实能写满
		// 1024; 反过来库里的档位更窄, 模型认为合法的值会被引擎拒。两种都不会自己暴露,
		// 所以在这里一次性说清库中与模型各自的宽度和宽度码。
		if types.IsStringType(fd.FieldType) || types.IsStringType(dbFT) ||
			types.IsDynamic(fd.FieldType) || types.IsDynamic(dbFT) {
			dbW := types.FieldSize(dbFT, dbParam)
			modelW := fd.Width()
			if dbW != modelW {
				return fmt.Errorf(
					"vrtx: 列 %s 的行内宽度与库中不一致 (库中 %s: %d 字节/宽度码 %d, "+
						"模型 %s: %d 字节/宽度码 %d); 定宽随 schema 落盘, 对不上说明模型与库"+
						"是两次不同的声明 — 请对齐 maxlen/type, 或重建该列",
					fd.ColName, types.TypeNameOf(dbFT), dbW, dbParam,
					types.TypeNameOf(fd.FieldType), modelW, fd.Param())
			}
		}
	}

	// 2. DB 有但 struct 没有 → 删除列 (dbCols 已排除主键列)
	for _, name := range dbCols {
		if _, ok := structCols[name]; !ok {
			// 错误必须上抛: 静默继续会留下"库里有、struct 没有"的列,
			// 后续 UseTable 反复尝试删除, 库与模型的不一致也无从排查。
			// (主键列等删不掉的场景由 DropField 守卫转为明确错误。)
			if err := coreTbl.DropField(name); err != nil {
				return fmt.Errorf("同步 schema 删除列 %q: %w", name, err)
			}
		}
	}
	return nil
}

// DropTable 删除表
func (db *DB) DropTable(name string) error {
	return db.coreDB.DropTable(name)
}

// CoreDB 返回底层 core.DB (用于高级操作如创建二级索引)
func (db *DB) CoreDB() *core.DB {
	return db.coreDB
}

// VACUUM 整库重写, 让数据库文件真正缩小。
//
// 与自动 compact (40% 失效字节触发) 的区别:
//   - 自动 compact: 页内墓碑回收, 文件大小几乎不变
//   - VACUUM:       整库重写, 文件真正缩小, 归还所有空闲页给 OS
//
// 任何阶段失败 → 删除临时文件, 旧库保持不变。
// 调用后 db 句柄会被关闭并重新打开, 调用者继续使用即可。
func (db *DB) VACUUM(opts ...core.VACUUMOptions) (*core.VACUUMResult, error) {
	return db.coreDB.VACUUM(opts...)
}

// SetAutoVACUUM 开启自动 VACUUM, 周期触发。
//
// interval: 两次 VACUUM 的最小间隔 (例如 1*time.Hour)
// options:  每次 VACUUM 使用的选项
//
// 关闭: SetAutoVACUUM(0)
func (db *DB) SetAutoVACUUM(interval time.Duration, opts ...core.VACUUMOptions) {
	db.coreDB.SetAutoVACUUM(interval, opts...)
}

// LastVACUUM 返回上次自动 VACUUM 的时间
func (db *DB) LastVACUUM() time.Time {
	return db.coreDB.LastVACUUM()
}

// typeNameToTableName 从类型名推导表名（Employee -> employee）
func typeNameToTableName(v any) string {
	t := reflect.TypeOf(v)
	name := t.Name()
	if name == "" {
		return "table"
	}
	// 首字母小写：Employee -> employee, OrderItem -> orderItem
	return strings.ToLower(name[:1]) + name[1:]
}