Notify P2P participants after HTTP forwarding

server/im_http.go

@@ -9028,6 +9028,47 @@ func broadcastHTTPForwardMessage(tenantID types.TenantID, msg *types.Message) {
 	default:
 		logs.Warn.Printf("im message forward: hub route queue full topic=%s seq=%d", msg.Topic, msg.SeqId)
 	}
+	broadcastHTTPForwardPresence(tenantID, msg, timestamp)
+}
+
+func broadcastHTTPForwardPresence(tenantID types.TenantID, msg *types.Message, timestamp time.Time) {
+	if globals.hub == nil || msg == nil || msg.SeqId <= 0 || msg.Topic == "" {
+		return
+	}
+	uid1, uid2, err := types.ParseP2P(msg.Topic)
+	if err != nil {
+		return
+	}
+	from := types.ParseUid(msg.From).UserId()
+	for _, rcpt := range []struct {
+		uid types.Uid
+		src string
+	}{
+		{uid: uid1, src: uid2.UserId()},
+		{uid: uid2, src: uid1.UserId()},
+	} {
+		actor := from
+		if actor == rcpt.uid.UserId() {
+			actor = ""
+		}
+		select {
+		case globals.hub.routeSrv <- &ServerComMessage{
+			TenantID:  tenantID,
+			RcptTo:    rcpt.uid.UserId(),
+			Timestamp: timestamp,
+			Pres: &MsgServerPres{
+				Topic:     "me",
+				What:      "msg",
+				Src:       rcpt.src,
+				SeqId:     msg.SeqId,
+				AcsActor:  actor,
+				SkipTopic: msg.Topic,
+			},
+		}:
+		default:
+			logs.Warn.Printf("im message forward: hub presence queue full topic=%s seq=%d user=%s", msg.Topic, msg.SeqId, rcpt.uid.UserId())
+		}
+	}
 }
 
 func forwardedMessageHead(msg *types.Message, clientMid string) types.KVMap {

server/im_http_test.go

@@ -4793,7 +4793,7 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) {
 	topics := mock_store.NewMockTopicsPersistenceInterface(ctrl)
 	messages := mock_store.NewMockMessagesPersistenceInterface(ctrl)
 	kafkaBus = &kafkaRuntime{}
-	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
+	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 3)}
 	store.Subs = subs
 	store.Topics = topics
 	store.Messages = messages
@@ -4883,6 +4883,24 @@ func TestServeIMMessageForwardKafkaAssignsSeqAndBroadcasts(t *testing.T) {
 	default:
 		t.Fatal("expected forwarded message to be broadcast")
 	}
+	presByUser := make(map[string]*MsgServerPres)
+	for range 2 {
+		select {
+		case delivered := <-globals.hub.routeSrv:
+			if delivered.Pres == nil {
+				t.Fatalf("unexpected non-presence message: %#v", delivered)
+			}
+			presByUser[delivered.RcptTo] = delivered.Pres
+		default:
+			t.Fatal("expected forwarded message presence notifications")
+		}
+	}
+	if pres := presByUser[types.Uid(42).UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != targetUser.UserId() || pres.SeqId != 35 || pres.AcsActor != "" || pres.SkipTopic != targetTopic {
+		t.Fatalf("unexpected sender presence: %#v", pres)
+	}
+	if pres := presByUser[targetUser.UserId()]; pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != types.Uid(42).UserId() || pres.SeqId != 35 || pres.AcsActor != types.Uid(42).UserId() || pres.SkipTopic != targetTopic {
+		t.Fatalf("unexpected receiver presence: %#v", pres)
+	}
 }
 
 func TestServeIMMessageForwardBundleSavesForwardBundle(t *testing.T) {