TsDB2/analyzer/store.go

package main

import (
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"

	"git.ycl.pub/yl32940/TsDB2/DB"
)

// ===== 审计结论缓存库 =====
//
// 为什么要落盘
// ------------
// 全库审计在千万行级别要十几秒 CPU、数 GB 内存 (见 indexcheck.go 里那几张指纹集),
// 而结论只在库内容真的变了以后才需要重算。把结论存下来, 下次打开同一个库直接取,
// 既省掉一遍全库扫描, 也不用再把中间过程堆在内存里。
//
// 存在哪
// ------
// 被分析库旁边的独立库文件 "<db>.analyzer", 用**引擎自己的 ORM** 组织:
//   audit_cache  每个指纹一行 —— 摘要 (判决 / 页账 / 每表每索引统计 / 说明) 的 JSON
//   audit_issue  索引内容问题明细, 一行一条 → 可按页取, 不必整份读进内存
//   audit_unacc  孤儿页明细, 一行一条
// 明细拆表就是为了分页: 库里出了问题可能有成千上万条, 一次性塞给前端既慢又占内存。
//
// 什么时候算"同一个库"
// --------------------
// 指纹 = sha256(文件大小 + 头部页 4096 字节)。头部里有 ModifyTime / TxnID / FreeStart /
// NextFreePage / TablePage 等字段, 引擎每写一次都会动其中之一, 所以指纹一致基本等价于
// "文件没被改过"; 不一致就老老实实重算 —— 宁可多算一次, 也不能拿旧结论糊弄人。
//
// 容错
// ----
// 缓存是纯粹的加速层。任何一步出错 (文件被占 / 引擎报错 / 结构不兼容) 都只记日志并
// 退回"跑一遍全量审计", 绝不影响结论正确性。

// auditCacheRow 一次审计的摘要。Payload 是 AuditResult 的 JSON (去掉了明细列表,
// 明细在下面两张表里)。
type auditCacheRow struct {
	DB.BaseModel
	FP      string `vrtx:"maxlen:80,index"` // 库指纹
	Updated int64  // 写入时间 (Unix 秒)
	Payload []byte // 动态列: 存 JSON 本体
}

// auditIssueRow 索引内容问题明细, 供分页查看。
type auditIssueRow struct {
	DB.BaseModel
	FP     string `vrtx:"maxlen:80,index"`
	Table  string `vrtx:"maxlen:64"`
	Index  string `vrtx:"maxlen:64"`
	IType  string `vrtx:"maxlen:24"`
	Kind   string `vrtx:"maxlen:32"`
	Page   uint32
	At     int
	Detail string `vrtx:"maxlen:256"`
}

// auditUnaccRow 孤儿页明细 (既不属于任何结构、也不在空闲链上的页)。
type auditUnaccRow struct {
	DB.BaseModel
	FP   string `vrtx:"maxlen:80,index"`
	Page uint32
}

var (
	storeMu sync.Mutex // 引擎非线程安全 —— 缓存库的所有访问都串行化
	cacheFP string     // 当前库指纹 (由 openH 设置)
)

// cachePath 缓存库路径。可用 TSDB_ANALYZER_CACHE 指定到别处 (比如非持久盘)。
func cachePath(dbPath string) string {
	if v := os.Getenv("TSDB_ANALYZER_CACHE"); v != "" {
		return v
	}
	return dbPath + ".analyzer"
}

// setCacheFP 记录当前库指纹 (供分页查询定位缓存行)。
func setCacheFP(fp string) {
	cacheMu.Lock()
	cacheFP = fp
	cacheMu.Unlock()
}

// fingerprint 当前库的指纹。必须在头部页已加载 (rHdr) 之后调用。
func fingerprint() string {
	if dbF == "" || len(hdr) == 0 {
		return ""
	}
	h := sha256.New()
	var sz [8]byte
	if fi, err := os.Stat(dbF); err == nil {
		binary.LittleEndian.PutUint64(sz[:], uint64(fi.Size()))
	}
	h.Write(sz[:])
	h.Write(hdr[:])
	return hex.EncodeToString(h.Sum(nil)[:20])
}

