GORM（Golang）实现类似Hibernate Envers的数据库审计方案问询

核心思路 #

基于GORM的回调机制+自定义钩子，结合不同操作场景（单条/批量/条件操作）的处理逻辑，解决审计日志所需的实体ID、前后数据提取问题，同时通过上下文传递操作用户信息。

1. 定义审计日志结构体与表结构 #

适配PostgreSQL的jsonb类型存储前后数据，用字符串统一兼容不同主键类型的实体ID：

import (
    "time"
    "gorm.io/driver/postgres"
    "gorm.io/gorm"
    "github.com/jackc/pgtype"
)

type AuditLog struct {
    ID          uint           `gorm:"primaryKey"`
    EntityClass string         `gorm:"not null"` // 操作的表名
    EntityID    string         `gorm:"not null"` // 记录主键（统一转字符串）
    Action      string         `gorm:"size:10;not null"` // CREATE/UPDATE/DELETE
    ActionTaker string         `gorm:"not null"` // 操作用户ID
    PrevData    pgtype.JSONB   `gorm:"type:jsonb"` // CREATE时为NULL
    CurData     pgtype.JSONB   `gorm:"type:jsonb;not null"`
    At          time.Time      `gorm:"default:current_timestamp"`
}

func (AuditLog) TableName() string {
    return "audit_logs"
}

2. 上下文传递操作用户ID #

封装方法将用户ID存入GORM上下文，供后续回调提取：

func WithActionTaker(db *gorm.DB, userID string) *gorm.DB {
    return db.WithContext(context.WithValue(db.Statement.Context, "action_taker", userID))
}

3. 实现单条操作的审计回调 #

定义Auditable接口统一获取实体ID，避免硬编码主键字段：

type Auditable interface {
    GetEntityID() string
}

// 示例：User模型实现Auditable接口
type User struct {
    ID   uint   `gorm:"primaryKey"`
    Name string
}

func (u *User) GetEntityID() string {
    return strconv.FormatUint(uint64(u.ID), 10)
}

单条CREATE操作回调 #

func init() {
    // CREATE后生成审计日志
    gorm.DefaultCallback.Create().After("gorm:create").Register("audit:create", func(db *gorm.DB) {
        if db.Error != nil {
            return
        }
        auditable, ok := db.Statement.Model.(Auditable)
        if !ok {
            return
        }
        userID, _ := db.Statement.Context.Value("action_taker").(string)
        if userID == "" {
            return
        }
        curData, _ := json.Marshal(db.Statement.Model)
        auditLog := AuditLog{
            EntityClass: db.Statement.Table,
            EntityID:    auditable.GetEntityID(),
            Action:      "CREATE",
            ActionTaker: userID,
            CurData:     pgtype.JSONB{Bytes: curData, Valid: true},
        }
        // 跳过钩子避免递归触发审计
        db.Session(&gorm.Session{SkipHooks: true}).Create(&auditLog)
    })
}

单条UPDATE操作回调 #

需要在更新前查询原数据，存入上下文供后续使用：

// UPDATE前查询原数据
gorm.DefaultCallback.Update().Before("gorm:update").Register("audit:update_before", func(db *gorm.DB) {
    if db.Error != nil || db.Statement.Model == nil {
        return
    }
    auditable, ok := db.Statement.Model.(Auditable)
    if !ok {
        return
    }
    // 反射创建原数据实例
    originalType := reflect.TypeOf(db.Statement.Model).Elem()
    originalData := reflect.New(originalType).Interface()
    if err := db.Session(&gorm.Session{SkipHooks: true}).First(originalData, auditable.GetEntityID()).Error; err != nil {
        db.Error = err
        return
    }
    db.Statement.Context = context.WithValue(db.Statement.Context, "prev_data", originalData)
})

