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

fix: sync persist conversation system notices

Browse Source
This commit is contained in:
richjohnrambo
2026-09-24 11:26:27 +08:00
parent 5ce04bad05
commit 5cea4d1e2e
2 changed files with 47 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/im_business_events.go
+16 -2
View File
@@ -75,16 +75,30 @@ func createIMSystemNotification(tenantID types.TenantID, uid types.Uid, eventID,
} }
func publishIMRelationEvent(tenantID types.TenantID, actor, target types.Uid, stableID, eventType, title, summary string) error { func publishIMRelationEvent(tenantID types.TenantID, actor, target types.Uid, stableID, eventType, title, summary string) error {
return publishIMRelationEventWithAction(tenantID, actor, target, stableID, eventType, title, summary, nil)
}
func publishIMRelationEventWithAction(tenantID types.TenantID, actor, target types.Uid, stableID, eventType, title, summary string, action types.KVMap) error {
if target.IsZero() { if target.IsZero() {
return types.ErrMalformed return types.ErrMalformed
} }
eventID := stableIMEventID("relation", stableID, eventType, target.UserId()) eventID := stableIMEventID("relation", stableID, eventType, target.UserId())
payload := map[string]any{"user_id": actor.UserId()}
notificationAction := types.KVMap{"user_id": actor.UserId()}
for key, value := range action {
key = strings.TrimSpace(key)
if key == "" || key == "user_id" {
continue
}
payload[key] = value
notificationAction[key] = value
}
if target != actor && title != "" { if target != actor && title != "" {
if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), eventType, if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), eventType,
title, summary, types.KVMap{"user_id": actor.UserId()}); err != nil { title, summary, notificationAction); err != nil {
return err return err
} }
} }
return enqueueIMBusinessEvent(tenantID, eventID, "user", target.UserId(), actor, eventType, return enqueueIMBusinessEvent(tenantID, eventID, "user", target.UserId(), actor, eventType,
map[string]any{"user_id": actor.UserId()}) payload)
} }
server/im_http.go
+31 -19
View File
@@ -3916,19 +3916,22 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) {
eventType := "friend.request.created" eventType := "friend.request.created"
title := "新的好友申请" title := "新的好友申请"
content := message content := message
action := types.KVMap{}
if created != nil && created.Status == types.FriendRequestAccepted { if created != nil && created.Status == types.FriendRequestAccepted {
eventType = "friend.request.accepted" eventType = "friend.request.accepted"
title = "已成为好友" title = "已成为好友"
content = "你们已成为好友，现在可以开始聊天了。" content = "你们已成为好友，现在可以开始聊天了。"
publishIMConversationSystemNotice(principal.TenantID, principal.Uid.P2PName(target), topic := principal.Uid.P2PName(target)
action["topic"] = topic
publishIMConversationSystemNotice(principal.TenantID, topic,
"system_notice", "friend_added", content, "system_notice", "friend_added", content,
[]imSystemMessageActor{ []imSystemMessageActor{
{UserID: principal.Uid.UserId(), Click: map[string]any{"type": "user_profile", "target_id": principal.Uid.UserId()}}, {UserID: principal.Uid.UserId(), Click: map[string]any{"type": "user_profile", "target_id": principal.Uid.UserId()}},
{UserID: target.UserId(), Click: map[string]any{"type": "user_profile", "target_id": target.UserId()}}, {UserID: target.UserId(), Click: map[string]any{"type": "user_profile", "target_id": target.UserId()}},
}, nil, map[string]any{"request_id": created.RequestID}) }, nil, map[string]any{"request_id": created.RequestID})
} }
if err = publishIMRelationEvent(principal.TenantID, principal.Uid, target, created.RequestID, if err = publishIMRelationEventWithAction(principal.TenantID, principal.Uid, target, created.RequestID,
eventType, title, content); err != nil { eventType, title, content, action); err != nil {
logs.Warn.Printf("im friend request create: relation event failed request_id=%s friend_request_id=%s target=%s: %v", logs.Warn.Printf("im friend request create: relation event failed request_id=%s friend_request_id=%s target=%s: %v",
req.Header.Get("X-Request-ID"), created.RequestID, target.UserId(), err) req.Header.Get("X-Request-ID"), created.RequestID, target.UserId(), err)
} }
@@ -6210,7 +6213,8 @@ func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) {
return return
} }
if status == types.FriendRequestAccepted && imReliableHTTPEventsEnabled(req) { if status == types.FriendRequestAccepted && imReliableHTTPEventsEnabled(req) {
publishIMConversationSystemNotice(principal.TenantID, updated.ApplicantID.P2PName(updated.TargetID), topic := updated.ApplicantID.P2PName(updated.TargetID)
publishIMConversationSystemNotice(principal.TenantID, topic,
"system_notice", "friend_added", "你们已成为好友，现在可以开始聊天了。", "system_notice", "friend_added", "你们已成为好友，现在可以开始聊天了。",
[]imSystemMessageActor{ []imSystemMessageActor{
{UserID: updated.ApplicantID.UserId(), Click: map[string]any{"type": "user_profile", "target_id": updated.ApplicantID.UserId()}}, {UserID: updated.ApplicantID.UserId(), Click: map[string]any{"type": "user_profile", "target_id": updated.ApplicantID.UserId()}},
@@ -6219,13 +6223,17 @@ func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) {
} }
if imReliableHTTPEventsEnabled(req) { if imReliableHTTPEventsEnabled(req) {
eventType := "friend.request." + status eventType := "friend.request." + status
if err = publishIMRelationEvent(principal.TenantID, principal.Uid, updated.ApplicantID, action := types.KVMap{}
updated.RequestID+":"+status, eventType, "好友申请已处理", status); err != nil { if status == types.FriendRequestAccepted {
action["topic"] = updated.ApplicantID.P2PName(updated.TargetID)
}
if err = publishIMRelationEventWithAction(principal.TenantID, principal.Uid, updated.ApplicantID,
updated.RequestID+":"+status, eventType, "好友申请已处理", status, action); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
if err = publishIMRelationEvent(principal.TenantID, principal.Uid, principal.Uid, if err = publishIMRelationEventWithAction(principal.TenantID, principal.Uid, principal.Uid,
updated.RequestID+":"+status+":self", eventType, "", ""); err != nil { updated.RequestID+":"+status+":self", eventType, "", "", action); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
@@ -6254,8 +6262,9 @@ func serveIMBlockUser(wrt http.ResponseWriter, req *http.Request) {
return return
} }
if imReliableHTTPEventsEnabled(req) { if imReliableHTTPEventsEnabled(req) {
if err := publishIMRelationEvent(principal.TenantID, principal.Uid, principal.Uid, if err := publishIMRelationEventWithAction(principal.TenantID, principal.Uid, principal.Uid,
stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "relationship.blocked", "", ""); err != nil { stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "relationship.blocked", "", "",
types.KVMap{"topic": principal.Uid.P2PName(peer)}); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
@@ -6266,8 +6275,9 @@ func serveIMBlockUser(wrt http.ResponseWriter, req *http.Request) {
return return
} }
if imReliableHTTPEventsEnabled(req) { if imReliableHTTPEventsEnabled(req) {
if err := publishIMRelationEvent(principal.TenantID, principal.Uid, principal.Uid, if err := publishIMRelationEventWithAction(principal.TenantID, principal.Uid, principal.Uid,
stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "relationship.unblocked", "", ""); err != nil { stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "relationship.unblocked", "", "",
types.KVMap{"topic": principal.Uid.P2PName(peer)}); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
@@ -6364,13 +6374,14 @@ func serveIMFriendResource(wrt http.ResponseWriter, req *http.Request) {
} }
if imReliableHTTPEventsEnabled(req) { if imReliableHTTPEventsEnabled(req) {
stableID := stableIMEventID(req.Header.Get("Idempotency-Key"), principal.Uid.UserId(), peer.UserId()) stableID := stableIMEventID(req.Header.Get("Idempotency-Key"), principal.Uid.UserId(), peer.UserId())
if err := publishIMRelationEvent(principal.TenantID, principal.Uid, peer, stableID, action := types.KVMap{"topic": principal.Uid.P2PName(peer)}
"friend.deleted", "好友关系已解除", ""); err != nil { if err := publishIMRelationEventWithAction(principal.TenantID, principal.Uid, peer, stableID,
"friend.deleted", "好友关系已解除", "", action); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
if err := publishIMRelationEvent(principal.TenantID, principal.Uid, principal.Uid, stableID+":self", if err := publishIMRelationEventWithAction(principal.TenantID, principal.Uid, principal.Uid, stableID+":self",
"friend.deleted", "", ""); err != nil { "friend.deleted", "", "", action); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
@@ -6385,8 +6396,9 @@ func serveIMFriendResource(wrt http.ResponseWriter, req *http.Request) {
return return
} }
if imReliableHTTPEventsEnabled(req) { if imReliableHTTPEventsEnabled(req) {
if err := publishIMRelationEvent(principal.TenantID, principal.Uid, principal.Uid, if err := publishIMRelationEventWithAction(principal.TenantID, principal.Uid, principal.Uid,
stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "friend.remark.updated", "", ""); err != nil { stableIMEventID(req.Header.Get("Idempotency-Key"), peer.UserId()), "friend.remark.updated", "", "",
types.KVMap{"topic": principal.Uid.P2PName(peer)}); err != nil {
writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil) writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
return return
} }
@@ -10729,7 +10741,7 @@ func publishIMConversationSystemNotice(tenantID types.TenantID, topic, msgType,
}, },
Content: content, Content: content,
} }
saved, err := saveMessageMaybeKafka(context.Background(), tenantID, msg, nil, false) saved, err := store.Messages.Save(tenantID, msg, nil, false)
if err != nil && !errors.Is(err, types.ErrDuplicate) { if err != nil && !errors.Is(err, types.ErrDuplicate) {
logs.Warn.Printf("im system message: failed to save %s for topic %s: %v", template, topic, err) logs.Warn.Printf("im system message: failed to save %s for topic %s: %v", template, topic, err)
return return