diff --git a/server/internal/logic/sys/im_kafka_audit.go b/server/internal/logic/sys/im_kafka_audit.go index 6656bf1..1e187a8 100644 --- a/server/internal/logic/sys/im_kafka_audit.go +++ b/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 } diff --git a/server/internal/queues/im_audit_event.go b/server/internal/queues/im_audit_event.go index 8d122ff..dadb6b4 100644 --- a/server/internal/queues/im_audit_event.go +++ b/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 } diff --git a/server/internal/queues/im_dead_letter.go b/server/internal/queues/im_dead_letter.go index e63729f..b1dd6fc 100644 --- a/server/internal/queues/im_dead_letter.go +++ b/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 }