feat: 升级 samber/do v1 → v2
This commit is contained in:
@@ -0,0 +1,426 @@
|
||||
package base
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.fsdpf.net/go/db"
|
||||
"git.fsdpf.net/go/db/exp"
|
||||
"git.fsdpf.net/go/db/schema"
|
||||
"git.fsdpf.net/go/req"
|
||||
"github.com/samber/lo"
|
||||
)
|
||||
|
||||
var changeLogOnce sync.Map // key: Conn，每个数据库连接只建一次 _change_log 表
|
||||
|
||||
// isLocalDB 判断是否为本地文件型数据库（LastInsertId 返回最后一条而非第一条）
|
||||
func isLocalDB(dialect string) bool {
|
||||
return dialect == "sqlite3" || dialect == "duckdb"
|
||||
}
|
||||
|
||||
type ResourceHooks struct {
|
||||
res Resource
|
||||
u req.User
|
||||
tx *db.TxDatabase
|
||||
snapshots []map[string]any // UPDATE/DELETE 前预查的行数据
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) isTracked() bool {
|
||||
if len(rh.res.HistoryRoles) == 0 || rh.res.IsVirtual() {
|
||||
return false
|
||||
}
|
||||
if lo.Contains(rh.res.HistoryRoles, "00000000-0000-0000-0000-000000000000") {
|
||||
return true
|
||||
}
|
||||
return lo.Some(rh.res.HistoryRoles, rh.u.Roles())
|
||||
}
|
||||
|
||||
// autoCreateChangeLog 在当前连接上建 _change_log 表，每个连接只执行一次
|
||||
func (rh *ResourceHooks) autoCreateChangeLog() error {
|
||||
conn := rh.res.DB()
|
||||
v, _ := changeLogOnce.LoadOrStore(rh.res.Conn, &sync.Once{})
|
||||
var err error
|
||||
v.(*sync.Once).Do(func() {
|
||||
sb := schema.New(conn)
|
||||
err = sb.Create("_change_log", func(bp *schema.Blueprint) {
|
||||
bp.Comment = "资源数据变更记录"
|
||||
bp.BigIncrements("id").AutoIncrement().Comment("ID")
|
||||
bp.Boolean("enabled").Default("1").Comment("是否有效")
|
||||
bp.Char("created_user", 36).Default("00000000-0000-0000-0000-000000000000").Comment("操作用户")
|
||||
bp.Char("owned_user", 36).Default("00000000-0000-0000-0000-000000000000").Comment("拥有者")
|
||||
bp.Timestamp("created_at").UseCurrent().Comment("操作时间")
|
||||
bp.Timestamp("updated_at").UseCurrent().Default(db.L("ON UPDATE CURRENT_TIMESTAMP")).Comment("更新时间")
|
||||
bp.DateTime("deleted_at").Nullable().Comment("删除时间")
|
||||
bp.Char("resource_uuid", 36).Default("").Comment("资源UUID")
|
||||
bp.String("category", 10).Default("").Comment("操作类型 INSERT/UPDATE/DELETE")
|
||||
bp.Char("trace_id", 36).Default("").Comment("请求追踪ID")
|
||||
bp.Integer("row_id").Default("0").Comment("受影响行ID")
|
||||
bp.Json("snapshot").Nullable().Comment("操作前数据快照")
|
||||
})
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// captureSnapshot 在写操作执行前，查出受影响的行存入 snapshots
|
||||
func (rh *ResourceHooks) captureSnapshot(where exp.ExpressionList) {
|
||||
sd := rh.res.DB().From(rh.res.GetTable())
|
||||
if where != nil && len(where.Expressions()) > 0 {
|
||||
sd = sd.Where(where.Expressions()...)
|
||||
}
|
||||
rows, err := sd.Executor().GetRecords()
|
||||
if err != nil {
|
||||
log.Printf("_change_log captureSnapshot err: %v", err)
|
||||
return
|
||||
}
|
||||
rh.snapshots = rows
|
||||
}
|
||||
|
||||
// writeChangeLogs 将变更写入 _change_log
|
||||
func (rh *ResourceHooks) writeChangeLogs(category string, result sql.Result) {
|
||||
if err := rh.autoCreateChangeLog(); err != nil {
|
||||
log.Printf("_change_log init err: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
traceId := rh.u.Runtime().TraceId()
|
||||
opUser := rh.u.Uuid()
|
||||
|
||||
var rows []any
|
||||
|
||||
switch category {
|
||||
case "INSERT":
|
||||
lastId, _ := result.LastInsertId()
|
||||
count, _ := result.RowsAffected()
|
||||
for i := int64(0); i < count; i++ {
|
||||
rowId := lastId + i // MySQL: lastId 是第一条
|
||||
if isLocalDB(rh.res.DB().Dialect()) {
|
||||
rowId = lastId - count + 1 + i // SQLite: lastId 是最后一条
|
||||
}
|
||||
rows = append(rows, map[string]any{
|
||||
"enabled": true,
|
||||
"created_user": opUser,
|
||||
"owned_user": opUser,
|
||||
"resource_uuid": rh.res.Uuid,
|
||||
"category": category,
|
||||
"trace_id": traceId,
|
||||
"row_id": rowId,
|
||||
"snapshot": nil,
|
||||
})
|
||||
}
|
||||
|
||||
case "UPDATE", "DELETE":
|
||||
for _, snap := range rh.snapshots {
|
||||
rowId := int64(0)
|
||||
if id, ok := snap["id"]; ok {
|
||||
rowId = toChangeLogInt64(id)
|
||||
}
|
||||
snapJSON, _ := json.Marshal(snap)
|
||||
rows = append(rows, map[string]any{
|
||||
"enabled": true,
|
||||
"created_user": opUser,
|
||||
"owned_user": opUser,
|
||||
"resource_uuid": rh.res.Uuid,
|
||||
"category": category,
|
||||
"trace_id": traceId,
|
||||
"row_id": rowId,
|
||||
"snapshot": string(snapJSON),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
if len(rows) == 0 {
|
||||
return
|
||||
}
|
||||
inserter := rh.res.DB().Insert("_change_log")
|
||||
if rh.tx != nil {
|
||||
inserter = rh.tx.Insert("_change_log")
|
||||
}
|
||||
if _, err := inserter.Rows(rows...).Executor().Exec(); err != nil {
|
||||
log.Printf("_change_log write err: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func toChangeLogInt64(v any) int64 {
|
||||
switch val := v.(type) {
|
||||
case int64:
|
||||
return val
|
||||
case int:
|
||||
return int64(val)
|
||||
case int32:
|
||||
return int64(val)
|
||||
case float64:
|
||||
return int64(val)
|
||||
case string:
|
||||
n, _ := strconv.ParseInt(val, 10, 64)
|
||||
return n
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) Before(dataset interface{}) error {
|
||||
switch d := dataset.(type) {
|
||||
case *db.SelectDataset:
|
||||
return rh.beforeSelectDataset(d)
|
||||
case *db.InsertDataset:
|
||||
return rh.beforeInsertDataset(d)
|
||||
case *db.UpdateDataset:
|
||||
return rh.beforeUpdateDataset(d)
|
||||
case *db.DeleteDataset:
|
||||
return rh.beforeDeleteDataset(d)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) After(dataset interface{}, result interface{}) error {
|
||||
if rh.res.IsVirtual() {
|
||||
// 返回鉴权后的 DB Builder
|
||||
return nil
|
||||
}
|
||||
|
||||
// 用户事件
|
||||
// if rh.res.IsHistoryRecord {
|
||||
// rh.res.onUserEvent(builder, user)
|
||||
// }
|
||||
|
||||
switch dataset.(type) {
|
||||
case *db.SelectDataset:
|
||||
case *db.InsertDataset:
|
||||
r := result.(sql.Result)
|
||||
rh.res.onResEvent("INSERT", r)
|
||||
if rh.isTracked() {
|
||||
rh.writeChangeLogs("INSERT", r)
|
||||
}
|
||||
case *db.UpdateDataset:
|
||||
r := result.(sql.Result)
|
||||
rh.res.onResEvent("UPDATE", r)
|
||||
if rh.isTracked() {
|
||||
rh.writeChangeLogs("UPDATE", r)
|
||||
}
|
||||
case *db.DeleteDataset:
|
||||
r := result.(sql.Result)
|
||||
rh.res.onResEvent("DELETE", r)
|
||||
if rh.isTracked() {
|
||||
rh.writeChangeLogs("DELETE", r)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeInsertDataset(id *db.InsertDataset) error {
|
||||
switch true {
|
||||
case id.GetClauses().HasRows():
|
||||
return rh.beforeInsertRows(id)
|
||||
case id.GetClauses().HasVals():
|
||||
return rh.beforeInsertColsVals(id)
|
||||
case id.GetClauses().HasFrom():
|
||||
return rh.beforeInsertFromQuery(id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeInsertRows(id *db.InsertDataset) error {
|
||||
rows := id.GetClauses().Rows()
|
||||
|
||||
for i := range rows {
|
||||
rowValue := reflect.ValueOf(rows[i])
|
||||
if rowValue.Kind() == reflect.Ptr {
|
||||
rowValue = rowValue.Elem()
|
||||
}
|
||||
if rowValue.Kind() == reflect.Struct {
|
||||
if row, err := exp.NewRecordFromStruct(rowValue.Interface(), true, false); err != nil {
|
||||
return err
|
||||
} else {
|
||||
rows[i] = row
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for i := range rows {
|
||||
row := rows[i].(map[string]any)
|
||||
// 格式化保存数据
|
||||
if err := rh.normalizeSaveValue(row); err != nil {
|
||||
return err
|
||||
}
|
||||
// 填充默认数据
|
||||
if err := rh.applyDefaultValue(row, true, false); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeInsertColsVals(id *db.InsertDataset) error {
|
||||
cols := id.GetClauses().Cols()
|
||||
vals := id.GetClauses().Vals()
|
||||
|
||||
colsName := []string{}
|
||||
for _, col := range cols.Columns() {
|
||||
colsName = append(colsName, col.(exp.IdentifierExpression).GetCol().(string))
|
||||
}
|
||||
|
||||
colsVal := []any{}
|
||||
for k, v := range lo.OmitByKeys(rh.getFieldsDefaultValue(true, false), colsName) {
|
||||
cols = cols.Append(db.C(k))
|
||||
colsVal = append(colsVal, v)
|
||||
}
|
||||
|
||||
for i := 0; i < len(vals); i++ {
|
||||
for j := 0; j < len(vals[i]); j++ {
|
||||
if filed, ok := rh.res.GetField(colsName[j]); ok {
|
||||
vals[i][j] = filed.ToValue(vals[i][j])
|
||||
}
|
||||
}
|
||||
vals[i] = append(vals[i], colsVal...)
|
||||
}
|
||||
|
||||
*id = *id.Cols(cols)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeInsertFromQuery(id *db.InsertDataset) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// 规范化保存数据
|
||||
// 1. 移除系统缺省字段, 如果是 insert 需要调用 applyDefaultValue
|
||||
// 2. 移除系统中没有的字段
|
||||
func (rh *ResourceHooks) normalizeSaveValue(row db.Record) error {
|
||||
for k, v := range row {
|
||||
if k == "id" || k == "created_user" || k == "created_at" || k == "deleted_at" || k == "updated_at" {
|
||||
delete(row, k)
|
||||
} else if val, ok := v.(db.Expression); ok {
|
||||
row[k] = val
|
||||
} else if field, ok := rh.res.GetField(k); ok {
|
||||
row[k] = field.ToValue(v)
|
||||
} else {
|
||||
delete(row, k)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// 填充默认数据
|
||||
func (rh *ResourceHooks) applyDefaultValue(row db.Record, forInsert, forUpdate bool) error {
|
||||
// 填充默认数据
|
||||
for k, v := range rh.getFieldsDefaultValue(forInsert, forUpdate) {
|
||||
if _, ok := row[k]; !ok {
|
||||
row[k] = v
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) getFieldsDefaultValue(forInsert, forUpdate bool) map[string]db.Expression {
|
||||
defs := map[string]db.Expression{}
|
||||
|
||||
if forInsert {
|
||||
for _, item := range rh.res.Fields {
|
||||
if item.GetCode() == "updated_at" || item.GetCode() == "created_user" || item.GetCode() == "owned_user" {
|
||||
continue
|
||||
}
|
||||
if len(item.Default) > 4 && strings.ToLower(item.Default[0:4]) == "sql:" {
|
||||
defs[item.GetCode()] = item.GetRawDefault()
|
||||
} else if item.DataType == req.ResJson {
|
||||
defs[item.GetCode()] = item.GetRawDefault()
|
||||
}
|
||||
}
|
||||
|
||||
if _, ok := defs["owned_user"]; !ok {
|
||||
defs["owned_user"] = db.V(rh.u.Uuid())
|
||||
}
|
||||
|
||||
defs["created_user"] = db.V(rh.u.Uuid())
|
||||
}
|
||||
|
||||
if forUpdate && isLocalDB(rh.res.DB().Dialect()) {
|
||||
defs["updated_at"] = db.L("CURRENT_TIMESTAMP")
|
||||
}
|
||||
|
||||
return defs
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeSelectDataset(sd *db.SelectDataset) error {
|
||||
sub, ex := rh.res.GetRolesCondition(rh.u)
|
||||
if ex != nil {
|
||||
*sd = *sd.Where(ex)
|
||||
} else if sub != nil {
|
||||
table := sd.GetClauses().From().Columns()[0]
|
||||
if alias, ok := table.(exp.AliasedExpression); ok {
|
||||
*sd = *sd.From(db.V(sub.Expression()).As(alias.GetAs()))
|
||||
} else {
|
||||
*sd = *sd.From(db.V(sub.Expression()).As(table))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rh *ResourceHooks) beforeUpdateDataset(ud *db.UpdateDataset) error {
|
||||
// 虚拟资源暂时不考虑鉴权
|
||||
if rh.res.IsVirtual() {
|
||||
return nil
|
||||
}
|
||||
sub, ex := rh.res.GetRolesCondition(rh.u)
|
||||
if ex != nil {
|
||||
*ud = *ud.Where(ex)
|
||||
} else if sub != nil {
|
||||
table := ud.GetClauses().Table()
|
||||
if alias, ok := table.(exp.AliasedExpression); ok {
|
||||
*ud = *ud.Where(alias.GetAs().Col("id").In(sub.Select(db.T(rh.res.GetCode()).Col("id"))))
|
||||
} else {
|
||||
*ud = *ud.Where(db.L("?.id", table).In(sub.Select(db.T(rh.res.GetCode()).Col("id"))))
|
||||
}
|
||||
}
|
||||
|
||||
// 记录变更前快照（role 条件已应用，WHERE 与实际 UPDATE 一致）
|
||||
if rh.isTracked() {
|
||||
rh.captureSnapshot(ud.GetClauses().Where())
|
||||
}
|
||||
|
||||
// 格式化保存数据
|
||||
if ud.GetClauses().HasSetValues() {
|
||||
udv := ud.GetClauses().SetValues()
|
||||
|
||||
switch data := udv.(type) {
|
||||
case map[string]any:
|
||||
return rh.normalizeSaveValue(data)
|
||||
case db.Record:
|
||||
return rh.normalizeSaveValue(data)
|
||||
default:
|
||||
return fmt.Errorf("unsupported type: %T", udv)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
func (rh *ResourceHooks) beforeDeleteDataset(dd *db.DeleteDataset) error {
|
||||
// 虚拟资源暂时不考虑鉴权
|
||||
if rh.res.IsVirtual() {
|
||||
return nil
|
||||
}
|
||||
sub, ex := rh.res.GetRolesCondition(rh.u)
|
||||
if ex != nil {
|
||||
*dd = *dd.Where(ex)
|
||||
} else if sub != nil {
|
||||
table := dd.GetClauses().From().Columns()[0]
|
||||
if alias, ok := table.(exp.AliasedExpression); ok {
|
||||
*dd = *dd.Where(alias.GetAs().Col("id").In(sub.Select(db.T(rh.res.GetCode()).Col("id"))))
|
||||
} else {
|
||||
*dd = *dd.Where(db.L("?.id", table).In(sub.Select(db.T(rh.res.GetCode()).Col("id"))))
|
||||
}
|
||||
}
|
||||
|
||||
// 记录变更前快照（role 条件已应用，WHERE 与实际 DELETE 一致）
|
||||
if rh.isTracked() {
|
||||
rh.captureSnapshot(dd.GetClauses().Where())
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user