diff --git a/internal/rpc/channels_leave_rpc_test.go b/internal/rpc/channels_leave_rpc_test.go index 22c09360..31e53c8a 100644 --- a/internal/rpc/channels_leave_rpc_test.go +++ b/internal/rpc/channels_leave_rpc_test.go @@ -109,60 +109,6 @@ func TestMessagesGetFutureChatCreatorAfterLeaveAndCreatorLeaveTransfers(t *testi } } -func TestLeaveChannelInvalidatesStaleFullChannelProjection(t *testing.T) { - ctx := context.Background() - userStore := memory.NewUserStore() - owner, _ := userStore.Create(ctx, domain.User{AccessHash: 9401, Phone: "15550009401", FirstName: "Owner"}) - member, _ := userStore.Create(ctx, domain.User{AccessHash: 9402, Phone: "15550009402", FirstName: "Member"}) - channelStore := memory.NewChannelStore() - channelService := appchannels.NewService(channelStore) - r := New(Config{}, Deps{ - Users: appusers.NewService(userStore), - Channels: channelService, - }, zaptest.NewLogger(t), fixedClock{now: time.Unix(1700009400, 0)}) - created, err := channelService.CreateChannel(ctx, owner.ID, domain.CreateChannelRequest{ - CreatorUserID: owner.ID, - Title: "leave projection", - Megagroup: true, - Date: 1700009400, - }) - if err != nil { - t.Fatalf("create channel: %v", err) - } - if _, err := channelService.UpdateUsername(ctx, owner.ID, domain.UpdateChannelUsernameRequest{ - ChannelID: created.Channel.ID, - Username: "leave_projection_pub", - }); err != nil { - t.Fatalf("publish channel: %v", err) - } - inputChannel := &tg.InputChannel{ChannelID: created.Channel.ID, AccessHash: created.Channel.AccessHash} - - if _, err := r.onChannelsJoinChannel(WithUserID(ctx, member.ID), inputChannel); err != nil { - t.Fatalf("member joins: %v", err) - } - // Warm the channels.getFullChannel projection cache while still a member. - full, err := r.onChannelsGetFullChannel(WithUserID(ctx, member.ID), inputChannel) - if err != nil { - t.Fatalf("full channel while joined: %v", err) - } - if chat, ok := full.Chats[0].(*tg.Channel); !ok || chat.Left { - t.Fatalf("joined full chat = %#v, want member (not left)", full.Chats[0]) - } - - if _, err := r.onChannelsLeaveChannel(WithUserID(ctx, member.ID), inputChannel); err != nil { - t.Fatalf("member leaves: %v", err) - } - - after, err := r.onChannelsGetFullChannel(WithUserID(ctx, member.ID), inputChannel) - if err != nil { - t.Fatalf("full channel after leave: %v", err) - } - chat, ok := after.Chats[0].(*tg.Channel) - if !ok || !chat.Left { - t.Fatalf("post-leave full chat = %#v, want left=true (stale projection served)", after.Chats[0]) - } -} - func TestMessagesEditChatCreatorTransfersWithoutChannelPts(t *testing.T) { ctx := context.Background() userStore := memory.NewUserStore() diff --git a/internal/rpc/channels_members.go b/internal/rpc/channels_members.go index 5a3fc768..0ad8b911 100644 --- a/internal/rpc/channels_members.go +++ b/internal/rpc/channels_members.go @@ -341,7 +341,6 @@ func (r *Router) onChannelsInviteToChannel(ctx context.Context, req *tg.Channels return nil, channelInviteErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) - r.invalidateChannelMembershipProjection(res.Channel.ID, channelMemberUserIDs(res.Members)) r.addOnlineChannelMemberships(res.Channel.ID, channelMemberUserIDs(res.Members)...) cache := newViewerPeerCache(r) updates := r.channelOperationUpdatesWithPeerCache(ctx, userID, res, cache) @@ -380,7 +379,6 @@ func (r *Router) onChannelsJoinChannel(ctx context.Context, input tg.InputChanne return nil, channelInviteErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) - r.invalidateChannelMembershipProjection(res.Channel.ID, channelMemberUserIDs(res.Members)) r.addOnlineChannelMemberships(res.Channel.ID, channelMemberUserIDs(res.Members)...) updates := r.channelOperationUpdates(ctx, userID, res) r.pushChannelUpdates(ctx, userID, res.Channel.ID, res.Recipients, func(viewerUserID int64) *tg.Updates { @@ -408,11 +406,6 @@ func (r *Router) onChannelsLeaveChannel(ctx context.Context, input tg.InputChann return nil, channelAdminErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) - membershipChanged := channelMemberUserIDs(res.Members) - if len(membershipChanged) == 0 { - membershipChanged = []int64{userID} - } - r.invalidateChannelMembershipProjection(res.Channel.ID, membershipChanged) r.removeOnlineChannelMemberships(res.Channel.ID, userID) r.recordChannelStateForUser(ctx, userID, res.Channel.ID, true) updates := r.channelOperationUpdates(ctx, userID, res) @@ -665,7 +658,6 @@ func (r *Router) onMessagesHideChatJoinRequest(ctx context.Context, req *tg.Mess return nil, channelInviteErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) - r.invalidateChannelMembershipProjection(res.Channel.ID, channelMemberUserIDs(res.Members)) r.addOnlineChannelMemberships(res.Channel.ID, channelMemberUserIDs(res.Members)...) updates := r.channelOperationUpdates(ctx, userID, res) r.appendPendingJoinRequestsUpdate(ctx, userID, updates, res.Channel) @@ -704,7 +696,6 @@ func (r *Router) onMessagesHideAllChatJoinRequests(ctx context.Context, req *tg. return nil, channelInviteErr(err) } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) - r.invalidateChannelMembershipProjection(res.Channel.ID, channelMemberUserIDs(res.Members)) r.addOnlineChannelMemberships(res.Channel.ID, channelMemberUserIDs(res.Members)...) updates := r.channelOperationUpdates(ctx, userID, res) r.appendPendingJoinRequestsUpdate(ctx, userID, updates, res.Channel) diff --git a/internal/rpc/rpc_projection_cache.go b/internal/rpc/rpc_projection_cache.go index f96abf4f..f81a6ad5 100644 --- a/internal/rpc/rpc_projection_cache.go +++ b/internal/rpc/rpc_projection_cache.go @@ -290,24 +290,6 @@ func (r *Router) invalidateRPCProjectionForPeer(ownerUserID int64, peer domain.P } } -// invalidateChannelMembershipProjection drops the cached channels.getFullChannel -// projection for each user whose membership in channelID just changed (join, -// leave, invite, request approval). Without it a client that polls -// channels.getFullChannel right after channels.leaveChannel keeps getting a -// projection that still shows it as an active member (left=false) until the -// entry's TTL lapses, so it keeps an open compose box even though sends are -// already rejected with CHANNEL_PRIVATE. -func (r *Router) invalidateChannelMembershipProjection(channelID int64, userIDs []int64) { - if r.channelFullProjectionCache == nil || channelID == 0 { - return - } - for _, userID := range userIDs { - if userID != 0 { - r.channelFullProjectionCache.DeletePair(userID, channelID) - } - } -} - func (r *Router) invalidateRPCProjectionForChannel(channelID int64) { if r.channelFullProjectionCache != nil { r.channelFullProjectionCache.DeleteChannel(channelID) diff --git a/internal/store/postgres/channel_invite_import.go b/internal/store/postgres/channel_invite_import.go index 50ce7401..8cf5ceb9 100644 --- a/internal/store/postgres/channel_invite_import.go +++ b/internal/store/postgres/channel_invite_import.go @@ -103,7 +103,6 @@ func (s *ChannelStore) ImportInvite(ctx context.Context, req domain.ImportChanne return domain.CreateChannelResult{}, fmt.Errorf("commit import channel invite: %w", err) } committed = true - s.invalidateChannelMembershipCaches(result.Channel.ID, req.UserID) recipients, _ := s.ListActiveChannelMemberIDs(ctx, req.UserID, result.Channel.ID, 0) result.Recipients = recipients return result, nil diff --git a/internal/store/postgres/channel_invite_members.go b/internal/store/postgres/channel_invite_members.go index 7a2e582c..ee546aaf 100644 --- a/internal/store/postgres/channel_invite_members.go +++ b/internal/store/postgres/channel_invite_members.go @@ -125,7 +125,6 @@ func (s *ChannelStore) InviteToChannel(ctx context.Context, channelID, inviterUs return domain.CreateChannelResult{}, fmt.Errorf("commit invite channel: %w", err) } committed = true - s.invalidateChannelMembershipCaches(channelID, invitedIDs...) recipients, _ := s.ListActiveChannelMemberIDs(ctx, inviterUserID, channelID, 0) return domain.CreateChannelResult{Channel: channel, Members: members, Message: msg, Event: event, Recipients: recipients}, nil } diff --git a/internal/store/postgres/channel_member_join.go b/internal/store/postgres/channel_member_join.go index aa188c11..5a77bd45 100644 --- a/internal/store/postgres/channel_member_join.go +++ b/internal/store/postgres/channel_member_join.go @@ -130,7 +130,6 @@ WHERE channel_id = $1 AND user_id = $2`, channelID, userID, member.ReadInboxMaxI return domain.CreateChannelResult{}, fmt.Errorf("commit join channel: %w", err) } committed = true - s.invalidateChannelMembershipCaches(channelID, userID) recipients, _ := s.ListActiveChannelMemberIDs(ctx, userID, channelID, 0) return domain.CreateChannelResult{Channel: channel, Members: []domain.ChannelMember{member}, Message: msg, Event: event, Recipients: recipients}, nil } @@ -260,11 +259,6 @@ WHERE id = $1`, channelID, channel.CreatorUserID, adminsDelta); err != nil { return domain.CreateChannelResult{}, fmt.Errorf("commit leave channel: %w", err) } committed = true - leftUserIDs := make([]int64, 0, len(members)) - for _, m := range members { - leftUserIDs = append(leftUserIDs, m.UserID) - } - s.invalidateChannelMembershipCaches(channelID, leftUserIDs...) recipients = append(recipients, userID) return domain.CreateChannelResult{Channel: channel, Members: members, Message: msg, Event: event, Recipients: recipients}, nil } diff --git a/internal/store/postgres/channel_store.go b/internal/store/postgres/channel_store.go index 54de6c3d..b7c3e329 100644 --- a/internal/store/postgres/channel_store.go +++ b/internal/store/postgres/channel_store.go @@ -95,33 +95,6 @@ func (s *ChannelStore) boostCacheActive(db sqlcgen.DBTX) bool { return s.boostCache != nil && db == s.db } -// invalidateChannelMembershipCaches drops the in-process reads that a membership -// change (join, leave, invite, kick) makes stale for the given users. The -// ReadModelChangeListener also clears these off the async NOTIFY, but callers -// must not depend on that round-trip: a client that polls channels.getFullChannel -// right after channels.leaveChannel would otherwise keep seeing itself as an -// active member (and keep an open compose box) until the notify lands. Call it -// post-commit. -func (s *ChannelStore) invalidateChannelMembershipCaches(channelID int64, userIDs ...int64) { - if channelID == 0 { - return - } - if s.rowCache != nil { - s.rowCache.delete(channelID) - } - for _, userID := range userIDs { - if userID == 0 { - continue - } - if s.memberCache != nil { - s.memberCache.delete(channelID, userID) - } - if s.dialogCache != nil { - s.dialogCache.delete(userID, channelID) - } - } -} - // NewChannelStore 基于 pgx 连接池（或事务）创建 ChannelStore。 func NewChannelStore(db sqlcgen.DBTX, opts ...ChannelStoreOption) *ChannelStore { s := &ChannelStore{db: db}