channels: drop stale membership caches on join/leave
After channels.leaveChannel, a client that polls channels.getFullChannel kept receiving a projection that still showed it as an active member (left=false) until the per-(viewer,channel) RPC projection cache and the store-level member cache lapsed on their own or the async read-model NOTIFY landed. The client therefore kept an open compose box while every send was already rejected with CHANNEL_PRIVATE - most visible on public forum supergroups, where getFullChannel keeps succeeding via the preview path instead of tearing the chat down. Every other membership-mutating path already busts these caches synchronously; join/leave/invite/request-approval did not. Add: - store: invalidateChannelMembershipCaches (row + member + dialog caches), called post-commit from JoinChannel, LeaveChannel, ImportInvite, InviteToChannel. - rpc: invalidateChannelMembershipProjection (channelFullProjectionCache pair), called from the join/leave/invite/hide-requests handlers for every user whose membership changed.
This commit is contained in:
parent
2c782aab95
commit
55a6e0bb35
7 changed files with 116 additions and 0 deletions
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -379,6 +380,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 {
|
||||
|
|
@ -406,6 +408,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)
|
||||
|
|
@ -658,6 +665,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)
|
||||
|
|
@ -696,6 +704,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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -125,6 +125,7 @@ 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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -130,6 +130,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
|
||||
}
|
||||
|
|
@ -259,6 +260,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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -95,6 +95,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}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue