package queues import ( "context" "encoding/json" "hotgo/internal/consts" "hotgo/internal/library/imkafka" "hotgo/internal/library/queue" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" ) func init() { queue.RegisterConsumer(ImAuditEvent) } var ImAuditEvent = &qImAuditEvent{} type qImAuditEvent struct{} func (q *qImAuditEvent) GetTopic() string { return consts.QueueImAuditEventTopic } func (q *qImAuditEvent) Handle(ctx context.Context, mqMsg queue.MqMsg) (err error) { var envelope imkafka.Envelope if err = json.Unmarshal(mqMsg.Body, &envelope); err != nil { return err } payload := map[string]any{} 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() return err }