// payloadOf 复制一份可落盘的审计结果: 明细列表与空闲页列表都摘出去。
// 注意 Free / Reach 是**指针**, 必须先复制再改, 不能就地清空 —— 那份结果可能还在用。
func payloadOf(a *AuditResult) *AuditResult {
	b := *a
	b.IdxIssues = nil
	if b.Free != nil {
		f := *b.Free
		f.Pages = nil // 空闲页由 /api/free-pages 现走链取, 不在这份里重复存
		b.Free = &f
	}
	if b.Reach != nil {
		r := *b.Reach
		r.UnaccountedIDs = nil
		b.Reach = &r
	}
	return &b
}

// cacheOpen 打开缓存库并关掉自动提交 (由我们显式 Commit)。
func cacheOpen() (*DB.DB, error) {
	db, err := DB.Open(cachePath(dbF))
	if err != nil {
		return nil, err
	}
	db.SetAutoCommitDelay(0)
	return db, nil
}

func removeCache() {
	p := cachePath(dbF)
	_ = os.Remove(p)
	_ = os.Remove(p + ".wal")
}

// storeSave 把一份**完整**的审计结论落盘。
// 整库重写: 缓存只服务于一个库路径, 全量重建比"按指纹删旧行"简单得多, 也不会留下
// 历史指纹的垃圾行。
func storeSave(fp string, a *AuditResult) {
	if fp == "" || a == nil || a.IdxPending || a.Reach == nil {
		return
	}
	payload, err := json.Marshal(payloadOf(a))
	if err != nil {
		log.Printf("[cache] 序列化失败: %v", err)
		return
	}
	if err := doSave(fp, payload, a); err != nil {
		// 结构不兼容 / 文件损坏: 删掉重建一次, 还不行就放弃缓存。
		removeCache()
		if err2 := doSave(fp, payload, a); err2 != nil {
			log.Printf("[cache] 写入失败, 本次不缓存: %v", err2)
		}
	}
}

func doSave(fp string, payload []byte, a *AuditResult) error {
	storeMu.Lock()
	defer storeMu.Unlock()

	removeCache() // 全量重建
	db, err := cacheOpen()
	if err != nil {
		return err
	}
	defer db.Close()

	meta, err := DB.UseTable[auditCacheRow](db, "audit_cache")
	if err != nil {
		return err
	}
	iss, err := DB.UseTable[auditIssueRow](db, "audit_issue")
	if err != nil {
		return err
	}
	una, err := DB.UseTable[auditUnaccRow](db, "audit_unacc")
	if err != nil {
		return err
	}

	if err := meta.Insert(&auditCacheRow{FP: fp, Updated: time.Now().Unix(), Payload: payload}); err != nil {
		return err
	}
	for i := range a.IdxIssues {
		is := &a.IdxIssues[i]
		if err := iss.Insert(&auditIssueRow{
			FP: fp, Table: is.Table, Index: is.Index, IType: is.IType, Kind: is.Kind,
			Page: is.Page, At: is.At, Detail: clip(is.Detail, 240),
		}); err != nil {
			return err
		}
	}
	for _, p := range a.Reach.UnaccountedIDs {
		if err := una.Insert(&auditUnaccRow{FP: fp, Page: p}); err != nil {
			return err
		}
	}
	return db.Commit()
}

// storeLoad 尝试取回同一指纹的审计结论。命中返回 (结果, true)。
// gen 为调用时刻的切库代数 —— 期间用户换库就别往回装了。
func storeLoad(fp string, gen uint64) (*AuditResult, bool) {
	if fp == "" {
		return nil, false
	}
	a, err := doLoad(fp)
	if err != nil || a == nil {
		if err != nil {
			log.Printf("[cache] 读取失败, 转为全量分析: %v", err)
		}
		return nil, false
	}
	a.IdxPending = false
	_ = gen
	return a, true
}

