fix: notify p2p message mutation on user channel

server/im_http.go

@@ -4234,6 +4234,32 @@ func serveIMMessageMutations(wrt http.ResponseWriter, req *http.Request) {
 	})
 }
 
+func enqueueIMP2PMessageMutationEvents(tenantID types.TenantID, topic string, actor types.Uid, seq int64, version int, mutationEvent string) error {
+	if tenantID.IsZero() || topic == "" || actor.IsZero() || seq <= 0 || version <= 0 || mutationEvent == "" {
+		return types.ErrMalformed
+	}
+	uid1, uid2, err := types.ParseP2P(topic)
+	if err != nil {
+		return nil
+	}
+	payload := map[string]any{
+		"topic":   topic,
+		"seq":     seq,
+		"version": version,
+		"event":   mutationEvent,
+	}
+	for _, uid := range []types.Uid{uid1, uid2} {
+		if uid.IsZero() {
+			continue
+		}
+		eventID := stableIMEventID("message-mutation", topic, seq, version, mutationEvent, uid.UserId())
+		if err := enqueueIMBusinessEvent(tenantID, eventID, "user", uid.UserId(), actor, "message.mutation", payload); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func serveIMMessageContext(wrt http.ResponseWriter, req *http.Request) {
 	if req.Method != http.MethodGet {
 		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
@@ -4411,6 +4437,10 @@ func serveIMMessageRecall(wrt http.ResponseWriter, req *http.Request) {
 			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
 			return
 		}
+		if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.recalled"); err != nil {
+			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
+			return
+		}
 	}
 	writeIMSuccess(wrt, req, messageMutationPayload(updated, "recalled"))
 }
@@ -4473,6 +4503,8 @@ func serveIMMessageBatchRecall(wrt http.ResponseWriter, req *http.Request) {
 			if err = enqueueIMBusinessEvent(principal.TenantID, eventID, "topic", topic, principal.Uid,
 				"message.recalled", map[string]any{"topic": topic, "seq": item.Seq, "version": updated.Version}); err != nil {
 				result["event_status"] = "enqueue_failed"
+			} else if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, item.Seq, updated.Version, "message.recalled"); err != nil {
+				result["event_status"] = "enqueue_failed"
 			}
 		}
 		items = append(items, result)
@@ -4541,6 +4573,10 @@ func serveIMMessageRevision(wrt http.ResponseWriter, req *http.Request) {
 			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
 			return
 		}
+		if err = enqueueIMP2PMessageMutationEvents(principal.TenantID, topic, principal.Uid, seq, updated.Version, "message.edited"); err != nil {
+			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
+			return
+		}
 	}
 	writeIMSuccess(wrt, req, messageMutationPayload(updated, "edited"))
 }

server/im_http_test.go

@@ -7966,6 +7966,72 @@ func TestServeIMMessageRecallNormalizesP2PUserTopic(t *testing.T) {
 	}
 }
 
+func TestEnqueueIMP2PMessageMutationEventsDeliversUserDirtyMarker(t *testing.T) {
+	oldStore := globals.sessionStore
+	defer func() { globals.sessionStore = oldStore }()
+
+	tenantID := types.TenantID(7)
+	actor := types.Uid(42)
+	peer := types.Uid(43)
+	topic := actor.P2PName(peer)
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestSession("actor-session", tenantID, actor, "me"),
+		realtimeTestSession("peer-session", tenantID, peer, "me"),
+	)
+
+	if err := enqueueIMP2PMessageMutationEvents(tenantID, topic, actor, 9, 2, "message.recalled"); err != nil {
+		t.Fatal(err)
+	}
+
+	for _, tc := range []struct {
+		sid string
+		uid types.Uid
+	}{
+		{"actor-session", actor},
+		{"peer-session", peer},
+	} {
+		sess := globals.sessionStore.Get(tc.sid)
+		select {
+		case raw := <-sess.send:
+			msg, ok := raw.(*ServerComMessage)
+			if !ok || msg.Info == nil {
+				t.Fatalf("unexpected outbound for %s: %#v", tc.sid, raw)
+			}
+			if msg.Info.Topic != tc.uid.UserId() || msg.Info.Event != "message.mutation" || msg.Info.What != "im" {
+				t.Fatalf("unexpected mutation marker for %s: %#v", tc.sid, msg.Info)
+			}
+			var payload map[string]any
+			if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil {
+				t.Fatal(err)
+			}
+			if payload["topic"] != topic || payload["event"] != "message.recalled" ||
+				payload["seq"].(float64) != 9 || payload["version"].(float64) != 2 {
+				t.Fatalf("unexpected mutation payload for %s: %#v", tc.sid, payload)
+			}
+		default:
+			t.Fatalf("expected mutation marker for %s", tc.sid)
+		}
+	}
+}
+
+func TestEnqueueIMP2PMessageMutationEventsIgnoresGroupTopic(t *testing.T) {
+	oldStore := globals.sessionStore
+	defer func() { globals.sessionStore = oldStore }()
+
+	tenantID := types.TenantID(7)
+	actor := types.Uid(42)
+	globals.sessionStore = realtimeTestSessionStore(
+		realtimeTestSession("actor-session", tenantID, actor, "me"),
+	)
+
+	if err := enqueueIMP2PMessageMutationEvents(tenantID, "grpRecall", actor, 9, 2, "message.recalled"); err != nil {
+		t.Fatal(err)
+	}
+	if len(globals.sessionStore.Get("actor-session").send) != 0 {
+		t.Fatal("group topic should not emit per-user mutation markers")
+	}
+}
+
 func TestServeIMMessageRevision(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()