// UPDATE后生成审计日志
gorm.DefaultCallback.Update().After("gorm:update").Register("audit:update_after", func(db *gorm.DB) {
    if db.Error != nil {
        return
    }
    auditable, ok := db.Statement.Model.(Auditable)
    if !ok {
        return
    }
    userID, _ := db.Statement.Context.Value("action_taker").(string)
    if userID == "" {
        return
    }
    prevDataRaw := db.Statement.Context.Value("prev_data")
    prevData, _ := json.Marshal(prevDataRaw)
    curData, _ := json.Marshal(db.Statement.Model)
    auditLog := AuditLog{
        EntityClass: db.Statement.Table,
        EntityID:    auditable.GetEntityID(),
        Action:      "UPDATE",
        ActionTaker: userID,
        PrevData:    pgtype.JSONB{Bytes: prevData, Valid: true},
        CurData:     pgtype.JSONB{Bytes: curData, Valid: true},
    }
    db.Session(&gorm.Session{SkipHooks: true}).Create(&auditLog)
})

单条DELETE操作回调 #

类似UPDATE，先查询原数据再生成日志：

// DELETE前查询原数据
gorm.DefaultCallback.Delete().Before("gorm:delete").Register("audit:delete_before", func(db *gorm.DB) {
    if db.Error != nil || db.Statement.Model == nil {
        return
    }
    auditable, ok := db.Statement.Model.(Auditable)
    if !ok {
        return
    }
    originalType := reflect.TypeOf(db.Statement.Model).Elem()
    originalData := reflect.New(originalType).Interface()
    if err := db.Session(&gorm.Session{SkipHooks: true}).First(originalData, auditable.GetEntityID()).Error; err != nil {
        if !errors.Is(err, gorm.ErrRecordNotFound) {
            db.Error = err
        }
        return
    }
    db.Statement.Context = context.WithValue(db.Statement.Context, "prev_data", originalData)
})

// DELETE后生成审计日志
gorm.DefaultCallback.Delete().After("gorm:delete").Register("audit:delete_after", func(db *gorm.DB) {
    if db.Error != nil {
        return
    }
    auditable, ok := db.Statement.Model.(Auditable)
    if !ok {
        return
    }
    userID, _ := db.Statement.Context.Value("action_taker").(string)
    if userID == "" {
        return
    }
    prevDataRaw := db.Statement.Context.Value("prev_data")
    prevData, _ := json.Marshal(prevDataRaw)
    auditLog := AuditLog{
        EntityClass: db.Statement.Table,
        EntityID:    auditable.GetEntityID(),
        Action:      "DELETE",
        ActionTaker: userID,
        PrevData:    pgtype.JSONB{Bytes: prevData, Valid: true},
    }
    db.Session(&gorm.Session{SkipHooks: true}).Create(&auditLog)
})

4. 处理批量操作与条件操作 #

批量CREATE操作 #

遍历切片中的每个实体生成日志：

gorm.DefaultCallback.Create().After("gorm:create").Register("audit:batch_create", func(db *gorm.DB) {
    if db.Error != nil {
        return
    }
    if reflect.TypeOf(db.Statement.Model).Kind() != reflect.Slice {
        return
    }
    userID, _ := db.Statement.Context.Value("action_taker").(string)
    if userID == "" {
        return
    }
    sliceVal := reflect.ValueOf(db.Statement.Model)
    var logs []AuditLog
    for i := 0; i < sliceVal.Len(); i++ {
        elem := sliceVal.Index(i).Interface()
        auditable, ok := elem.(Auditable)
        if !ok {
            continue
        }
        curData, _ := json.Marshal(elem)
        logs = append(logs, AuditLog{
            EntityClass: db.Statement.Table,
            EntityID:    auditable.GetEntityID(),
            Action:      "CREATE",
            ActionTaker: userID,
            CurData:     pgtype.JSONB{Bytes: curData, Valid: true},
        })
    }
    // 批量插入优化性能
    db.Session(&gorm.Session{SkipHooks: true}).CreateInBatches(&logs, 100)
})

条件批量UPDATE/DELETE #