func doLoad(fp string) (*AuditResult, error) {
	storeMu.Lock()
	defer storeMu.Unlock()

	if _, err := os.Stat(cachePath(dbF)); err != nil {
		return nil, nil // 没有缓存文件 = 未命中, 不是错误
	}
	db, err := cacheOpen()
	if err != nil {
		return nil, err
	}
	defer db.Close()

	meta, err := DB.UseTable[auditCacheRow](db, "audit_cache")
	if err != nil {
		return nil, err
	}
	rows, err := meta.Where("FP", "=", fp).Find()
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		return nil, nil
	}

	a := &AuditResult{}
	if err := json.Unmarshal(rows[0].Payload, a); err != nil {
		return nil, err
	}
	if a.Reach == nil {
		a.Reach = &ReachResult{ByRole: map[string]int{}, UnaccountedByType: map[string]int{}}
	}
	if a.Free == nil {
		a.Free = &FreeChainInfo{}
	}

	if iss, err := DB.UseTable[auditIssueRow](db, "audit_issue"); err == nil {
		if rs, err := iss.Where("FP", "=", fp).Find(); err == nil {
			for _, r := range rs {
				a.IdxIssues = append(a.IdxIssues, IndexIssue{
					Table: r.Table, Index: r.Index, IType: r.IType, Kind: r.Kind,
					Page: r.Page, At: r.At, Detail: r.Detail,
				})
			}
		}
	}
	if una, err := DB.UseTable[auditUnaccRow](db, "audit_unacc"); err == nil {
		if rs, err := una.Where("FP", "=", fp).Find(); err == nil {
			for _, r := range rs {
				a.Reach.UnaccountedIDs = append(a.Reach.UnaccountedIDs, r.Page)
			}
		}
	}
	return a, nil
}

// ===== 明细分页查询 (从缓存库里按页取, 不整份读进内存) =====

// cachePageIssues 分页取索引问题明细。第二个返回值是总数。
func cachePageIssues(offset, limit int) ([]IndexIssue, int, error) {
	storeMu.Lock()
	defer storeMu.Unlock()

	cacheMu.Lock()
	fp := cacheFP
	cacheMu.Unlock()
	if fp == "" {
		return nil, 0, os.ErrNotExist
	}
	db, err := cacheOpen()
	if err != nil {
		return nil, 0, err
	}
	defer db.Close()

	t, err := DB.UseTable[auditIssueRow](db, "audit_issue")
	if err != nil {
		return nil, 0, err
	}
	total, err := t.Where("FP", "=", fp).Count()
	if err != nil {
		return nil, 0, err
	}
	rows, err := t.Where("FP", "=", fp).OrderBy("_id", false).Offset(offset).Limit(limit).Find()
	if err != nil {
		return nil, 0, err
	}
	out := make([]IndexIssue, 0, len(rows))
	for _, r := range rows {
		out = append(out, IndexIssue{
			Table: r.Table, Index: r.Index, IType: r.IType, Kind: r.Kind,
			Page: r.Page, At: r.At, Detail: r.Detail,
		})
	}
	return out, int(total), nil
}

// cachePageUnacc 分页取孤儿页明细。第二个返回值是总数。
func cachePageUnacc(offset, limit int) ([]uint32, int, error) {
	storeMu.Lock()
	defer storeMu.Unlock()

	cacheMu.Lock()
	fp := cacheFP
	cacheMu.Unlock()
	if fp == "" {
		return nil, 0, os.ErrNotExist
	}
	db, err := cacheOpen()
	if err != nil {
		return nil, 0, err
	}
	defer db.Close()

	t, err := DB.UseTable[auditUnaccRow](db, "audit_unacc")
	if err != nil {
		return nil, 0, err
	}
	total, err := t.Where("FP", "=", fp).Count()
	if err != nil {
		return nil, 0, err
	}
	rows, err := t.Where("FP", "=", fp).OrderBy("_id", false).Offset(offset).Limit(limit).Find()
	if err != nil {
		return nil, 0, err
	}
	out := make([]uint32, 0, len(rows))
	for _, r := range rows {
		out = append(out, r.Page)
	}
	return out, int(total), nil
}

// clip 把字符串截到 n 字节以内 (按 rune 边界, 不切出半个汉字)。
func clip(s string, n int) string {
	if len(s) <= n {
		return s
	}
	rs := []rune(s)
	for len(string(rs)) > n {
		rs = rs[:len(rs)-1]
	}
	return string(rs)
}

// cacheFileInfo 缓存库的位置与大小, 仅供日志/调试。
func cacheFileInfo() (string, int64) {
	p := cachePath(dbF)
	if dbF == "" {
		return "", 0
	}
	if fi, err := os.Stat(p); err == nil {
		return filepath.Clean(p), fi.Size()
	}
	return filepath.Clean(p), 0
}