diff --git a/internal/rpc/channels_leave_rpc_test.go b/internal/rpc/channels_leave_rpc_test.go index 31e53c8a..22c09360 100644 --- a/internal/rpc/channels_leave_rpc_test.go +++ b/internal/rpc/channels_leave_rpc_test.go @@ -109,6 +109,60 @@ 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 3b59ebcf..087352ec 100644 --- a/internal/rpc/channels_members.go +++ b/internal/rpc/channels_members.go @@ -341,6 +341,7 @@ 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) @@ -382,6 +383,7 @@ 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 { @@ -409,6 +411,11 @@ 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) @@ -661,6 +668,7 @@ 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) @@ -699,6 +707,7 @@ 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 f81a6ad5..f96abf4f 100644 --- a/internal/rpc/rpc_projection_cache.go +++ b/internal/rpc/rpc_projection_cache.go @@ -290,6 +290,24 @@ 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 acdd0012..59ff741e 100644 --- a/internal/store/postgres/channel_invite_import.go +++ b/internal/store/postgres/channel_invite_import.go @@ -103,6 +103,7 @@ 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 014ac0c6..f12bb946 100644 --- a/internal/store/postgres/channel_invite_members.go +++ b/internal/store/postgres/channel_invite_members.go @@ -129,7 +129,9 @@ func (s *ChannelStore) InviteToChannel(ctx context.Context, channelID, inviterUs return domain.CreateChannelResult{}, fmt.Errorf("commit invite channel: %w", err) } committed = true - return domain.CreateChannelResult{Channel: channel, Members: members, Message: msg, Event: event}, nil + 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 } func canInviteToChannel(channel domain.Channel, member domain.ChannelMember) bool { diff --git a/internal/store/postgres/channel_member_join.go b/internal/store/postgres/channel_member_join.go index 67f3ac2f..54f5fc56 100644 --- a/internal/store/postgres/channel_member_join.go +++ b/internal/store/postgres/channel_member_join.go @@ -133,6 +133,7 @@ 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 } @@ -265,6 +266,11 @@ 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 1e88e86d..fa401bf3 100644 --- a/internal/store/postgres/channel_store.go +++ b/internal/store/postgres/channel_store.go @@ -118,6 +118,33 @@ 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}