diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 5f60fc6..e2515a8 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/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 { diff --git a/server/im_business_events.go b/server/im_business_events.go index b1eaa64..99bb4f4 100644 --- a/server/im_business_events.go +++ b/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 +} diff --git a/server/im_business_events_test.go b/server/im_business_events_test.go index de9583d..bf50a18 100644 --- a/server/im_business_events_test.go +++ b/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") + } +} diff --git a/server/im_http.go b/server/im_http.go index a93bd70..d17c927 100644 --- a/server/im_http.go +++ b/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,