这类操作无法通过GORM回调直接获取受影响记录，需封装自定义方法：

// 批量带条件更新并生成审计日志
func BatchUpdateWithAudit(db *gorm.DB, model interface{}, conds interface{}, updates interface{}) error {
    // 1. 查询符合条件的原数据
    originalType := reflect.TypeOf(model).Elem()
    originalSlice := reflect.MakeSlice(reflect.SliceOf(originalType), 0, 0)
    if err := db.Session(&gorm.Session{SkipHooks: true}).Where(conds).Find(originalSlice.Interface()).Error; err != nil {
        return err
    }
    // 2. 执行更新
    if err := db.Model(model).Where(conds).Updates(updates).Error; err != nil {
        return err
    }
    // 3. 生成审计日志
    userID, _ := db.Statement.Context.Value("action_taker").(string)
    if userID == "" {
        return nil
    }
    tableName := db.Model(model).Statement.Table
    var logs []AuditLog
    originalRecords := originalSlice.Interface().([]interface{})
    for _, record := range originalRecords {
        auditable, ok := record.(Auditable)
        if !ok {
            continue
        }
        // 查询更新后的最新数据
        curRecord := reflect.New(originalType).Interface()
        if err := db.Session(&gorm.Session{SkipHooks: true}).First(curRecord, auditable.GetEntityID()).Error; err != nil {
            continue
        }
        prevData, _ := json.Marshal(record)
        curData, _ := json.Marshal(curRecord)
        logs = append(logs, AuditLog{
            EntityClass: tableName,
            EntityID:    auditable.GetEntityID(),
            Action:      "UPDATE",
            ActionTaker: userID,
            PrevData:    pgtype.JSONB{Bytes: prevData, Valid: true},
            CurData:     pgtype.JSONB{Bytes: curData, Valid: true},
        })
    }
    db.Session(&gorm.Session{SkipHooks: true}).CreateInBatches(&logs, 100)
    return nil
}

// 批量带条件删除并生成审计日志
func BatchDeleteWithAudit(db *gorm.DB, model interface{}, conds interface{}) error {
    originalType := reflect.TypeOf(model).Elem()
    originalSlice := reflect.MakeSlice(reflect.SliceOf(originalType), 0, 0)
    if err := db.Session(&gorm.Session{SkipHooks: true}).Where(conds).Find(originalSlice.Interface()).Error; err != nil {
        return err
    }
    if err := db.Where(conds).Delete(model).Error; err != nil {
        return err
    }
    userID, _ := db.Statement.Context.Value("action_taker").(string)
    if userID == "" {
        return nil
    }
    tableName := db.Model(model).Statement.Table
    var logs []AuditLog
    originalRecords := originalSlice.Interface().([]interface{})
    for _, record := range originalRecords {
        auditable, ok := record.(Auditable)
        if !ok {
            continue
        }
        prevData, _ := json.Marshal(record)
        logs = append(logs, AuditLog{
            EntityClass: tableName,
            EntityID:    auditable.GetEntityID(),
            Action:      "DELETE",
            ActionTaker: userID,
            PrevData:    pgtype.JSONB{Bytes: prevData, Valid: true},
        })
    }
    db.Session(&gorm.Session{SkipHooks: true}).CreateInBatches(&logs, 100)
    return nil
}

5. 优化与注意事项 #

• 性能优化：批量操作时用CreateInBatches批量插入审计日志，减少DB请求。
• 软删除兼容：处理软删除模型时，需调整原数据查询条件（如Unscoped()），并标记审计日志的删除类型。
• 跳过审计：可通过上下文传递skip_audit标记，在回调中判断是否跳过日志生成。
• 错误隔离：审计日志的错误应记录到日志系统，避免影响主业务流程。
• 替代方案：若GORM回调复杂度高，可考虑PostgreSQL触发器，但无法直接获取应用层的用户ID，需通过会话变量传递。

内容的提问来源于stack exchange，提问作者Nabeel Qaiser