fix im business event delivery

server/db/mysql/tenant_adapter.go

@@ -7191,11 +7191,11 @@ func (a *adapter) TenantFriendRequestCreate(id t.TenantID, req *t.FriendRequest)
 	eventType := "friend.request.created"
 	eventID := t.StableEventID("relation", req.RequestID, eventType, req.TargetID.UserId())
 	if err = insertSystemNotificationTx(tx, req.TargetID, t.StableEventID(eventID, "notification"), eventType,
-		"新的好友申请", req.Message, map[string]any{"user_id": req.ApplicantID.UserId()}); err != nil {
+		"新的好友申请", req.Message, map[string]any{"type": "friend_request", "request_id": req.RequestID, "user_id": req.ApplicantID.UserId()}); err != nil {
 		return nil, err
 	}
 	if err = insertBusinessEventTx(tx, eventID, "user", req.TargetID.UserId(), req.ApplicantID, eventType,
-		map[string]any{"user_id": req.ApplicantID.UserId()}); err != nil {
+		map[string]any{"request_id": req.RequestID, "user_id": req.ApplicantID.UserId()}); err != nil {
 		return nil, err
 	}
 	if err = tx.Commit(); err != nil {
@@ -7539,15 +7539,15 @@ func (a *adapter) TenantFriendRequestAccept(id t.TenantID, requestID string, act
 	eventType := "friend.request." + t.FriendRequestAccepted
 	applicantEventID := t.StableEventID("relation", row.RequestID+":"+t.FriendRequestAccepted, eventType, applicant.UserId())
 	if err = insertSystemNotificationTx(tx, applicant, t.StableEventID(applicantEventID, "notification"), eventType,
-		"好友申请已处理", t.FriendRequestAccepted, map[string]any{"user_id": actor.UserId()}); err != nil {
+		"好友申请已处理", t.FriendRequestAccepted, map[string]any{"type": "friend_request", "request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
 		return nil, err
 	}
 	if err = insertBusinessEventTx(tx, applicantEventID, "user", applicant.UserId(), actor, eventType,
-		map[string]any{"user_id": actor.UserId()}); err != nil {
+		map[string]any{"request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
 		return nil, err
 	}
 	if err = insertBusinessEventTx(tx, t.StableEventID("relation", row.RequestID+":"+t.FriendRequestAccepted+":self", eventType, actor.UserId()),
-		"user", actor.UserId(), actor, eventType, map[string]any{"user_id": actor.UserId()}); err != nil {
+		"user", actor.UserId(), actor, eventType, map[string]any{"request_id": row.RequestID, "user_id": actor.UserId(), "topic": topic}); err != nil {
 		return nil, err
 	}
 	if err = tx.Commit(); err != nil {
@@ -7596,15 +7596,15 @@ func (a *adapter) TenantFriendRequestResolve(id t.TenantID, requestID string, ac
 	eventType := "friend.request." + status
 	eventID := t.StableEventID("relation", req.RequestID+":"+status, eventType, req.ApplicantID.UserId())
 	if err = insertSystemNotificationTx(tx, req.ApplicantID, t.StableEventID(eventID, "notification"), eventType,
-		"好友申请已处理", status, map[string]any{"user_id": actor.UserId()}); err != nil {
+		"好友申请已处理", status, map[string]any{"type": "friend_request", "request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
 		return nil, err
 	}
 	if err = insertBusinessEventTx(tx, eventID, "user", req.ApplicantID.UserId(), actor, eventType,
-		map[string]any{"user_id": actor.UserId()}); err != nil {
+		map[string]any{"request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
 		return nil, err
 	}
 	if err = insertBusinessEventTx(tx, t.StableEventID("relation", req.RequestID+":"+status+":self", eventType, actor.UserId()),
-		"user", actor.UserId(), actor, eventType, map[string]any{"user_id": actor.UserId()}); err != nil {
+		"user", actor.UserId(), actor, eventType, map[string]any{"request_id": req.RequestID, "user_id": actor.UserId()}); err != nil {
 		return nil, err
 	}
 	if err = tx.Commit(); err != nil {

server/im_business_events.go

@@ -149,6 +149,13 @@ func dispatchIMBusinessEvents(tenantID types.TenantID) {
 			TenantID: tenantID, RcptTo: event.Target, AsUser: from, Timestamp: event.CreatedAt,
 			Info: &MsgServerInfo{Topic: event.Target, From: from, What: "im", Event: event.EventType, Payload: payload},
 		}
+		if deliverIMBusinessEventToUserSessions(tenantID, event, message) {
+			if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil {
+				logs.Warn.Printf("business outbox: mark delivered event=%s: %v", event.EventID, err)
+				return
+			}
+			continue
+		}
 		select {
 		case globals.hub.routeSrv <- message:
 			if err = store.HTTPInfrastructure.MarkEventDelivered(tenantID, event.EventID, now); err != nil {
@@ -192,3 +199,24 @@ func startIMExternalBusinessEventPoller() func() {
 	}()
 	return func() { close(stop) }
 }
+
+func deliverIMBusinessEventToUserSessions(tenantID types.TenantID, event types.BusinessEvent, message *ServerComMessage) bool {
+	if event.TargetType != "user" || globals.sessionStore == nil || message == nil {
+		return false
+	}
+	uid := types.ParseUserId(event.Target)
+	if uid.IsZero() {
+		return false
+	}
+	delivered := false
+	globals.sessionStore.Range(func(_ string, session *Session) bool {
+		if session == nil || session.uid != uid || session.tenantID != tenantID {
+			return true
+		}
+		if session.queueOut(message.copy()) {
+			delivered = true
+		}
+		return true
+	})
+	return delivered
+}

server/im_business_events_test.go

@@ -75,3 +75,46 @@ func TestDispatchRegistrationInviteUsageEvent(t *testing.T) {
 		t.Fatal("invite usage event was routed as a regular client event")
 	}
 }
+
+func TestDispatchUserBusinessEventToOnlineSession(t *testing.T) {
+	oldInfrastructure := store.HTTPInfrastructure
+	oldHub, oldSessions := globals.hub, globals.sessionStore
+	defer func() {
+		store.HTTPInfrastructure = oldInfrastructure
+		globals.hub, globals.sessionStore = oldHub, oldSessions
+	}()
+
+	uid := types.ParseUserId("usrv97xq86jw3c")
+	infrastructure := &passwordResetEventStore{}
+	infrastructure.events = []types.BusinessEvent{{
+		EventID: "friend-request-1", EventType: "friend.request.created",
+		TargetType: "user", Target: uid.UserId(), Payload: []byte(`{"request_id":"frq_1","user_id":"usrSrc"}`),
+		CreatedAt: time.Now().UTC(),
+	}}
+	store.HTTPInfrastructure = infrastructure
+	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
+	globals.sessionStore = NewSessionStore(time.Minute)
+	session := &Session{sid: "s1", proto: WEBSOCK, uid: uid, tenantID: types.TenantID(1), send: make(chan any, 1)}
+	globals.sessionStore.sessCache[session.sid] = session
+
+	dispatchIMBusinessEvents(types.TenantID(1))
+
+	if len(infrastructure.delivered) != 1 || infrastructure.delivered[0] != "friend-request-1" {
+		t.Fatalf("business event was not marked delivered: %v", infrastructure.delivered)
+	}
+	if len(globals.hub.routeSrv) != 0 {
+		t.Fatal("online user event should be delivered directly to session without topic routing")
+	}
+	select {
+	case raw := <-session.send:
+		msg, ok := raw.(*ServerComMessage)
+		if !ok || msg.Info == nil {
+			t.Fatalf("unexpected outbound message: %#v", raw)
+		}
+		if msg.Info.What != "im" || msg.Info.Event != "friend.request.created" || msg.Info.Topic != uid.UserId() {
+			t.Fatalf("unexpected info event: %#v", msg.Info)
+		}
+	default:
+		t.Fatal("online user session did not receive business info event")
+	}
+}

server/im_http.go

@@ -3892,6 +3892,7 @@ func serveIMFriendRequestCreate(wrt http.ResponseWriter, req *http.Request) {
 		writeIMRelationError(wrt, req, err)
 		return
 	}
+	dispatchIMBusinessEvents(principal.TenantID)
 	if imReliableHTTPEventsEnabled(req) {
 		if err = publishIMRelationEvent(principal.TenantID, principal.Uid, target, created.RequestID,
 			"friend.request.created", "新的好友申请", message); err != nil {
@@ -5548,6 +5549,7 @@ func serveIMFriendRequestAction(wrt http.ResponseWriter, req *http.Request) {
 		writeIMRelationError(wrt, req, err)
 		return
 	}
+	dispatchIMBusinessEvents(principal.TenantID)
 	if imReliableHTTPEventsEnabled(req) {
 		eventType := "friend.request." + status
 		if err = publishIMRelationEvent(principal.TenantID, principal.Uid, updated.ApplicantID,