pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 接入后台消息治理与登录审计

Browse Source
This commit is contained in:
richjohnrambo
2026-09-07 11:53:24 +08:00
parent 90185b6c88
commit 6b20a633ce
17 changed files with 1062 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/kafka_admin_commands.go
+68 -19
View File
@@ -33,6 +33,8 @@ const (
adminCommandUserCancel = "user.account.cancel"
adminCommandUserDelete = "user.account.delete"
adminCommandUserPasswordReset = "user.password.reset"
adminCommandMessageHide = "message.hide"
adminCommandMessageDelete = "message.delete"
adminCommandKafkaDeadReplay = "kafka.dead_letter.replay"
adminCommandKafkaDeadIgnore = "kafka.dead_letter.ignore"
adminCommandKafkaDeadResolve = "kafka.dead_letter.resolve"
@@ -41,22 +43,24 @@ const (
var publishKafkaDeadLetter = publishKafkaEnvelope
type adminCommandPayload struct {
CommandID string `json:"command_id"`
ActorUserID string `json:"actor_user_id"`
Topic string `json:"topic"`
UserID string `json:"user_id"`
UserIDs []string `json:"user_ids"`
TargetID string `json:"target_user_id"`
DeviceID string `json:"device_id"`
PasswordHash string `json:"password_hash"`
Version int `json:"version"`
Reason string `json:"reason"`
DeadID string `json:"dead_id"`
Note string `json:"note"`
Hard bool `json:"hard"`
ExpiresAt *time.Time `json:"expires_at"`
Profile groupProfilePatch `json:"profile"`
Governance groupGovernancePatch `json:"governance"`
CommandID string `json:"command_id"`
ActorUserID string `json:"actor_user_id"`
Topic string `json:"topic"`
ConversationType string `json:"conversation_type"`
SeqID int64 `json:"seq_id"`
UserID string `json:"user_id"`
UserIDs []string `json:"user_ids"`
TargetID string `json:"target_user_id"`
DeviceID string `json:"device_id"`
PasswordHash string `json:"password_hash"`
Version int `json:"version"`
Reason string `json:"reason"`
DeadID string `json:"dead_id"`
Note string `json:"note"`
Hard bool `json:"hard"`
ExpiresAt *time.Time `json:"expires_at"`
Profile groupProfilePatch `json:"profile"`
Governance groupGovernancePatch `json:"governance"`
}
type groupProfilePatch struct {
@@ -166,6 +170,15 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
if eventType == adminCommandUserPasswordReset {
return executeKafkaUserPasswordResetCommand(tenantID, payload)
}
if eventType == adminCommandMessageHide || eventType == adminCommandMessageDelete {
return executeKafkaMessageAdminCommand(tenantID, envelope, payload)
}
if eventType == adminCommandUserBan || eventType == adminCommandUserUnban ||
eventType == adminCommandUserFreeze || eventType == adminCommandUserUnfreeze ||
eventType == adminCommandUserDeviceKick || eventType == adminCommandUserLogoutAll ||
eventType == adminCommandUserCancel || eventType == adminCommandUserDelete {
return executeKafkaUserGovernanceCommand(tenantID, envelope, payload, types.ZeroUid)
}
actor := parseKafkaCommandUser(payload.ActorUserID)
if actor.IsZero() {
return nil, types.ErrMalformed
@@ -187,9 +200,6 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
return nil, err
}
return map[string]any{"dead_id": payload.DeadID, "status": types.KafkaDeadStatusResolved}, nil
case adminCommandUserBan, adminCommandUserUnban, adminCommandUserFreeze, adminCommandUserUnfreeze,
adminCommandUserDeviceKick, adminCommandUserLogoutAll, adminCommandUserCancel, adminCommandUserDelete:
return executeKafkaUserGovernanceCommand(tenantID, envelope, payload, actor)
}
if store.GroupBusiness == nil {
return nil, errors.New("group business store is not initialized")
@@ -320,6 +330,45 @@ func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, p
return nil, errors.New("unsupported admin command: " + envelope.EventType)
}
func executeKafkaMessageAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload) (map[string]any, error) {
conversationType := strings.TrimSpace(payload.ConversationType)
topic := strings.TrimSpace(payload.Topic)
if (conversationType != "private" && conversationType != "group") || topic == "" || payload.SeqID <= 0 {
return nil, types.ErrMalformed
}
action := strings.TrimPrefix(strings.TrimSpace(envelope.EventType), "message.")
actorAdminID := int64(0)
if envelope.Actor != nil {
actorAdminID, _ = strconv.ParseInt(strings.TrimSpace(envelope.Actor.ID), 10, 64)
}
now := time.Now().UTC().Round(time.Millisecond)
commandID := firstNonEmpty(payload.CommandID, envelope.EventID)
actionID := "action_" + envelope.EventID
if len(actionID) > 96 {
actionID = actionID[:96]
}
requestID := strings.TrimSpace(envelope.TraceID)
if len(requestID) > 64 {
requestID = requestID[:64]
}
if err := store.SecurityAudit.ApplyMessageAction(tenantID, &types.MessageAdminAction{
ActionID: actionID, CommandID: commandID, ConversationType: conversationType,
Topic: topic, SeqID: payload.SeqID, Action: action, Reason: strings.TrimSpace(payload.Reason),
ActorAdminID: actorAdminID, RequestID: requestID, ProcessedAt: now,
}); err != nil {
return nil, err
}
if conversationType == "group" {
if err := store.InvalidateHotMessages(tenantID, topic, []types.Range{{Low: payload.SeqID}}); err != nil {
logs.Warn.Printf("message admin cache invalidation failed tenant:%d topic:%s seq:%d err:%v", tenantID, topic, payload.SeqID, err)
}
}
return map[string]any{
"conversation_type": conversationType, "topic": topic, "seq_id": payload.SeqID,
"action": action, "status": "success", "processed_at": now,
}, nil
}
func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload, actor types.Uid) (map[string]any, error) {
target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID))
if target.IsZero() {