diff --git a/server/im_http.go b/server/im_http.go index 9bf985e..f3ff8c0 100644 --- a/server/im_http.go +++ b/server/im_http.go @@ -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}) }