fix: push realtime del presence to topic sessions on message recall

The REST recall response reached only the caller, so peers kept showing the
recalled message until a full resync. broadcastIMMessageRecalled now pushes
the native {pres what:del delseq} tombstone through the hub for both single
and batch recall (empty pres topic for p2p, required by handlePresence's
xoriginal check); clients flush the message and render the recalled
placeholder. Zentao #16, #2.
This commit is contained in:
Austin
2026-09-18 17:42:09 +08:00
parent 8b20118570
commit f4cedc7d75
+48
View File
@@ -3161,6 +3161,50 @@ func serveIMMessageReceipts(wrt http.ResponseWriter, req *http.Request) {
})
}
// broadcastIMMessageRecalled pushes the native {pres what:"del" delseq} frame to
// the sessions currently attached to the topic so peers learn about a recall in
// real time (the REST response alone reaches only the caller). The wire record
// carries "low" only (no "hi"): client SDKs then flush the message from their
// local cache and leave a content-less tombstone behind, which the webapp
// renders as the usual "recalled" placeholder. Routed through the hub so the
// p2p topic rewrites the frame's topic name to each recipient's own view.
func broadcastIMMessageRecalled(tenantID types.TenantID, topic string, seqs []int64, actor types.Uid) {
if globals.hub == nil || topic == "" || actor.IsZero() {
return
}
delseq := make([]MsgRange, 0, len(seqs))
for _, seq := range seqs {
if seq > 0 {
delseq = append(delseq, MsgRange{LowId: seq})
}
}
if len(delseq) == 0 {
return
}
// P2P topics are loaded under their canonical name with xoriginal cleared,
// so the presence frame must carry an empty topic name to pass the
// handlePresence xoriginal check; any other category uses the topic name.
presTopic := topic
if types.GetTopicCat(topic) == types.TopicCatP2P {
presTopic = ""
}
select {
case globals.hub.routeSrv <- &ServerComMessage{
TenantID: tenantID,
RcptTo: topic,
Pres: &MsgServerPres{
Topic: presTopic,
What: "del",
Src: actor.UserId(),
DelSeq: delseq,
FilterIn: int(types.ModeRead),
},
}:
default:
logs.Warn.Printf("im message recall: hub presence queue full topic=%s count=%d", topic, len(delseq))
}
}
func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) {
if req.Method != http.MethodPost {
writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
@@ -3215,6 +3259,7 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) {
return
}
}
broadcastIMMessageRecalled(principal.TenantID, updated.Topic, []int64{seq}, principal.Uid)
writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled"))
}
@@ -3242,6 +3287,7 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) {
return
}
items := make([]map[string]any, 0, len(body.Items))
recalledSeqs := make([]int64, 0, len(body.Items))
recalledAt := time.Now().UTC().Round(time.Millisecond)
for _, item := range body.Items {
if item.Seq <= 0 {
@@ -3281,7 +3327,9 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) {
}
}
items = append(items, result)
recalledSeqs = append(recalledSeqs, item.Seq)
}
broadcastIMMessageRecalled(principal.TenantID, topic, recalledSeqs, principal.Uid)
writeIMSuccess(wrt, req, map[string]any{"topic": topic, "items": items})
}