From 779a44d3cceacd3ccbea21ba22aa73a00035cf31 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 16:34:36 +0000 Subject: [PATCH] feat(room-worker): publish InboxMemberEvent to local INBOX + fix inbox-worker Subjects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the publisher side of the INBOX-based search-sync pipeline landed in #109. After this, search-sync-worker's spotlight + user-room collections actually receive events end-to-end instead of spinning up with an empty feed. room-worker changes (handler.go) -------------------------------- processAddMembers, processRemoveIndividual, processRemoveOrg now split their member list by home site and publish InboxMemberEvent: - Same-site accounts → one publish per call to `chat.inbox.{site}.member_added/removed` - Cross-site accounts → existing OUTBOX publish per destination, inner payload migrated MemberAddEvent/MemberRemoveEvent → InboxMemberEvent so the remote site's search-sync sees RoomName + RoomType (needed for spotlight typeahead indexing). InboxMemberEvent is a strict JSON superset of MemberAddEvent on the fields inbox-worker reads (Accounts, RoomID, SiteID, JoinedAt, HistorySharedSince), so inbox-worker's handleMemberAdded / handleMemberRemoved continue to unmarshal into their existing types and simply ignore the extra RoomName/RoomType fields. OrgID on MemberRemoveEvent is dropped from the INBOX payload — not currently read by any consumer, and InboxMemberEvent stays focused on what search-sync needs. If inbox-worker ever needs OrgID, add it then. Remove events omit RoomName/RoomType entirely (search-sync keys its deletes by {account}_{roomID} and script-removes by roomID; neither needs room metadata) so we skip the extra DB lookup on the remove path. inbox-worker changes (main.go) ------------------------------ One-line fix: `CreateOrUpdateStream` was copying only `.Name` from `stream.Inbox()` and dropping `.Subjects`. The INBOX stream was being created with zero subjects, so any publish to `chat.inbox.{site}.*` would have failed once room-worker started publishing. Now passes both. The remote-site Sources + SubjectTransforms federation config is intentionally left out — production multi-site isn't deployed yet, and when it is we want inbox-worker (not search-sync-worker's bootstrap path) to own that config. Tests ----- - TestHandler_ProcessAddMembers_PublishesToInbox — new, verifies local INBOX publish carries RoomName/RoomType + cross-site OUTBOX payload is InboxMemberEvent. - TestHandler_ProcessRemoveMember_SelfLeave_IndividualOnly + _OwnerRemovesIndividual + _OwnerRemovesOrg — bump expected publish count +1 and assert the local INBOX publish carries the right Accounts. - TestHandler_ProcessRemoveMember_CrossSiteOutbox — assert OUTBOX payload is InboxMemberEvent (not MemberRemoveEvent) and confirm NO local INBOX publish fires when the removed user is remote. Follows PR #109. https://claude.ai/code/session_01XTmSpmv5dT6UXX7NpRdYqN --- inbox-worker/main.go | 3 +- room-worker/handler.go | 138 +++++++++++++++++++++++++------- room-worker/handler_test.go | 152 ++++++++++++++++++++++++++++++++++-- 3 files changed, 257 insertions(+), 36 deletions(-) diff --git a/inbox-worker/main.go b/inbox-worker/main.go index 203cdb534..e66616c41 100644 --- a/inbox-worker/main.go +++ b/inbox-worker/main.go @@ -147,7 +147,8 @@ func main() { inboxCfg := stream.Inbox(cfg.SiteID) if _, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{ - Name: inboxCfg.Name, + Name: inboxCfg.Name, + Subjects: inboxCfg.Subjects, }); err != nil { slog.Error("create inbox stream failed", "error", err) os.Exit(1) diff --git a/room-worker/handler.go b/room-worker/handler.go index bc7cb63cd..f333c7d80 100644 --- a/room-worker/handler.go +++ b/room-worker/handler.go @@ -312,17 +312,37 @@ func (h *Handler) processRemoveIndividual(ctx context.Context, req *model.Remove slog.Error("system message publish failed", "error", err, "roomID", req.RoomID) } - // Cross-site outbox for federated users - if user.SiteID != h.siteID { + // Publish InboxMemberEvent for downstream search-sync indexing. Goes to + // local INBOX when the removed user is on this site, cross-site OUTBOX + // when federated. Remove events don't need RoomName/RoomType (search- + // sync keys deletes by {account}_{roomID}), so they're omitted to avoid + // an extra DB lookup. + inboxEvt := model.InboxMemberEvent{ + RoomID: req.RoomID, + SiteID: h.siteID, + Accounts: []string{req.Account}, + Timestamp: now.UnixMilli(), + } + if user.SiteID == h.siteID { outbox := model.OutboxEvent{ - Type: "member_removed", + Type: model.OutboxMemberRemoved, + SiteID: h.siteID, + DestSiteID: h.siteID, + Payload: mustMarshal(inboxEvt), + Timestamp: now.UnixMilli(), + } + if err := h.publish(ctx, subject.InboxMemberRemoved(h.siteID), mustMarshal(outbox)); err != nil { + slog.Error("inbox member_removed publish failed", "error", err, "roomID", req.RoomID) + } + } else { + outbox := model.OutboxEvent{ + Type: model.OutboxMemberRemoved, SiteID: h.siteID, DestSiteID: user.SiteID, - Payload: memberEvtData, + Payload: mustMarshal(inboxEvt), Timestamp: now.UnixMilli(), } - outboxData, _ := json.Marshal(outbox) - if err := h.publish(ctx, subject.Outbox(h.siteID, user.SiteID, "member_removed"), outboxData); err != nil { + if err := h.publish(ctx, subject.Outbox(h.siteID, user.SiteID, "member_removed"), mustMarshal(outbox)); err != nil { slog.Error("outbox publish failed", "error", err, "destSiteID", user.SiteID) } } @@ -427,24 +447,50 @@ func (h *Handler) processRemoveOrg(ctx context.Context, req *model.RemoveMemberR slog.Error("system message publish failed", "error", err, "roomID", req.RoomID) } - // Cross-site outbox grouped by destination site + // Publish InboxMemberEvent for downstream search-sync indexing. Split + // by home site — local accounts go to local INBOX, remote accounts go + // to per-site OUTBOX. Like processRemoveIndividual, remove events omit + // RoomName/RoomType (not needed for spotlight deletes or user-room + // script-removes). OrgID is NOT carried on InboxMemberEvent — it's an + // inbox-worker concern for org-level cleanup, not a search-index one. + var localAccounts []string siteAccounts := make(map[string][]string) for _, m := range toRemove { - if m.SiteID != h.siteID { + if m.SiteID == h.siteID { + localAccounts = append(localAccounts, m.Account) + } else { siteAccounts[m.SiteID] = append(siteAccounts[m.SiteID], m.Account) } } + + if len(localAccounts) > 0 { + inboxEvt := model.InboxMemberEvent{ + RoomID: req.RoomID, + SiteID: h.siteID, + Accounts: localAccounts, + Timestamp: now.UnixMilli(), + } + outbox := model.OutboxEvent{ + Type: model.OutboxMemberRemoved, + SiteID: h.siteID, + DestSiteID: h.siteID, + Payload: mustMarshal(inboxEvt), + Timestamp: now.UnixMilli(), + } + if err := h.publish(ctx, subject.InboxMemberRemoved(h.siteID), mustMarshal(outbox)); err != nil { + slog.Error("inbox member_removed publish failed", "error", err, "roomID", req.RoomID) + } + } + for destSiteID, accounts := range siteAccounts { - evt := model.MemberRemoveEvent{ - Type: "member_removed", + evt := model.InboxMemberEvent{ RoomID: req.RoomID, - Accounts: accounts, SiteID: h.siteID, - OrgID: req.OrgID, + Accounts: accounts, Timestamp: now.UnixMilli(), } outbox := model.OutboxEvent{ - Type: "member_removed", + Type: model.OutboxMemberRemoved, SiteID: h.siteID, DestSiteID: destSiteID, Payload: mustMarshal(evt), @@ -665,32 +711,64 @@ func (h *Handler) processAddMembers(ctx context.Context, data []byte) error { slog.Error("system message publish failed", "error", err, "roomID", req.RoomID) } - // 10. Outbox for cross-site members — batched by destination site + // 10. Publish InboxMemberEvent to local INBOX (same-site accounts) and + // OUTBOX (cross-site accounts). Split subscriptions by home site and + // fan out: one INBOX publish for all local accounts, one OUTBOX + // publish per remote destination site. InboxMemberEvent is a + // superset of MemberAddEvent (adds RoomName + RoomType), so + // inbox-worker on the destination site continues to unmarshal it + // into MemberAddEvent and ignores the extra fields. + var localAccounts []string remoteSiteMembers := make(map[string][]string) for _, sub := range subs { user, ok := userMap[sub.User.Account] - if !ok || user.SiteID == room.SiteID { + if !ok { continue } - remoteSiteMembers[user.SiteID] = append(remoteSiteMembers[user.SiteID], sub.User.Account) + if user.SiteID == room.SiteID { + localAccounts = append(localAccounts, sub.User.Account) + } else { + remoteSiteMembers[user.SiteID] = append(remoteSiteMembers[user.SiteID], sub.User.Account) + } } + + inboxEvt := model.InboxMemberEvent{ + RoomID: req.RoomID, + RoomName: room.Name, + RoomType: room.Type, + SiteID: room.SiteID, + Accounts: nil, // filled per publish below + HistorySharedSince: historySharedSince, + JoinedAt: req.Timestamp, + Timestamp: now.UnixMilli(), + } + + if len(localAccounts) > 0 { + local := inboxEvt + local.Accounts = localAccounts + outbox := model.OutboxEvent{ + Type: model.OutboxMemberAdded, + SiteID: room.SiteID, + DestSiteID: room.SiteID, + Payload: mustMarshal(local), + Timestamp: now.UnixMilli(), + } + if err := h.publish(ctx, subject.InboxMemberAdded(room.SiteID), mustMarshal(outbox)); err != nil { + slog.Error("inbox member_added publish failed", "error", err, "roomID", req.RoomID) + } + } + for destSiteID, accounts := range remoteSiteMembers { - siteEvt := model.MemberAddEvent{ - Type: "member_added", - RoomID: req.RoomID, - Accounts: accounts, - SiteID: room.SiteID, - JoinedAt: req.Timestamp, - HistorySharedSince: historySharedSince, - Timestamp: now.UnixMilli(), - } - siteEvtData, _ := json.Marshal(siteEvt) + remote := inboxEvt + remote.Accounts = accounts outbox := model.OutboxEvent{ - Type: "member_added", SiteID: room.SiteID, DestSiteID: destSiteID, - Payload: siteEvtData, Timestamp: now.UnixMilli(), + Type: model.OutboxMemberAdded, + SiteID: room.SiteID, + DestSiteID: destSiteID, + Payload: mustMarshal(remote), + Timestamp: now.UnixMilli(), } - outboxData, _ := json.Marshal(outbox) - if err := h.publish(ctx, subject.Outbox(room.SiteID, destSiteID, "member_added"), outboxData); err != nil { + if err := h.publish(ctx, subject.Outbox(room.SiteID, destSiteID, "member_added"), mustMarshal(outbox)); err != nil { return fmt.Errorf("outbox publish to %s failed: %w", destSiteID, err) } } diff --git a/room-worker/handler_test.go b/room-worker/handler_test.go index 0f72558d1..4e92eda72 100644 --- a/room-worker/handler_test.go +++ b/room-worker/handler_test.go @@ -411,8 +411,9 @@ func TestHandler_ProcessRemoveMember_SelfLeave_IndividualOnly(t *testing.T) { err := h.processRemoveMember(context.Background(), data) require.NoError(t, err) - // Expect: subscription update + member change event + system message = 3 publishes - assert.Len(t, published, 3, "expected 3 publishes: sub update, member event, sys msg") + // Expect: subscription update + member change event + system message + + // local INBOX publish (for search-sync) = 4 publishes + assert.Len(t, published, 4, "expected 4 publishes: sub update, member event, sys msg, inbox") subjSet := make(map[string]bool) for _, p := range published { @@ -421,6 +422,24 @@ func TestHandler_ProcessRemoveMember_SelfLeave_IndividualOnly(t *testing.T) { assert.True(t, subjSet[subject.SubscriptionUpdate(account)], "expected subscription update published") assert.True(t, subjSet[subject.MemberEvent(roomID)], "expected member event published") + assert.True(t, subjSet[subject.InboxMemberRemoved(siteID)], "expected local inbox member_removed published") + + // Verify the inbox publish wraps an InboxMemberEvent with the expected + // remove-path shape (no RoomName/RoomType, one account). + for _, p := range published { + if p.subj != subject.InboxMemberRemoved(siteID) { + continue + } + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + assert.Equal(t, model.OutboxMemberRemoved, outbox.Type) + assert.Equal(t, siteID, outbox.SiteID) + assert.Equal(t, siteID, outbox.DestSiteID, "local INBOX publish has DestSiteID == SiteID") + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, roomID, inbox.RoomID) + assert.Equal(t, []string{account}, inbox.Accounts) + } // Verify timestamps on all events for _, p := range published { @@ -585,7 +604,7 @@ func TestHandler_ProcessRemoveMember_OwnerRemovesIndividual(t *testing.T) { err := h.processRemoveMember(context.Background(), data) require.NoError(t, err) - assert.Len(t, published, 3, "expected 3 publishes: sub update, member event, sys msg") + assert.Len(t, published, 4, "expected 4 publishes: sub update, member event, sys msg, inbox") // Verify the sys msg has type "member_removed" for _, p := range published { @@ -596,6 +615,23 @@ func TestHandler_ProcessRemoveMember_OwnerRemovesIndividual(t *testing.T) { assert.Contains(t, evt.Accounts, account) } } + + // Verify local inbox publish carries the expected InboxMemberEvent. + var sawInbox bool + for _, p := range published { + if p.subj != subject.InboxMemberRemoved(siteID) { + continue + } + sawInbox = true + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + assert.Equal(t, model.OutboxMemberRemoved, outbox.Type) + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, []string{account}, inbox.Accounts) + assert.Equal(t, roomID, inbox.RoomID) + } + assert.True(t, sawInbox, "expected local inbox member_removed publish") } // --- processAddMembers tests --- @@ -658,6 +694,78 @@ func TestHandler_ProcessAddMembers(t *testing.T) { assert.Equal(t, 1, outboxCount, "should publish exactly 1 batched outbox event per destination site") } +// TestHandler_ProcessAddMembers_PublishesToInbox verifies the INBOX publishes +// added for search-sync-worker: +// - Local accounts fan out to `chat.inbox.{site}.member_added` with an +// InboxMemberEvent carrying RoomName + RoomType. +// - Cross-site OUTBOX payload was migrated from MemberAddEvent to +// InboxMemberEvent so the remote site's search-sync sees room metadata. +func TestHandler_ProcessAddMembers_PublishesToInbox(t *testing.T) { + ctrl := gomock.NewController(t) + store := NewMockSubscriptionStore(ctrl) + + var published []publishedMsg + publish := func(_ context.Context, subj string, data []byte) error { + published = append(published, publishedMsg{subj: subj, data: data}) + return nil + } + h := NewHandler(store, "site-a", publish) + + // Room with a name + type — these must propagate to InboxMemberEvent. + store.EXPECT().GetRoom(gomock.Any(), "r1").Return(&model.Room{ + ID: "r1", Name: "engineering", Type: model.RoomTypeGroup, SiteID: "site-a", + }, nil) + // Two local users (site-a) + one remote (site-b). Local accounts batch + // into one INBOX publish; remote user goes to OUTBOX for site-b. + store.EXPECT().FindUsersByAccounts(gomock.Any(), []string{"bob", "carol", "dave"}).Return([]model.User{ + {ID: "u2", Account: "bob", SiteID: "site-a"}, + {ID: "u3", Account: "carol", SiteID: "site-a"}, + {ID: "u4", Account: "dave", SiteID: "site-b"}, + }, nil) + store.EXPECT().BulkCreateSubscriptions(gomock.Any(), gomock.Any()).Return(nil) + store.EXPECT().IncrementUserCount(gomock.Any(), "r1", 3).Return(nil) + store.EXPECT().HasOrgRoomMembers(gomock.Any(), "r1").Return(false, nil) + + req := model.AddMembersRequest{ + RoomID: "r1", Users: []string{"bob", "carol", "dave"}, + History: model.HistoryConfig{Mode: model.HistoryModeAll}, + Timestamp: 1735689600000, + } + reqData, _ := json.Marshal(req) + + err := h.processAddMembers(context.Background(), reqData) + require.NoError(t, err) + + var sawLocalInbox, sawRemoteOutbox bool + for _, p := range published { + switch p.subj { + case subject.InboxMemberAdded("site-a"): + sawLocalInbox = true + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + assert.Equal(t, model.OutboxMemberAdded, outbox.Type) + assert.Equal(t, "site-a", outbox.SiteID) + assert.Equal(t, "site-a", outbox.DestSiteID, "local INBOX: DestSiteID == SiteID") + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, "r1", inbox.RoomID) + assert.Equal(t, "engineering", inbox.RoomName) + assert.Equal(t, model.RoomTypeGroup, inbox.RoomType) + assert.ElementsMatch(t, []string{"bob", "carol"}, inbox.Accounts) + case subject.Outbox("site-a", "site-b", "member_added"): + sawRemoteOutbox = true + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, "engineering", inbox.RoomName, "cross-site payload must carry RoomName") + assert.Equal(t, []string{"dave"}, inbox.Accounts) + } + } + assert.True(t, sawLocalInbox, "expected local INBOX publish for same-site accounts") + assert.True(t, sawRemoteOutbox, "expected cross-site OUTBOX publish for remote account") +} + func TestHandler_ProcessAddMembers_HistoryAll(t *testing.T) { ctrl := gomock.NewController(t) store := NewMockSubscriptionStore(ctrl) @@ -849,8 +957,9 @@ func TestHandler_ProcessRemoveMember_OwnerRemovesOrg(t *testing.T) { err := h.processRemoveMember(context.Background(), data) require.NoError(t, err) - // Expect: 2 sub updates (carol, dave) + 1 member event + 1 sys msg = 4 publishes - assert.Len(t, published, 4, "expected 4 publishes: 2 sub updates, member event, sys msg") + // Expect: 2 sub updates (carol, dave) + 1 member event + 1 sys msg + + // 1 local INBOX publish (for search-sync) = 5 publishes + assert.Len(t, published, 5, "expected 5 publishes: 2 sub updates, member event, sys msg, inbox") subjSet := make(map[string]bool) for _, p := range published { @@ -861,6 +970,20 @@ func TestHandler_ProcessRemoveMember_OwnerRemovesOrg(t *testing.T) { assert.True(t, subjSet[subject.SubscriptionUpdate("dave")], "expected subscription update for dave") assert.False(t, subjSet[subject.SubscriptionUpdate("eve")], "eve has individual membership, should not be removed") assert.True(t, subjSet[subject.MemberEvent(roomID)], "expected member event published") + assert.True(t, subjSet[subject.InboxMemberRemoved(siteID)], "expected local inbox member_removed") + + // Local INBOX publish batches all removed accounts for same-site users. + for _, p := range published { + if p.subj != subject.InboxMemberRemoved(siteID) { + continue + } + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, roomID, inbox.RoomID) + assert.ElementsMatch(t, []string{"carol", "dave"}, inbox.Accounts) + } } func TestHandler_ProcessRemoveMember_CrossSiteOutbox(t *testing.T) { @@ -917,6 +1040,25 @@ func TestHandler_ProcessRemoveMember_CrossSiteOutbox(t *testing.T) { subjSet[p.subj] = true } assert.True(t, subjSet[outboxSubj], "expected outbox event published for remote user") + // No local INBOX publish — the user is on a different site, so search- + // sync indexing happens on the remote side via federation. + assert.False(t, subjSet[subject.InboxMemberRemoved(localSite)], + "local INBOX publish must NOT fire when the removed user is remote") + + // Verify the cross-site OUTBOX payload is an InboxMemberEvent (so the + // remote site's search-sync sees RoomID + Accounts for spotlight/user-room + // updates via the Sources+SubjectTransform federation). + for _, p := range published { + if p.subj != outboxSubj { + continue + } + var outbox model.OutboxEvent + require.NoError(t, json.Unmarshal(p.data, &outbox)) + var inbox model.InboxMemberEvent + require.NoError(t, json.Unmarshal(outbox.Payload, &inbox)) + assert.Equal(t, roomID, inbox.RoomID) + assert.Equal(t, []string{account}, inbox.Accounts) + } } func TestHandler_ProcessRemoveMember_UnmarshalError(t *testing.T) {