fix: use unprefixed kafka audit tables

server/internal/logic/sys/im_kafka_audit.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"hotgo/internal/model/input/sysin"
 	"hotgo/internal/service"
+	"strings"
 
 	"github.com/gogf/gf/v2/errors/gerror"
 	"github.com/gogf/gf/v2/frame/g"
@@ -23,64 +24,117 @@ func (s *sSysImKafkaAudit) List(ctx context.Context, in *sysin.ImKafkaAuditListI
 	if err = in.Filter(ctx); err != nil {
 		return
 	}
-	mod := g.DB().Model("im_audit_events").Safe().Ctx(ctx).
-		Fields("id", "event_id", "tenant_code", "event_type", "biz_type", "biz_id", "actor_type", "actor_id", "action", "resource", "resource_id", "result", "reason", "occurred_at", "received_at")
-	if in.TenantCode != "" {
-		mod = mod.Where("tenant_code", in.TenantCode)
+	where, args := buildImKafkaAuditWhere(in)
+	totalValue, err := g.DB().GetValue(ctx, "SELECT COUNT(*) FROM im_audit_events "+where, args...)
+	if err != nil {
+		err = gerror.Wrap(err, "获取IM Kafka审计事件总数失败，请稍后重试！")
+		return
 	}
-	if in.EventId != "" {
-		mod = mod.WhereLike("event_id", "%"+in.EventId+"%")
+	totalCount = totalValue.Int()
+
+	page := in.Page
+	if page <= 0 {
+		page = 1
 	}
-	if in.EventType != "" {
-		mod = mod.WhereLike("event_type", "%"+in.EventType+"%")
+	perPage := in.PerPage
+	if perPage <= 0 {
+		perPage = 20
 	}
-	if in.BizType != "" {
-		mod = mod.Where("biz_type", in.BizType)
-	}
-	if in.BizId != "" {
-		mod = mod.WhereLike("biz_id", "%"+in.BizId+"%")
-	}
-	if in.ActorType != "" {
-		mod = mod.Where("actor_type", in.ActorType)
-	}
-	if in.ActorId != "" {
-		mod = mod.WhereLike("actor_id", "%"+in.ActorId+"%")
-	}
-	if in.Action != "" {
-		mod = mod.WhereLike("action", "%"+in.Action+"%")
-	}
-	if in.Resource != "" {
-		mod = mod.Where("resource", in.Resource)
-	}
-	if in.ResourceId != "" {
-		mod = mod.WhereLike("resource_id", "%"+in.ResourceId+"%")
-	}
-	if in.Result != "" {
-		mod = mod.Where("result", in.Result)
-	}
-	if len(in.CreatedAt) == 2 {
-		mod = mod.WhereBetween("received_at", in.CreatedAt[0], in.CreatedAt[1])
-	}
-	mod = mod.Page(in.Page, in.PerPage).OrderDesc("id")
-	if err = mod.ScanAndCount(&list, &totalCount, false); err != nil {
+	offset := (page - 1) * perPage
+	query := `SELECT id,event_id,tenant_code,event_type,biz_type,biz_id,actor_type,actor_id,
+		action,resource,resource_id,result,reason,occurred_at,received_at
+		FROM im_audit_events ` + where + ` ORDER BY id DESC LIMIT ? OFFSET ?`
+	queryArgs := append(append([]any{}, args...), perPage, offset)
+	rows, err := g.DB().GetAll(ctx, query, queryArgs...)
+	if err != nil {
 		err = gerror.Wrap(err, "获取IM Kafka审计事件列表失败，请稍后重试！")
+		return
+	}
+	if err = rows.Structs(&list); err != nil {
+		err = gerror.Wrap(err, "解析IM Kafka审计事件列表失败，请稍后重试！")
 	}
 	return
 }
 
+func buildImKafkaAuditWhere(in *sysin.ImKafkaAuditListInp) (string, []any) {
+	parts := []string{"1=1"}
+	args := []any{}
+	if in.TenantCode != "" {
+		parts = append(parts, "tenant_code=?")
+		args = append(args, in.TenantCode)
+	}
+	if in.EventId != "" {
+		parts = append(parts, "event_id LIKE ?")
+		args = append(args, "%"+in.EventId+"%")
+	}
+	if in.EventType != "" {
+		parts = append(parts, "event_type LIKE ?")
+		args = append(args, "%"+in.EventType+"%")
+	}
+	if in.BizType != "" {
+		parts = append(parts, "biz_type=?")
+		args = append(args, in.BizType)
+	}
+	if in.BizId != "" {
+		parts = append(parts, "biz_id LIKE ?")
+		args = append(args, "%"+in.BizId+"%")
+	}
+	if in.ActorType != "" {
+		parts = append(parts, "actor_type=?")
+		args = append(args, in.ActorType)
+	}
+	if in.ActorId != "" {
+		parts = append(parts, "actor_id LIKE ?")
+		args = append(args, "%"+in.ActorId+"%")
+	}
+	if in.Action != "" {
+		parts = append(parts, "action LIKE ?")
+		args = append(args, "%"+in.Action+"%")
+	}
+	if in.Resource != "" {
+		parts = append(parts, "resource=?")
+		args = append(args, in.Resource)
+	}
+	if in.ResourceId != "" {
+		parts = append(parts, "resource_id LIKE ?")
+		args = append(args, "%"+in.ResourceId+"%")
+	}
+	if in.Result != "" {
+		parts = append(parts, "result=?")
+		args = append(args, in.Result)
+	}
+	if len(in.CreatedAt) == 2 {
+		parts = append(parts, "received_at BETWEEN ? AND ?")
+		args = append(args, in.CreatedAt[0], in.CreatedAt[1])
+	}
+	return "WHERE " + strings.Join(parts, " AND "), args
+}
+
 func (s *sSysImKafkaAudit) View(ctx context.Context, in *sysin.ImKafkaAuditViewInp) (res *sysin.ImKafkaAuditViewModel, err error) {
 	if err = in.Filter(ctx); err != nil {
 		return nil, err
 	}
-	mod := g.DB().Model("im_audit_events").Safe().Ctx(ctx).
-		Fields("id", "event_id", "tenant_code", "event_type", "biz_type", "biz_id", "actor_type", "actor_id", "action", "resource", "resource_id", "result", "reason", "CAST(payload AS CHAR) payload", "CAST(envelope AS CHAR) envelope", "occurred_at", "received_at")
+	query := `SELECT id,event_id,tenant_code,event_type,biz_type,biz_id,actor_type,actor_id,
+		action,resource,resource_id,result,reason,CAST(payload AS CHAR) payload,
+		CAST(envelope AS CHAR) envelope,occurred_at,received_at FROM im_audit_events WHERE `
+	args := []any{}
 	if in.Id > 0 {
-		mod = mod.Where("id", in.Id)
+		query += "id=?"
+		args = append(args, in.Id)
 	} else {
-		mod = mod.Where("event_id", in.EventId)
+		query += "event_id=?"
+		args = append(args, in.EventId)
 	}
-	if err = mod.Scan(&res); err != nil {
+	query += " LIMIT 1"
+	row, err := g.DB().GetOne(ctx, query, args...)
+	if err != nil {
 		return nil, gerror.Wrap(err, "获取IM Kafka审计事件详情失败，请稍后重试！")
 	}
+	if row.IsEmpty() {
+		return nil, nil
+	}
+	if err = row.Struct(&res); err != nil {
+		return nil, gerror.Wrap(err, "解析IM Kafka审计事件详情失败，请稍后重试！")
+	}
 	return
 }

server/internal/queues/im_audit_event.go

@@ -32,23 +32,12 @@ func (q *qImAuditEvent) Handle(ctx context.Context, mqMsg queue.MqMsg) (err erro
 	if len(envelope.Payload) > 0 {
 		_ = json.Unmarshal(envelope.Payload, &payload)
 	}
-	_, err = g.DB().Model("im_audit_events").Safe().Ctx(ctx).Data(g.Map{
-		"event_id":    envelope.EventID,
-		"tenant_code": envelope.TenantCode,
-		"event_type":  envelope.EventType,
-		"biz_type":    envelope.BizType,
-		"biz_id":      envelope.BizID,
-		"actor_type":  firstString(payload["actor_type"]),
-		"actor_id":    firstString(payload["actor_id"]),
-		"action":      firstString(payload["action"]),
-		"resource":    firstString(payload["resource"]),
-		"resource_id": firstString(payload["resource_id"]),
-		"result":      firstString(payload["result"]),
-		"reason":      firstString(payload["reason"]),
-		"payload":     string(envelope.Payload),
-		"envelope":    string(mqMsg.Body),
-		"occurred_at": envelope.OccurredAt,
-		"received_at": gtime.Now(),
-	}).InsertIgnore()
+	_, err = g.DB().Exec(ctx, `INSERT IGNORE INTO im_audit_events
+		(event_id,tenant_code,event_type,biz_type,biz_id,actor_type,actor_id,action,resource,resource_id,result,reason,payload,envelope,occurred_at,received_at)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,CAST(NULLIF(?, '') AS JSON),CAST(? AS JSON),?,?)`,
+		envelope.EventID, envelope.TenantCode, envelope.EventType, envelope.BizType, envelope.BizID,
+		firstString(payload["actor_type"]), firstString(payload["actor_id"]), firstString(payload["action"]),
+		firstString(payload["resource"]), firstString(payload["resource_id"]), firstString(payload["result"]),
+		firstString(payload["reason"]), string(envelope.Payload), string(mqMsg.Body), envelope.OccurredAt, gtime.Now())
 	return err
 }

server/internal/queues/im_dead_letter.go

@@ -32,22 +32,13 @@ func (q *qImDeadLetter) Handle(ctx context.Context, mqMsg queue.MqMsg) (err erro
 	if len(envelope.Payload) > 0 {
 		_ = json.Unmarshal(envelope.Payload, &payload)
 	}
-	_, err = g.DB().Model("im_kafka_dead_letter_events").Safe().Ctx(ctx).Data(g.Map{
-		"event_id":        envelope.EventID,
-		"parent_event_id": envelope.ParentEventID,
-		"tenant_code":     envelope.TenantCode,
-		"event_type":      envelope.EventType,
-		"biz_type":        envelope.BizType,
-		"biz_id":          envelope.BizID,
-		"source_service":  envelope.SourceService,
-		"target_service":  envelope.TargetService,
-		"dead_id":         stringValue(payload["dead_id"]),
-		"reason":          firstString(payload["reason"], payload["last_reason"]),
-		"message":         firstString(payload["message"], payload["last_error"]),
-		"payload":         string(envelope.Payload),
-		"envelope":        string(mqMsg.Body),
-		"received_at":     gtime.Now(),
-	}).InsertIgnore()
+	_, err = g.DB().Exec(ctx, `INSERT IGNORE INTO im_kafka_dead_letter_events
+		(event_id,parent_event_id,tenant_code,event_type,biz_type,biz_id,source_service,target_service,dead_id,reason,message,payload,envelope,received_at)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?,CAST(NULLIF(?, '') AS JSON),CAST(? AS JSON),?)`,
+		envelope.EventID, envelope.ParentEventID, envelope.TenantCode, envelope.EventType, envelope.BizType, envelope.BizID,
+		envelope.SourceService, envelope.TargetService, stringValue(payload["dead_id"]),
+		firstString(payload["reason"], payload["last_reason"]), firstString(payload["message"], payload["last_error"]),
+		string(envelope.Payload), string(mqMsg.Body), gtime.Now())
 	return err
 }