diff --git a/internal/app/channels/service.go b/internal/app/channels/service.go index fc42eb56..27822a4a 100644 --- a/internal/app/channels/service.go +++ b/internal/app/channels/service.go @@ -2295,6 +2295,21 @@ func (s *Service) FilterActiveMemberIDs(ctx context.Context, channelID int64, us return s.channels.FilterActiveChannelMemberIDs(ctx, channelID, candidates) } +// FilterMessageAudienceIDs keeps active members and currently authorized +// public-preview viewers from a bounded online candidate set. The store performs +// one batched authoritative check per bounded chunk so runtime session indexes +// never become an access-control source of truth. +func (s *Service) FilterMessageAudienceIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) { + if s == nil || s.channels == nil || channelID == 0 { + return nil, domain.ErrChannelInvalid + } + candidates := uniqueNonZero(userIDs) + if len(candidates) == 0 { + return nil, nil + } + return s.channels.FilterChannelMessageAudienceIDs(ctx, channelID, candidates) +} + // GetDifference returns channel-scoped pts difference. func (s *Service) GetDifference(ctx context.Context, userID int64, req domain.ChannelDifferenceRequest) (domain.ChannelDifference, error) { if s == nil || s.channels == nil || userID == 0 || req.ChannelID == 0 { diff --git a/internal/app/channels/service_test.go b/internal/app/channels/service_test.go index daf5fa6f..427462ca 100644 --- a/internal/app/channels/service_test.go +++ b/internal/app/channels/service_test.go @@ -3014,12 +3014,37 @@ func TestPublicChannelPreviewAllowsNonMemberHistory(t *testing.T) { if err != nil { t.Fatalf("non-member GetDifference public preview: %v", err) } - if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.Events) != 0 || len(diff.NewMessages) != 0 || len(diff.OtherUpdates) != 0 { - t.Fatalf("preview diff = %+v, want empty public preview difference at current pts", diff) + if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.Events) != 1 || len(diff.NewMessages) != 1 || len(diff.OtherUpdates) != 0 { + t.Fatalf("preview diff = %+v, want one public preview message at current pts", diff) + } + if diff.NewMessages[0].ID != sent.Message.ID || diff.NewMessages[0].Body != sent.Message.Body { + t.Fatalf("preview diff message = %+v, want sent public post %+v", diff.NewMessages[0], sent.Message) } if diff.Dialog.UnreadCount != 0 || diff.Dialog.ReadInboxMaxID < sent.Message.ID { t.Fatalf("preview diff dialog = %+v, want read-only public preview dialog", diff.Dialog) } + audience, err := service.FilterMessageAudienceIDs(ctx, public.ID, []int64{viewerID, ownerID, viewerID}) + if err != nil || len(audience) != 2 { + t.Fatalf("public message audience = %v err %v, want owner and preview viewer", audience, err) + } + if _, err := service.JoinChannel(ctx, viewerID, public.ID, 21); err != nil { + t.Fatalf("JoinChannel public preview viewer: %v", err) + } + if _, err := service.LeaveChannel(ctx, viewerID, public.ID, 22); err != nil { + t.Fatalf("LeaveChannel public preview viewer: %v", err) + } + filtered, err := service.GetDifference(ctx, viewerID, domain.ChannelDifferenceRequest{ + ChannelID: public.ID, + Pts: sent.Event.Pts, + Limit: 10, + }) + if err != nil { + t.Fatalf("preview difference across participant events: %v", err) + } + if !filtered.Final || filtered.Pts != sent.Event.Pts || len(filtered.Events) != 0 || + len(filtered.NewMessages) != 0 || len(filtered.OtherUpdates) != 0 { + t.Fatalf("difference after transient participant changes = %+v, want unchanged PTS", filtered) + } private, err := service.CreateChannel(ctx, ownerID, domain.CreateChannelRequest{ Title: "Private Preview", @@ -3047,6 +3072,9 @@ func TestPublicChannelPreviewAllowsNonMemberHistory(t *testing.T) { if _, err := service.GetDifference(ctx, viewerID, domain.ChannelDifferenceRequest{ChannelID: public.ID, Pts: created.Event.Pts, Limit: 10}); !errors.Is(err, domain.ErrChannelUserBanned) { t.Fatalf("banned public preview GetDifference err = %v, want ErrChannelUserBanned", err) } + if audience, err := service.FilterMessageAudienceIDs(ctx, public.ID, []int64{viewerID}); err != nil || len(audience) != 0 { + t.Fatalf("banned public message audience = %v err %v, want empty", audience, err) + } } func TestChannelDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { diff --git a/internal/app/dialogs/service.go b/internal/app/dialogs/service.go index aae8c85a..b28b8c2d 100644 --- a/internal/app/dialogs/service.go +++ b/internal/app/dialogs/service.go @@ -342,11 +342,24 @@ func (s *Service) appendMissingChannelPeerPreviews(ctx context.Context, userID i if !ok || view.Forbidden { continue } + if view.Self.Status == domain.ChannelMemberLeft && !view.Self.Guest { + // A visible public preview still needs one transient dialog so a + // client can finish bootstrapping the requested peer. Keep the top + // message and read state at zero: the response is an admission + // token, not a persisted/chat-list dialog snapshot. In particular, + // clients that persist non-zero top dialogs will instead continue + // with messages.getHistory, which is the authoritative preview + // history path. + out.Dialogs = append(out.Dialogs, publicChannelPreviewBootstrapDialog(view)) + out.Channels = append(out.Channels, view.Channel) + out.Count++ + present[channelID] = struct{}{} + continue + } // Linked discussion guests need a transient peer-dialog snapshot so // TDesktop can finish materializing the comments History after // requestSelf. ChannelLeft keeps the snapshot out of the main chat list, - // and Guest guarantees this path never turns an ordinary public preview - // into a dialog. + // while Guest authorizes the target's real top-message snapshot. if view.Self.Status != domain.ChannelMemberActive && !view.Self.Guest { continue } @@ -378,6 +391,14 @@ func (s *Service) appendMissingChannelPeerPreviews(ctx context.Context, userID i return out, nil } +func publicChannelPreviewBootstrapDialog(view domain.ChannelView) domain.Dialog { + return domain.Dialog{ + Peer: domain.Peer{Type: domain.PeerTypeChannel, ID: view.Channel.ID}, + ChannelLeft: true, + Pts: view.Channel.Pts, + } +} + func isChannelPreviewAccessError(err error) bool { return errors.Is(err, domain.ErrChannelPrivate) || errors.Is(err, domain.ErrChannelUserBanned) || diff --git a/internal/app/dialogs/service_test.go b/internal/app/dialogs/service_test.go index dffb055f..dee10f7c 100644 --- a/internal/app/dialogs/service_test.go +++ b/internal/app/dialogs/service_test.go @@ -60,6 +60,7 @@ type countingDialogChannelStore struct { getChannelCalls int getChannelsCalls int getChannelDialogsCalls int + listHistoryCalls int } func (s *countingDialogChannelStore) GetChannel(ctx context.Context, viewerUserID, channelID int64) (domain.ChannelView, error) { @@ -77,6 +78,11 @@ func (s *countingDialogChannelStore) GetChannelDialogs(ctx context.Context, view return s.ChannelStore.GetChannelDialogs(ctx, viewerUserID, channelIDs) } +func (s *countingDialogChannelStore) ListChannelHistory(ctx context.Context, viewerUserID int64, filter domain.ChannelHistoryFilter) (domain.ChannelHistory, error) { + s.listHistoryCalls++ + return s.ChannelStore.ListChannelHistory(ctx, viewerUserID, filter) +} + func TestGetDialogsHashUsesWarmStableHashCacheAndInvalidatesOnWrite(t *testing.T) { ctx := context.Background() const ownerID int64 = 1001 @@ -695,7 +701,7 @@ func TestGetPeerDialogsRejectsHugeVector(t *testing.T) { } } -func TestGetPeerDialogsSkipsPublicChannelPreviewForNonMember(t *testing.T) { +func TestGetPeerDialogsReturnsZeroTopBootstrapForPublicChannelPreview(t *testing.T) { ctx := context.Background() channelStore := memory.NewChannelStore() channels := appchannels.NewService(channelStore) @@ -716,12 +722,13 @@ func TestGetPeerDialogsSkipsPublicChannelPreviewForNonMember(t *testing.T) { }); err != nil { t.Fatalf("UpdateUsername public: %v", err) } - if _, err := channels.SendMessage(ctx, 1001, domain.SendChannelMessageRequest{ + sent, err := channels.SendMessage(ctx, 1001, domain.SendChannelMessageRequest{ ChannelID: public.Channel.ID, RandomID: 99, Message: "public peer dialog top", Date: 1700002010, - }); err != nil { + }) + if err != nil { t.Fatalf("SendMessage public: %v", err) } private, err := channels.CreateChannel(ctx, 1001, domain.CreateChannelRequest{ @@ -740,8 +747,19 @@ func TestGetPeerDialogsSkipsPublicChannelPreviewForNonMember(t *testing.T) { if err != nil { t.Fatalf("GetPeerDialogs public preview: %v", err) } - if len(list.Dialogs) != 0 || len(list.ChannelMessages) != 0 || len(list.Channels) != 0 || list.Count != 0 { - t.Fatalf("peer dialogs = %+v, want no materialized public preview dialog", list) + if len(list.Dialogs) != 1 || len(list.ChannelMessages) != 0 || len(list.Channels) != 1 || list.Count != 1 { + t.Fatalf("peer dialogs = %+v, want one zero-top public preview bootstrap", list) + } + dialog := list.Dialogs[0] + if dialog.Peer.Type != domain.PeerTypeChannel || dialog.Peer.ID != public.Channel.ID || + !dialog.ChannelLeft || dialog.TopMessage != 0 || dialog.TopMessageDate != 0 || + dialog.ReadInboxMaxID != 0 || dialog.ReadOutboxMaxID != 0 || + dialog.UnreadCount != 0 || dialog.UnreadMentions != 0 || + dialog.UnreadReactions != 0 || dialog.Pts != sent.Event.Pts { + t.Fatalf("public preview bootstrap dialog = %+v", dialog) + } + if list.Channels[0].ID != public.Channel.ID { + t.Fatalf("public preview channels = %+v, want channel %d", list.Channels, public.Channel.ID) } } @@ -810,6 +828,7 @@ func TestGetPeerDialogsBatchesMissingChannelVisibilityChecks(t *testing.T) { channelStore.getChannelCalls = 0 channelStore.getChannelsCalls = 0 + channelStore.listHistoryCalls = 0 list, err := dialogs.GetPeerDialogs(ctx, 1002, []domain.Peer{ {Type: domain.PeerTypeChannel, ID: first.Channel.ID}, {Type: domain.PeerTypeChannel, ID: private.Channel.ID}, @@ -822,8 +841,16 @@ func TestGetPeerDialogsBatchesMissingChannelVisibilityChecks(t *testing.T) { if channelStore.getChannelsCalls != 1 || channelStore.getChannelCalls != 0 { t.Fatalf("visibility channel calls: GetChannels=%d GetChannel=%d, want one batch call only", channelStore.getChannelsCalls, channelStore.getChannelCalls) } - if len(list.Dialogs) != 0 || len(list.ChannelMessages) != 0 || len(list.Channels) != 0 || list.Count != 0 { - t.Fatalf("peer dialogs = %+v, want no public preview dialogs", list) + if channelStore.listHistoryCalls != 0 { + t.Fatalf("public preview history calls = %d, want zero", channelStore.listHistoryCalls) + } + if len(list.Dialogs) != 2 || len(list.ChannelMessages) != 0 || len(list.Channels) != 2 || list.Count != 2 { + t.Fatalf("peer dialogs = %+v, want two deduplicated zero-top public previews", list) + } + for _, dialog := range list.Dialogs { + if !dialog.ChannelLeft || dialog.TopMessage != 0 || dialog.ReadInboxMaxID != 0 || dialog.ReadOutboxMaxID != 0 { + t.Fatalf("public preview bootstrap dialog = %+v", dialog) + } } } diff --git a/internal/mtprotoedge/session_manager.go b/internal/mtprotoedge/session_manager.go index 7ff5a463..04b31acc 100644 --- a/internal/mtprotoedge/session_manager.go +++ b/internal/mtprotoedge/session_manager.go @@ -56,6 +56,11 @@ const ( // 登记的 channel 数上限。membership 源于真实成员关系(大账号可能很多),interest 受客户端 // 直接控制;两者都设一个宽松上界防内存放大,超出即截断并记日志。 maxChannelIndexPerSession = 8192 + // Official clients short-poll at most ten opened channels per session. Keep the + // server-side passive subscription index at the same hard bound. + maxChannelSubscriptionsPerSession = 10 + defaultChannelSubscriptionTTL = 75 * time.Second + maxChannelSubscriptionTTL = 2 * time.Minute ) // forceCloseBatchTimeout is one deadline for a whole revoke/replace/eviction batch. Conn.Close @@ -136,6 +141,11 @@ type sessionKey struct { sessionID int64 } +type channelSubscription struct { + userID int64 + expiresAt int64 +} + // SessionLifecycleObserver receives active connection lifecycle events. type SessionLifecycleObserver interface { SessionOffline(rawAuthKeyID [8]byte, sessionID, userID int64, lastForUser bool) @@ -164,18 +174,20 @@ type SessionManager struct { // claims owns the provisional -> active gap. A claimant is intentionally // absent from every push/online index until its required session control frame // is on the wire and PublishActivation validates the same owner. - claims map[sessionKey]*Conn - claimsByAuth map[[8]byte]map[int64]*Conn // raw authKeyID -> sessionID -> provisional claim - byAuthKey map[[8]byte]map[int64]*Conn // raw authKeyID → sessionID → Conn - byBusinessAuthKey map[[8]byte]map[sessionKey]*Conn - byUser map[int64]map[sessionKey]*Conn - byChannel map[int64]map[sessionKey]int64 // channelID → session → userID,用于频道 active-viewer 临时推送 - bySessionChannels map[sessionKey]map[int64]struct{} - byMemberChannel map[int64]map[sessionKey]int64 // channelID → session → userID,用于已上线成员持久 update 推送 - bySessionMembers map[sessionKey]map[int64]struct{} - pending map[sessionKey][]queuedPush // updates-ready 前暂存的主动推送 - flushing map[sessionKey]bool // 置位时暂存正在排空的 session;排空完成前推送继续进 pending 保序 - pendingBudget *outboundTrackedBudget // 未就绪 session 暂存 encoded body 的进程级上限 + claims map[sessionKey]*Conn + claimsByAuth map[[8]byte]map[int64]*Conn // raw authKeyID -> sessionID -> provisional claim + byAuthKey map[[8]byte]map[int64]*Conn // raw authKeyID → sessionID → Conn + byBusinessAuthKey map[[8]byte]map[sessionKey]*Conn + byUser map[int64]map[sessionKey]*Conn + byChannel map[int64]map[sessionKey]int64 // channelID → session → userID,用于频道 active-viewer 临时推送 + bySessionChannels map[sessionKey]map[int64]struct{} + bySubscribedChannel map[int64]map[sessionKey]channelSubscription + bySessionSubscriptions map[sessionKey]map[int64]int64 + byMemberChannel map[int64]map[sessionKey]int64 // channelID → session → userID,用于已上线成员持久 update 推送 + bySessionMembers map[sessionKey]map[int64]struct{} + pending map[sessionKey][]queuedPush // updates-ready 前暂存的主动推送 + flushing map[sessionKey]bool // 置位时暂存正在排空的 session;排空完成前推送继续进 pending 保序 + pendingBudget *outboundTrackedBudget // 未就绪 session 暂存 encoded body 的进程级上限 lifecycle SessionLifecycleObserver log *zap.Logger @@ -187,20 +199,22 @@ func NewSessionManager(log *zap.Logger) *SessionManager { log = zap.NewNop() } return &SessionManager{ - bySession: make(map[sessionKey]*Conn), - claims: make(map[sessionKey]*Conn), - claimsByAuth: make(map[[8]byte]map[int64]*Conn), - byAuthKey: make(map[[8]byte]map[int64]*Conn), - byBusinessAuthKey: make(map[[8]byte]map[sessionKey]*Conn), - byUser: make(map[int64]map[sessionKey]*Conn), - byChannel: make(map[int64]map[sessionKey]int64), - bySessionChannels: make(map[sessionKey]map[int64]struct{}), - byMemberChannel: make(map[int64]map[sessionKey]int64), - bySessionMembers: make(map[sessionKey]map[int64]struct{}), - pending: make(map[sessionKey][]queuedPush), - flushing: make(map[sessionKey]bool), - pendingBudget: newOutboundTrackedBudget(defaultPendingPushMaxBytes), - log: log, + bySession: make(map[sessionKey]*Conn), + claims: make(map[sessionKey]*Conn), + claimsByAuth: make(map[[8]byte]map[int64]*Conn), + byAuthKey: make(map[[8]byte]map[int64]*Conn), + byBusinessAuthKey: make(map[[8]byte]map[sessionKey]*Conn), + byUser: make(map[int64]map[sessionKey]*Conn), + byChannel: make(map[int64]map[sessionKey]int64), + bySessionChannels: make(map[sessionKey]map[int64]struct{}), + bySubscribedChannel: make(map[int64]map[sessionKey]channelSubscription), + bySessionSubscriptions: make(map[sessionKey]map[int64]int64), + byMemberChannel: make(map[int64]map[sessionKey]int64), + bySessionMembers: make(map[sessionKey]map[int64]struct{}), + pending: make(map[sessionKey][]queuedPush), + flushing: make(map[sessionKey]bool), + pendingBudget: newOutboundTrackedBudget(defaultPendingPushMaxBytes), + log: log, } } @@ -730,8 +744,7 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) { if old := c.userID.Swap(userID); old != 0 { removeUserIndex(m.byUser, old, key) if old != userID { - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) // 身份变化即丢弃暂存推送:它们属于前一个账号,flush 给新账号是跨账号泄露。 // 同时取消进行中的排空(runFlush 还另有 owner 校验做批内兜底)。 @@ -743,8 +756,7 @@ func (m *SessionManager) bindUserLocked(c *Conn, key sessionKey, userID int64) { if userID != 0 { addUserIndex(m.byUser, userID, key, c) } else { - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) m.deletePendingLocked(key) delete(m.flushing, key) @@ -808,8 +820,7 @@ func (m *SessionManager) bindAuthKeyLocked(c *Conn, key sessionKey, authKeyID [8 if oldUserID != 0 { removeUserIndex(m.byUser, oldUserID, key) } - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) m.deletePendingLocked(key) delete(m.flushing, key) @@ -1065,8 +1076,7 @@ func (m *SessionManager) UnbindAuthKey(authKeyID [8]byte) int { if old := c.userID.Swap(0); old != 0 { removeUserIndex(m.byUser, old, key) } - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) // 授权解除后暂存推送属于已登出的账号,不能等下一个登录者置位时 flush 出去。 m.deletePendingLocked(key) @@ -1085,8 +1095,7 @@ func (m *SessionManager) UnbindAuthKey(authKeyID [8]byte) int { func (m *SessionManager) setReceivesUpdatesLocked(c *Conn, key sessionKey, receives bool) (int64, bool) { if !receives { c.receivesUpdates.Store(false) - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) // 取消进行中的排空激活:runFlush 在置位前会复查该标志,标志已删则放弃置位, // 避免把刚置 false 的开关翻回 true。 @@ -1099,8 +1108,7 @@ func (m *SessionManager) setReceivesUpdatesLocked(c *Conn, key sessionKey, recei // pending until generated exact admission freezes a real profile; do not // start a flush which would fail layer binding and retire a healthy socket. c.receivesUpdates.Store(false) - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) c.membershipsSynced.Store(false) delete(m.flushing, key) return 0, false @@ -1900,6 +1908,102 @@ func (m *SessionManager) OnlineChannelUserIDs(channelID int64, limit int) []int6 return m.onlineChannelUsers(m.byChannel, channelID, limit) } +// RefreshChannelSubscription refreshes one public-channel passive-update +// subscription without replacing the other short-polled channels of the same +// session. The index is runtime-only and bounded to the official client limit. +func (m *SessionManager) RefreshChannelSubscription(rawAuthKeyID [8]byte, sessionID, userID, channelID int64, ttl time.Duration) { + if userID == 0 || channelID == 0 { + return + } + if ttl <= 0 { + ttl = defaultChannelSubscriptionTTL + } else if ttl > maxChannelSubscriptionTTL { + ttl = maxChannelSubscriptionTTL + } + key := sessionKey{authKeyID: rawAuthKeyID, sessionID: sessionID} + now := time.Now().UnixNano() + expiresAt := now + int64(ttl) + m.mu.Lock() + defer m.mu.Unlock() + c, ok := m.bySession[key] + if !ok || c.userID.Load() != userID { + return + } + m.pruneSessionSubscriptionsLocked(key, now) + channels := m.bySessionSubscriptions[key] + if channels == nil { + channels = make(map[int64]int64, 1) + m.bySessionSubscriptions[key] = channels + } + if _, exists := channels[channelID]; !exists && len(channels) >= maxChannelSubscriptionsPerSession { + m.log.Warn("Channel passive subscription ignored at per-session cap", + zap.String("auth_key_id", sessionKeyLog(rawAuthKeyID)), + zap.Int64("session_id", sessionID), + zap.Int64("channel_id", channelID), + zap.Int("cap", maxChannelSubscriptionsPerSession)) + return + } + channels[channelID] = expiresAt + sessions := m.bySubscribedChannel[channelID] + if sessions == nil { + sessions = make(map[sessionKey]channelSubscription) + m.bySubscribedChannel[channelID] = sessions + } + sessions[key] = channelSubscription{userID: userID, expiresAt: expiresAt} +} + +// OnlineChannelSubscriberUserIDs returns users for which at least one live +// session still holds an unexpired short-poll subscription. The user is +// deduplicated because passive updates are subsequently pushed account-wide. +func (m *SessionManager) OnlineChannelSubscriberUserIDs(channelID int64, limit int) []int64 { + return m.onlineChannelSubscriberUserIDsExcluding(channelID, nil, limit) +} + +func (m *SessionManager) OnlineChannelSubscriberUserIDsExcluding(channelID int64, exclude map[int64]struct{}, limit int) []int64 { + return m.onlineChannelSubscriberUserIDsExcluding(channelID, exclude, limit) +} + +func (m *SessionManager) onlineChannelSubscriberUserIDsExcluding(channelID int64, exclude map[int64]struct{}, limit int) []int64 { + if channelID == 0 { + return nil + } + now := time.Now().UnixNano() + m.mu.Lock() + defer m.mu.Unlock() + sessions := m.bySubscribedChannel[channelID] + if len(sessions) == 0 { + return nil + } + out := make([]int64, 0, positiveLimitOrLen(limit, len(sessions))) + seen := make(map[int64]struct{}, positiveLimitOrLen(limit, len(sessions))) + for key, subscription := range sessions { + if subscription.expiresAt <= now { + m.removeChannelSubscriptionLocked(key, channelID) + continue + } + if subscription.userID == 0 { + continue + } + if _, ok := exclude[subscription.userID]; ok { + continue + } + c, ok := m.bySession[key] + if !ok || c.userID.Load() != subscription.userID { + m.removeChannelSubscriptionLocked(key, channelID) + continue + } + if _, ok := seen[subscription.userID]; ok { + continue + } + seen[subscription.userID] = struct{}{} + out = append(out, subscription.userID) + if limit > 0 && len(out) >= limit { + break + } + } + return out +} + // ChannelMembershipGeneration 返回该 session 的 membership 索引修订号。 // 全量同步方必须在读取持久成员列表【之前】采样,并经 SetSessionChannelMemberships // 带回比对;session 不在线时返回 0(后续 Set 也会因查不到连接而放弃)。 @@ -2031,14 +2135,17 @@ func (m *SessionManager) OnlineChannelMemberUserIDsExcluding(channelID int64, ex return out } -// OnlineChannelIDsSnapshot returns every channel with at least one live joined-member session in -// strictly ascending order. The global SessionManager lock is held only while copying map keys; +// OnlineChannelIDsSnapshot returns every channel with at least one live joined-member session or +// unexpired passive subscriber in strictly ascending order. The global SessionManager lock is held +// only while copying map keys and pruning expired subscription entries; // sorting and all recovery database work happen after unlock. The fixed saturation-recovery actor // is the sole caller, so its exceptional-path temporary memory is one int64 slice (peak about 8*C // bytes) rather than repeated O(C) scans under the connection/membership lock. func (m *SessionManager) OnlineChannelIDsSnapshot() []int64 { - m.mu.RLock() - out := make([]int64, 0, len(m.byMemberChannel)) + m.mu.Lock() + now := time.Now().UnixNano() + seen := make(map[int64]struct{}, len(m.byMemberChannel)+len(m.bySubscribedChannel)) + out := make([]int64, 0, len(m.byMemberChannel)+len(m.bySubscribedChannel)) for channelID, sessions := range m.byMemberChannel { if channelID <= 0 || len(sessions) == 0 { continue @@ -2053,9 +2160,35 @@ func (m *SessionManager) OnlineChannelIDsSnapshot() []int64 { if !live { continue } + seen[channelID] = struct{}{} out = append(out, channelID) } - m.mu.RUnlock() + for channelID, sessions := range m.bySubscribedChannel { + if channelID <= 0 || len(sessions) == 0 { + continue + } + live := false + for key, subscription := range sessions { + if subscription.expiresAt <= now { + m.removeChannelSubscriptionLocked(key, channelID) + continue + } + c, ok := m.bySession[key] + if !ok || c.userID.Load() != subscription.userID { + m.removeChannelSubscriptionLocked(key, channelID) + continue + } + live = true + } + if !live { + continue + } + if _, exists := seen[channelID]; exists { + continue + } + out = append(out, channelID) + } + m.mu.Unlock() sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) return out } @@ -2105,8 +2238,7 @@ func (m *SessionManager) removeLocked(c *Conn, dropPending bool) int64 { if uid != 0 { removeUserIndex(m.byUser, uid, key) } - m.clearChannelInterestsLocked(key) - m.clearChannelMembershipsLocked(c, key) + m.clearSessionChannelIndexesLocked(c, key) if dropPending { m.deletePendingLocked(key) } @@ -2203,6 +2335,50 @@ func (m *SessionManager) clearChannelInterestsLocked(key sessionKey) { m.clearChannelIndexLocked(m.byChannel, m.bySessionChannels, key) } +func (m *SessionManager) clearChannelSubscriptionsLocked(key sessionKey) { + channels := m.bySessionSubscriptions[key] + if len(channels) == 0 { + delete(m.bySessionSubscriptions, key) + return + } + for channelID := range channels { + sessions := m.bySubscribedChannel[channelID] + delete(sessions, key) + if len(sessions) == 0 { + delete(m.bySubscribedChannel, channelID) + } + } + delete(m.bySessionSubscriptions, key) +} + +func (m *SessionManager) pruneSessionSubscriptionsLocked(key sessionKey, now int64) { + channels := m.bySessionSubscriptions[key] + for channelID, expiresAt := range channels { + if expiresAt <= now { + m.removeChannelSubscriptionLocked(key, channelID) + } + } +} + +func (m *SessionManager) removeChannelSubscriptionLocked(key sessionKey, channelID int64) { + channels := m.bySessionSubscriptions[key] + delete(channels, channelID) + if len(channels) == 0 { + delete(m.bySessionSubscriptions, key) + } + sessions := m.bySubscribedChannel[channelID] + delete(sessions, key) + if len(sessions) == 0 { + delete(m.bySubscribedChannel, channelID) + } +} + +func (m *SessionManager) clearSessionChannelIndexesLocked(c *Conn, key sessionKey) { + m.clearChannelInterestsLocked(key) + m.clearChannelSubscriptionsLocked(key) + m.clearChannelMembershipsLocked(c, key) +} + // clearChannelMembershipsLocked 整体清除某连接的 membership 索引并递增其修订号, // 使在飞的全量同步(SetSessionChannelMemberships)能检测到清除并放弃过期替换。 func (m *SessionManager) clearChannelMembershipsLocked(c *Conn, key sessionKey) { diff --git a/internal/mtprotoedge/session_manager_test.go b/internal/mtprotoedge/session_manager_test.go index 0caaccf2..5db9f1ff 100644 --- a/internal/mtprotoedge/session_manager_test.go +++ b/internal/mtprotoedge/session_manager_test.go @@ -858,6 +858,51 @@ func TestSessionManagerChannelInterestIndex(t *testing.T) { } } +func TestSessionManagerChannelSubscriptionsAreBoundedDeduplicatedAndExpire(t *testing.T) { + sm := NewSessionManager(zaptest.NewLogger(t)) + rawOne := [8]byte{1, 2, 3} + rawTwo := [8]byte{4, 5, 6} + first := &Conn{sessionID: 41, authKeyID: rawOne} + second := &Conn{sessionID: 42, authKeyID: rawTwo} + sm.Register(first) + sm.Register(second) + sm.BindUserForAuthKey(rawOne, 41, 100) + sm.BindUserForAuthKey(rawTwo, 42, 100) + + sm.RefreshChannelSubscription(rawOne, 41, 100, 10, time.Second) + sm.RefreshChannelSubscription(rawTwo, 42, 100, 10, time.Second) + if got := sm.OnlineChannelSubscriberUserIDs(10, 10); len(got) != 1 || got[0] != 100 { + t.Fatalf("deduplicated subscribers = %v, want [100]", got) + } + if got := sm.OnlineChannelIDsSnapshot(); len(got) != 1 || got[0] != 10 { + t.Fatalf("subscribed channel snapshot = %v, want [10]", got) + } + + for channelID := int64(20); channelID < 20+maxChannelSubscriptionsPerSession; channelID++ { + sm.RefreshChannelSubscription(rawOne, 41, 100, channelID, time.Second) + } + // Channel 10 already consumes one of the ten slots, so channel 29 must be + // refused rather than growing the session-controlled index to eleven. + if got := sm.OnlineChannelSubscriberUserIDs(29, 10); len(got) != 0 { + t.Fatalf("subscriber beyond per-session cap = %v, want empty", got) + } + if got := sm.OnlineChannelSubscriberUserIDsExcluding(10, map[int64]struct{}{100: {}}, 10); len(got) != 0 { + t.Fatalf("excluded subscribers = %v, want empty", got) + } + + sm.RefreshChannelSubscription(rawTwo, 42, 100, 99, 10*time.Millisecond) + time.Sleep(30 * time.Millisecond) + if got := sm.OnlineChannelSubscriberUserIDs(99, 10); len(got) != 0 { + t.Fatalf("expired subscribers = %v, want empty", got) + } + + sm.Unregister(first) + sm.Unregister(second) + if got := sm.OnlineChannelSubscriberUserIDs(10, 10); len(got) != 0 { + t.Fatalf("subscribers after unregister = %v, want empty", got) + } +} + func TestSessionManagerClearsChannelIndexesOnAuthAndReadinessChanges(t *testing.T) { sm := NewSessionManager(zaptest.NewLogger(t)) raw := [8]byte{1, 2, 3} @@ -869,6 +914,7 @@ func TestSessionManagerClearsChannelIndexesOnAuthAndReadinessChanges(t *testing. track := func() { sm.TrackChannelInterest(raw, 42, 100, []int64{10}) + sm.RefreshChannelSubscription(raw, 42, 100, 10, time.Second) sm.SetSessionChannelMemberships(raw, 42, 100, []int64{10}, sm.ChannelMembershipGeneration(raw, 42)) if got := sm.OnlineChannelUserIDs(10, 10); len(got) != 1 || got[0] != 100 { t.Fatalf("channel viewers before cleanup = %v, want [100]", got) @@ -876,6 +922,9 @@ func TestSessionManagerClearsChannelIndexesOnAuthAndReadinessChanges(t *testing. if got := sm.OnlineChannelMemberUserIDs(10, 10); len(got) != 1 || got[0] != 100 { t.Fatalf("channel members before cleanup = %v, want [100]", got) } + if got := sm.OnlineChannelSubscriberUserIDs(10, 10); len(got) != 1 || got[0] != 100 { + t.Fatalf("channel subscribers before cleanup = %v, want [100]", got) + } } assertCleared := func(label string) { if got := sm.OnlineChannelUserIDs(10, 10); len(got) != 0 { @@ -884,6 +933,9 @@ func TestSessionManagerClearsChannelIndexesOnAuthAndReadinessChanges(t *testing. if got := sm.OnlineChannelMemberUserIDs(10, 10); len(got) != 0 { t.Fatalf("%s members = %v, want empty", label, got) } + if got := sm.OnlineChannelSubscriberUserIDs(10, 10); len(got) != 0 { + t.Fatalf("%s subscribers = %v, want empty", label, got) + } } track() diff --git a/internal/rpc/channel_fanout_dispatcher.go b/internal/rpc/channel_fanout_dispatcher.go index 827aebd6..25e8d701 100644 --- a/internal/rpc/channel_fanout_dispatcher.go +++ b/internal/rpc/channel_fanout_dispatcher.go @@ -634,9 +634,9 @@ func (d *channelFanoutDispatcher) Enqueue(reqCtx context.Context, job channelFan d.releaseQueuedJob(job) } d.dropped.Add(1) - if job.scope != channelFanoutMembers || job.pts <= 0 { - // 当前所有 enqueue 入口均为 members + durable pts;若未来新增其它 scope,必须先 - // 定义其 overflow 恢复面,不能误把 viewer-only/no-pts 更新伪装成 channel nudge。 + if job.scope != channelFanoutMembers && job.scope != channelFanoutMessageBox || job.pts <= 0 { + // 只有 durable member/message-box payload 可折叠为 channel nudge;viewer-only/no-pts + // 更新没有 difference 恢复契约,不能伪装成 channel PTS 水位。 d.log.Error("channel fanout queue full for non-coalescible job; overflow contract violated", zap.Int64("channel_id", job.channelID), zap.Int("pts", job.pts), zap.Int("scope", int(job.scope))) d.enqueueMu.RUnlock() @@ -867,7 +867,7 @@ func (r *Router) runChannelFanoutOverflowNudge(ctx context.Context, channelID in if r.deps.Sessions == nil || channelID == 0 || pts <= 0 { return true } - return r.nudgeBeyondCapChannelMembers(ctx, channelID, pts, nil) + return r.nudgeBeyondCapChannelMessageAudience(ctx, channelID, pts, nil) } // runChannelFanoutJob 执行一条 fan-out:与同步 pushChannelUpdatesWithScope 等价,区别是 @@ -918,6 +918,8 @@ func (r *Router) runChannelFanoutJob(ctx context.Context, job channelFanoutJob) // 仅对会推进客户端 channel PtsWaiter 的真实 payload(members scope + 带 channel pts)做。 if job.scope == channelFanoutMembers && job.pts > 0 { r.nudgeBeyondCapChannelMembers(pushCtx, job.channelID, job.pts, seen) + } else if job.scope == channelFanoutMessageBox && job.pts > 0 { + r.nudgeBeyondCapChannelMessageAudience(pushCtx, job.channelID, job.pts, seen) } } @@ -975,7 +977,7 @@ func (r *Router) enqueueChannelMessageFanout(ctx context.Context, originUserID i fanoutCache := newViewerPeerCache(r) ownerIDs := channelMessageFanoutOwnerIDs(res, extraUserIDs) skip := skipDeliverySet(res.SkipDeliveryUserIDs) - r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMembers, originUserID, res.Channel.ID, res.Event.Pts, res.Recipients, + r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, res.Channel.ID, res.Event.Pts, res.Recipients, 0, func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) @@ -1056,7 +1058,7 @@ func (r *Router) enqueueChannelEditMessageFanout(ctx context.Context, originUser fanoutCache := newViewerPeerCache(r) ownerIDs := channelEditMessageFanoutOwnerIDs(res) nudgePts := max(res.Event.Pts, res.ServiceEvent.Pts) - r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMembers, originUserID, res.Channel.ID, nudgePts, res.Recipients, + r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, res.Channel.ID, nudgePts, res.Recipients, 0, func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) @@ -1073,7 +1075,7 @@ func (r *Router) enqueueChannelMessagesFanout(ctx context.Context, originUserID, r.enqueueBotAPIChannelMessagesUpdate(ctx, originUserID, results) fanoutCache := newViewerPeerCache(r) ownerIDs := channelMessagesFanoutOwnerIDs(results, extraUserIDs) - r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMembers, originUserID, channelID, pts, recipients, + r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, channelID, pts, recipients, int64(len(results))*(64<<10), func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) @@ -1136,3 +1138,81 @@ func (r *Router) nudgeBeyondCapChannelMembers(ctx context.Context, channelID int } return ctx.Err() == nil } + +// nudgeBeyondCapChannelMessageAudience extends member recovery to users with an +// unexpired public-channel short-poll subscription. Subscribers are selected +// first (normally a tiny set), then the remaining bounded capacity is filled +// with joined members. One authoritative batched audience check prevents stale +// runtime indexes from leaking even the channel id/pts to revoked viewers. +func (r *Router) nudgeBeyondCapChannelMessageAudience(ctx context.Context, channelID int64, pts int, delivered map[int64]struct{}) bool { + if r.deps.Sessions == nil || channelID == 0 || pts <= 0 { + return true + } + if r.deps.Channels == nil { + return r.nudgeBeyondCapChannelMembers(ctx, channelID, pts, delivered) + } + audience, ok := r.deps.Channels.(ChannelMessageAudienceService) + if !ok { + return r.nudgeBeyondCapChannelMembers(ctx, channelID, pts, delivered) + } + limit := r.channelNudgeMaxTargets() + excluded := make(map[int64]struct{}, len(delivered)+16) + for userID := range delivered { + excluded[userID] = struct{}{} + } + candidates := make([]int64, 0, min(limit, 64)) + if subscriptions, ok := r.deps.Sessions.(ChannelSubscriptionProvider); ok { + for _, userID := range subscriptions.OnlineChannelSubscriberUserIDsExcluding(channelID, excluded, limit) { + if userID == 0 { + continue + } + excluded[userID] = struct{}{} + candidates = append(candidates, userID) + if len(candidates) >= limit { + break + } + } + } + if len(candidates) < limit { + if members, ok := r.deps.Sessions.(ChannelNudgeProvider); ok { + for _, userID := range members.OnlineChannelMemberUserIDsExcluding(channelID, excluded, limit-len(candidates)) { + if userID == 0 { + continue + } + excluded[userID] = struct{}{} + candidates = append(candidates, userID) + } + } + } + if len(candidates) == 0 { + return true + } + targets, err := audience.FilterMessageAudienceIDs(ctx, channelID, candidates) + if err != nil { + r.log.Warn("channel message audience nudge authorization failed", + zap.Int64("channel_id", channelID), zap.Int("pts", pts), zap.Error(err)) + return false + } + if len(targets) == 0 { + return true + } + date := int(r.clock.Now().Unix()) + tooLong := &tg.UpdateChannelTooLong{ChannelID: channelID} + tooLong.SetPts(pts) + updates := &tg.Updates{ + Updates: []tg.UpdateClass{tooLong}, + Users: []tg.UserClass{}, + Chats: []tg.ChatClass{}, + Date: date, + Seq: 0, + } + for _, userID := range targets { + select { + case <-ctx.Done(): + return false + default: + } + r.pushUserUpdates(ctx, userID, updates) + } + return true +} diff --git a/internal/rpc/channel_interest.go b/internal/rpc/channel_interest.go index f65ed54c..a0d986f1 100644 --- a/internal/rpc/channel_interest.go +++ b/internal/rpc/channel_interest.go @@ -2,6 +2,7 @@ package rpc import ( "context" + "time" "go.uber.org/zap" @@ -9,6 +10,7 @@ import ( ) const channelMembershipSyncPageSize = domain.MaxSynchronousChannelDialogFanout +const publicChannelSubscriptionTTL = 75 * time.Second func (r *Router) trackChannelInterest(ctx context.Context, userID int64, channelIDs ...int64) { if userID == 0 || r.deps.Sessions == nil { @@ -37,6 +39,25 @@ func (r *Router) clearChannelInterest(ctx context.Context, userID int64) { r.trackChannelInterest(ctx, userID) } +func (r *Router) refreshPublicChannelSubscription(ctx context.Context, userID, channelID int64) { + if userID == 0 || channelID == 0 || r.deps.Sessions == nil { + return + } + provider, ok := r.deps.Sessions.(ChannelSubscriptionProvider) + if !ok { + return + } + rawAuthKeyID, ok := RawAuthKeyIDFrom(ctx) + if !ok { + return + } + sessionID, ok := SessionIDFrom(ctx) + if !ok { + return + } + provider.RefreshChannelSubscription(rawAuthKeyID, sessionID, userID, channelID, publicChannelSubscriptionTTL) +} + func (r *Router) syncSessionChannelMemberships(ctx context.Context, userID int64) { if userID == 0 || r.deps.Sessions == nil || r.deps.Channels == nil { return diff --git a/internal/rpc/channels_constants.go b/internal/rpc/channels_constants.go index 62f3c760..5ca6539e 100644 --- a/internal/rpc/channels_constants.go +++ b/internal/rpc/channels_constants.go @@ -17,5 +17,8 @@ const ( const ( channelFanoutMembers channelFanoutScope = iota channelFanoutViewers + // channelFanoutMessageBox is the durable channel message-box audience: + // online members plus users with an unexpired public short-poll subscription. + channelFanoutMessageBox channelFanoutExplicit ) diff --git a/internal/rpc/channels_legacy_chat.go b/internal/rpc/channels_legacy_chat.go index 84766559..e3b16959 100644 --- a/internal/rpc/channels_legacy_chat.go +++ b/internal/rpc/channels_legacy_chat.go @@ -615,7 +615,7 @@ func (r *Router) enqueueChannelWallpaperFanout(ctx context.Context, originUserID } fanoutCache := newViewerPeerCache(r) ownerIDs := channelMessageFanoutOwnerIDs(sendRes, nil) - r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMembers, originUserID, res.Channel.ID, res.Event.Pts, res.Recipients, + r.enqueueChannelFanoutWithPrefetch(ctx, channelFanoutMessageBox, originUserID, res.Channel.ID, res.Event.Pts, res.Recipients, 0, func(bgCtx context.Context, viewers []int64) { r.prefetchChannelFanoutUsers(bgCtx, fanoutCache, viewers, ownerIDs) diff --git a/internal/rpc/channels_messages.go b/internal/rpc/channels_messages.go index f7a5c80c..a5703fb6 100644 --- a/internal/rpc/channels_messages.go +++ b/internal/rpc/channels_messages.go @@ -335,13 +335,13 @@ func (r *Router) onChannelsDeleteMessages(ctx context.Context, req *tg.ChannelsD if res.Event.Pts != 0 { // 删除 fan-out 异步化(设计 Phase 0)。channelDeleteMessagesUpdates 是纯 CPU 构建 // (不碰 PG、不取 ctx),async 无竞态;同 channel 串行保 pts 单调。 - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, res.Channel, res.Event) }) // 被删 broadcast post 的讨论组转发根级联删除同样要让讨论组成员收敛。 for _, cascade := range res.DiscussionDeletes { cascade := cascade - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, cascade.Channel.ID, cascade.Event.Pts, cascade.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, cascade.Channel.ID, cascade.Event.Pts, cascade.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, cascade.Channel, cascade.Event) }) } @@ -357,12 +357,12 @@ func (r *Router) NotifyModerationChannelDeletion(ctx context.Context, res domain if r == nil || res.Event.Pts == 0 { return } - r.enqueueChannelFanout(ctx, channelFanoutMembers, domain.OfficialSystemUserID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, domain.OfficialSystemUserID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, res.Channel, res.Event) }) for _, cascade := range res.DiscussionDeletes { cascade := cascade - r.enqueueChannelFanout(ctx, channelFanoutMembers, domain.OfficialSystemUserID, cascade.Channel.ID, cascade.Event.Pts, cascade.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, domain.OfficialSystemUserID, cascade.Channel.ID, cascade.Event.Pts, cascade.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, cascade.Channel, cascade.Event) }) } @@ -403,7 +403,7 @@ func (r *Router) onChannelsDeleteHistory(ctx context.Context, req *tg.ChannelsDe pushBatch := func(batch domain.DeleteChannelHistoryResult) *tg.Updates { out := r.channelDeleteMessagesUpdates(userID, batch.Channel, batch.Event) // 每批 fan-out 异步化;批次按 pts 递增顺序入同一 channel 分片 → FIFO 保单调。 - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, batch.Channel.ID, batch.Event.Pts, batch.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, batch.Channel.ID, batch.Event.Pts, batch.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, batch.Channel, batch.Event) }) return out diff --git a/internal/rpc/channels_public_preview_rpc_test.go b/internal/rpc/channels_public_preview_rpc_test.go index fe40fdbf..d4cdd678 100644 --- a/internal/rpc/channels_public_preview_rpc_test.go +++ b/internal/rpc/channels_public_preview_rpc_test.go @@ -16,6 +16,7 @@ import ( func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { ctx := context.Background() + sessions := &captureSessions{} userStore := memory.NewUserStore() owner, _ := userStore.Create(ctx, domain.User{AccessHash: 92001, Phone: "15550092001", FirstName: "Owner"}) viewer, _ := userStore.Create(ctx, domain.User{AccessHash: 92002, Phone: "15550092002", FirstName: "Viewer"}) @@ -26,6 +27,7 @@ func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { Users: appusers.NewService(userStore), Channels: channelService, Dialogs: dialogService, + Sessions: sessions, }, zaptest.NewLogger(t), clock.System) public, err := channelService.CreateChannel(ctx, owner.ID, domain.CreateChannelRequest{ Title: "Public Preview RPC", @@ -123,7 +125,8 @@ func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { t.Fatalf("history chat = %T %+v, want left public channel", history.Chats[0], history.Chats[0]) } - diff, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, viewer.ID), &tg.UpdatesGetChannelDifferenceRequest{ + viewerCtx := WithSessionID(WithRawAuthKeyID(WithUserID(ctx, viewer.ID), [8]byte{9, 2}), 9202) + diff, err := r.onUpdatesGetChannelDifference(viewerCtx, &tg.UpdatesGetChannelDifferenceRequest{ Channel: input, Pts: public.Event.Pts, Limit: 10, @@ -131,9 +134,66 @@ func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { if err != nil { t.Fatalf("non-member getChannelDifference public preview: %v", err) } - emptyDiff, ok := diff.(*tg.UpdatesChannelDifferenceEmpty) - if !ok || !emptyDiff.Final || emptyDiff.Pts != sent.Event.Pts { - t.Fatalf("channel difference = %T %+v, want empty public preview difference at current pts", diff, diff) + fullDiff, ok := diff.(*tg.UpdatesChannelDifference) + if !ok || !fullDiff.Final || fullDiff.Pts != sent.Event.Pts || len(fullDiff.NewMessages) != 1 { + t.Fatalf("channel difference = %T %+v, want one public preview message", diff, diff) + } + message, ok := fullDiff.NewMessages[0].(*tg.Message) + if !ok || message.ID != sent.Message.ID || message.Message != sent.Message.Body { + t.Fatalf("channel difference message = %T %+v, want sent public post", fullDiff.NewMessages[0], fullDiff.NewMessages[0]) + } + if subscribers := sessions.OnlineChannelSubscriberUserIDs(public.Channel.ID, 10); len(subscribers) != 1 || subscribers[0] != viewer.ID { + t.Fatalf("public channel subscribers = %v, want viewer %d", subscribers, viewer.ID) + } + + live, err := channelService.SendMessage(ctx, owner.ID, domain.SendChannelMessageRequest{ + ChannelID: public.Channel.ID, + RandomID: 202, + Message: "public preview live post", + Date: 1700010120, + }) + if err != nil { + t.Fatalf("send live public post: %v", err) + } + sessions.clearMessages() + r.enqueueChannelMessageFanout(WithUserID(ctx, owner.ID), owner.ID, live, nil) + if !fanoutHasID(sessions.pushedUserIDs(), viewer.ID) { + t.Fatalf("live public preview fanout users = %v, want viewer %d", sessions.pushedUserIDs(), viewer.ID) + } + liveUpdates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(liveUpdates.Updates) == 0 { + t.Fatalf("live public preview update = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) + } + foundLive := false + for _, update := range liveUpdates.Updates { + newMessage, ok := update.(*tg.UpdateNewChannelMessage) + if !ok { + continue + } + if item, ok := newMessage.Message.(*tg.Message); ok && item.ID == live.Message.ID && item.Message == live.Message.Body { + foundLive = true + } + } + if !foundLive { + t.Fatalf("live public preview updates = %+v, want new message %d", liveUpdates.Updates, live.Message.ID) + } + sessions.clearMessages() + if ok := r.runChannelFanoutOverflowNudge(ctx, public.Channel.ID, live.Event.Pts); !ok { + t.Fatal("public preview overflow nudge did not complete") + } + if !fanoutHasID(sessions.pushedUserIDs(), viewer.ID) { + t.Fatalf("public preview overflow nudge users = %v, want viewer %d", sessions.pushedUserIDs(), viewer.ID) + } + nudgeUpdates, ok := sessions.lastUserPush().(*tg.Updates) + if !ok || len(nudgeUpdates.Updates) != 1 { + t.Fatalf("public preview overflow nudge = %T %+v", sessions.lastUserPush(), sessions.lastUserPush()) + } + tooLong, ok := nudgeUpdates.Updates[0].(*tg.UpdateChannelTooLong) + if !ok || tooLong.ChannelID != public.Channel.ID { + t.Fatalf("public preview overflow update = %T %+v, want channel %d tooLong", nudgeUpdates.Updates[0], nudgeUpdates.Updates[0], public.Channel.ID) + } + if pts, ok := tooLong.GetPts(); !ok || pts != live.Event.Pts { + t.Fatalf("public preview overflow pts = %d/%v, want %d", pts, ok, live.Event.Pts) } domainPeers, err := r.dialogPeersFromInput(WithUserID(ctx, viewer.ID), viewer.ID, []tg.InputDialogPeerClass{&tg.InputDialogPeer{Peer: peer}}) @@ -147,8 +207,12 @@ func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { if err != nil { t.Fatalf("dialog service public preview: %v", err) } - if len(directPeerDialogs.Dialogs) != 0 || len(directPeerDialogs.ChannelMessages) != 0 || len(directPeerDialogs.Channels) != 0 { - t.Fatalf("direct peer dialogs = %+v, want no public preview dialog/message/channel", directPeerDialogs) + if len(directPeerDialogs.Dialogs) != 1 || len(directPeerDialogs.ChannelMessages) != 0 || len(directPeerDialogs.Channels) != 1 { + t.Fatalf("direct peer dialogs = %+v, want one zero-top public preview bootstrap", directPeerDialogs) + } + directDialog := directPeerDialogs.Dialogs[0] + if directDialog.TopMessage != 0 || !directDialog.ChannelLeft || directDialog.Pts != live.Event.Pts { + t.Fatalf("direct public preview dialog = %+v, want left zero-top bootstrap", directDialog) } peerDialogsReq := &tg.MessagesGetPeerDialogsRequest{ @@ -166,8 +230,17 @@ func TestPublicChannelPreviewRPCsAllowNonMember(t *testing.T) { if !ok { t.Fatalf("getPeerDialogs response = %T, want peer dialogs", peerDialogsEnc) } - if len(peerDialogs.Dialogs) != 0 || len(peerDialogs.Messages) != 0 || len(peerDialogs.Chats) != 0 { - t.Fatalf("peer dialogs = %+v, want no public preview dialog/message/channel", peerDialogs) + if len(peerDialogs.Dialogs) != 1 || len(peerDialogs.Messages) != 0 || len(peerDialogs.Chats) != 1 { + t.Fatalf("peer dialogs = %+v, want one zero-top public preview bootstrap", peerDialogs) + } + tgDialog, ok := peerDialogs.Dialogs[0].(*tg.Dialog) + if !ok || tgDialog.TopMessage != 0 || tgDialog.ReadInboxMaxID != 0 || + tgDialog.ReadOutboxMaxID != 0 || tgDialog.UnreadCount != 0 { + t.Fatalf("peer dialog = %T %+v, want zero-state dialog", peerDialogs.Dialogs[0], peerDialogs.Dialogs[0]) + } + peerDialogChat, ok := peerDialogs.Chats[0].(*tg.Channel) + if !ok || !peerDialogChat.Left || peerDialogChat.ID != public.Channel.ID { + t.Fatalf("peer dialog chat = %T %+v, want left public channel", peerDialogs.Chats[0], peerDialogs.Chats[0]) } if _, err := channelService.JoinChannel(ctx, viewer.ID, public.Channel.ID, 1700010120); err != nil { diff --git a/internal/rpc/channels_stubs.go b/internal/rpc/channels_stubs.go index 9527c716..5d4ff631 100644 --- a/internal/rpc/channels_stubs.go +++ b/internal/rpc/channels_stubs.go @@ -201,7 +201,7 @@ func (r *Router) onMessagesUnpinAllMessages(ctx context.Context, req *tg.Message return nil, channelAdminErr(err) } r.invalidateRPCProjectionForChannel(res.Channel.ID) - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelPinnedUpdates(viewerUserID, res) }) return &tg.MessagesAffectedHistory{ @@ -370,23 +370,41 @@ func (r *Router) channelFanoutRecipients(ctx context.Context, scope channelFanou online = provider.OnlineChannelMemberUserIDs(channelID, domain.MaxChannelRealtimeFanout) case channelFanoutViewers: online = provider.OnlineChannelUserIDs(channelID, domain.MaxChannelRealtimeFanout) + case channelFanoutMessageBox: + online = provider.OnlineChannelMemberUserIDs(channelID, domain.MaxChannelRealtimeFanout) + if subscriptions, ok := r.deps.Sessions.(ChannelSubscriptionProvider); ok { + online = append(online, subscriptions.OnlineChannelSubscriberUserIDs(channelID, domain.MaxChannelRealtimeFanout)...) + } } if len(online) == 0 { return uniqueRecipientIDs(explicit) } - active, err := r.deps.Channels.FilterActiveMemberIDs(ctx, channelID, online) + var ( + authorized []int64 + err error + ) + if scope == channelFanoutMembers { + authorized, err = r.deps.Channels.FilterActiveMemberIDs(ctx, channelID, online) + } else if audience, ok := r.deps.Channels.(ChannelMessageAudienceService); ok { + authorized, err = audience.FilterMessageAudienceIDs(ctx, channelID, online) + } else { + // Test/minimal adapters without public-preview authorization retain the + // former member-only behavior; production channels.Service implements + // ChannelMessageAudienceService. + authorized, err = r.deps.Channels.FilterActiveMemberIDs(ctx, channelID, online) + } if err != nil { return uniqueRecipientIDs(explicit) } - if len(active) == 0 && len(explicit) == 0 { + if len(authorized) == 0 && len(explicit) == 0 { return nil } - if len(active) > domain.MaxChannelRealtimeFanout { - active = active[:domain.MaxChannelRealtimeFanout] + if len(authorized) > domain.MaxChannelRealtimeFanout { + authorized = authorized[:domain.MaxChannelRealtimeFanout] } - out := uniqueRecipientIDs(active) + out := uniqueRecipientIDs(authorized) seen := make(map[int64]struct{}, len(out)+len(explicit)) - for _, userID := range active { + for _, userID := range authorized { if userID == 0 { continue } diff --git a/internal/rpc/channels_topics.go b/internal/rpc/channels_topics.go index 0b769fe8..e57bcf39 100644 --- a/internal/rpc/channels_topics.go +++ b/internal/rpc/channels_topics.go @@ -93,7 +93,7 @@ func (r *Router) onMessagesUpdatePinnedMessage(ctx context.Context, req *tg.Mess // builder 无 Users 数组(仅 pinned update + ChatMin),无需 owner 预热。pin 的真实变更由 // UpdatePinnedChannelMessages{pts} 承载、可经 getChannelDifference 兜底,bundled 的无 pts // UpdateChannel 对 pin 冗余(pts payload 已含变更),丢弃无害——与 unpinAll 取舍一致。 - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelPinnedUpdates(viewerUserID, res) }) return updates, nil diff --git a/internal/rpc/channels_updates.go b/internal/rpc/channels_updates.go index 66bebb59..bf63f3b6 100644 --- a/internal/rpc/channels_updates.go +++ b/internal/rpc/channels_updates.go @@ -45,6 +45,12 @@ func (r *Router) onUpdatesGetChannelDifference(ctx context.Context, req *tg.Upda } return nil, channelInvalidErr(err) } + if diff.Channel.Username != "" && diff.Self.Status != domain.ChannelMemberActive { + // Telegram's public-channel passive delivery is enabled only after a + // successful short-poll difference. The runtime subscription is renewed + // by subsequent polls and never creates membership/dialog/read state. + r.refreshPublicChannelSubscription(ctx, userID, channelID) + } diff = r.enrichChannelDifference(ctx, userID, diff) out := tgChannelDifference(userID, diff) if linked, ok := r.linkedDiscussionChat(ctx, userID, channelID); ok { diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 7375f64c..495a0d3b 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -214,6 +214,16 @@ type OnlineUserProvider interface { OnlineChannelMemberUserIDs(channelID int64, limit int) []int64 } +// ChannelSubscriptionProvider is the bounded process-local implementation of +// Telegram's public-channel short-poll subscription. A successful +// updates.getChannelDifference refresh from one session enables passive channel +// updates for the whole user account until the subscription expires. +type ChannelSubscriptionProvider interface { + RefreshChannelSubscription(rawAuthKeyID [8]byte, sessionID, userID, channelID int64, ttl time.Duration) + OnlineChannelSubscriberUserIDs(channelID int64, limit int) []int64 + OnlineChannelSubscriberUserIDsExcluding(channelID int64, exclude map[int64]struct{}, limit int) []int64 +} + // ChannelNudgeProvider 暴露「频道在线成员中排除已投递集合后的剩余 user id」,用于 >cap // 在线成员的 UpdateChannelTooLong nudge(P0-8)。SessionManager 实现;测试/未装配 fake 可不实现 // (type-assert 失败时跳过 nudge,不影响完整 payload 投递)。 @@ -761,6 +771,13 @@ type ChannelsService interface { FilterActiveMemberIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) } +// ChannelMessageAudienceService is the optional production authorization +// boundary for public short-poll subscribers. Lightweight test/domain adapters +// that only model joined members may omit it and retain member-only behavior. +type ChannelMessageAudienceService interface { + FilterMessageAudienceIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) +} + // ChannelAuthoritativeProjectionService bypasses long-lived channel read // models for a durable channel_state refresh emitted by an admin mutation. type ChannelAuthoritativeProjectionService interface { diff --git a/internal/rpc/expiry_dispatcher.go b/internal/rpc/expiry_dispatcher.go index a02d7af9..4d7ad82e 100644 --- a/internal/rpc/expiry_dispatcher.go +++ b/internal/rpc/expiry_dispatcher.go @@ -97,7 +97,7 @@ func (d *ExpiryDispatcher) dispatchChannels(ctx context.Context, now int) bool { d.log.Warn("delete expired channel messages", zap.Int64("channel_id", req.ChannelID), zap.Ints("ids", req.IDs), zap.Error(err)) continue } - d.router.enqueueChannelFanout(ctx, channelFanoutMembers, req.UserID, req.ChannelID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + d.router.enqueueChannelFanout(ctx, channelFanoutMessageBox, req.UserID, req.ChannelID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return d.router.channelDeleteMessagesUpdates(viewerUserID, res.Channel, res.Event) }) } diff --git a/internal/rpc/messages_delete.go b/internal/rpc/messages_delete.go index 69b2f8be..b66fd3a7 100644 --- a/internal/rpc/messages_delete.go +++ b/internal/rpc/messages_delete.go @@ -68,7 +68,7 @@ func (r *Router) onMessagesDeleteHistory(ctx context.Context, req *tg.MessagesDe return nil, channelDeleteErr(err) } if res.Event.Pts != 0 { - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return r.channelDeleteMessagesUpdates(viewerUserID, res.Channel, res.Event) }) return &tg.MessagesAffectedHistory{Pts: res.Event.Pts, PtsCount: res.Event.PtsCount, Offset: res.Offset}, nil diff --git a/internal/rpc/messages_forum.go b/internal/rpc/messages_forum.go index b85a7442..4a0e939a 100644 --- a/internal/rpc/messages_forum.go +++ b/internal/rpc/messages_forum.go @@ -207,7 +207,7 @@ func (r *Router) onMessagesDeleteTopicHistory(ctx context.Context, req *tg.Messa return nil, forumTopicError(err) } if res.Event.Pts != 0 { - r.enqueueChannelFanout(ctx, channelFanoutMembers, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { + r.enqueueChannelFanout(ctx, channelFanoutMessageBox, userID, res.Channel.ID, res.Event.Pts, res.Recipients, func(_ context.Context, viewerUserID int64) *tg.Updates { return &tg.Updates{ Updates: []tg.UpdateClass{tgChannelUpdate(viewerUserID, res.Event)}, Chats: []tg.ChatClass{tgChannelChatMin(viewerUserID, res.Channel)}, diff --git a/internal/rpc/rpc_testkit_sessions_test.go b/internal/rpc/rpc_testkit_sessions_test.go index df973112..457e8ad7 100644 --- a/internal/rpc/rpc_testkit_sessions_test.go +++ b/internal/rpc/rpc_testkit_sessions_test.go @@ -3,6 +3,7 @@ package rpc import ( "context" "sync" + "time" "github.com/iamxvbaba/td/bin" "github.com/iamxvbaba/td/proto" @@ -10,22 +11,23 @@ import ( ) type captureSessions struct { - mu sync.Mutex - rawAuthKeyID [8]byte - sessionID int64 - userID int64 - userResolved bool - authKeyID [8]byte - authKeyResolved bool - receives bool - receivesCalls int - messageType proto.MessageType - message bin.Encoder - userMessage bin.Encoder // 最近一次 PushToUser* 的消息(与 message 区分:message 也被 PushToSession 覆盖) - pushUserIDs []int64 - onlineUserIDs []int64 - channelViewers map[int64][]int64 - channelMembers map[int64][]int64 + mu sync.Mutex + rawAuthKeyID [8]byte + sessionID int64 + userID int64 + userResolved bool + authKeyID [8]byte + authKeyResolved bool + receives bool + receivesCalls int + messageType proto.MessageType + message bin.Encoder + userMessage bin.Encoder // 最近一次 PushToUser* 的消息(与 message 区分:message 也被 PushToSession 覆盖) + pushUserIDs []int64 + onlineUserIDs []int64 + channelViewers map[int64][]int64 + channelMembers map[int64][]int64 + channelSubscribers map[int64][]int64 // channelViewersLimit 记录最近一次 OnlineChannelUserIDs 收到的 limit,验证 fan-out 封顶传参。 channelViewersLimit int } @@ -258,6 +260,42 @@ func (s *captureSessions) OnlineChannelUserIDs(channelID int64, limit int) []int return limitIDs(s.channelViewers[channelID], limit) } +func (s *captureSessions) RefreshChannelSubscription(_ [8]byte, _ int64, userID, channelID int64, _ time.Duration) { + s.mu.Lock() + defer s.mu.Unlock() + if s.channelSubscribers == nil { + s.channelSubscribers = make(map[int64][]int64) + } + for _, existing := range s.channelSubscribers[channelID] { + if existing == userID { + return + } + } + s.channelSubscribers[channelID] = append(s.channelSubscribers[channelID], userID) +} + +func (s *captureSessions) OnlineChannelSubscriberUserIDs(channelID int64, limit int) []int64 { + s.mu.Lock() + defer s.mu.Unlock() + return limitIDs(s.channelSubscribers[channelID], limit) +} + +func (s *captureSessions) OnlineChannelSubscriberUserIDsExcluding(channelID int64, exclude map[int64]struct{}, limit int) []int64 { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]int64, 0, len(s.channelSubscribers[channelID])) + for _, userID := range s.channelSubscribers[channelID] { + if _, skip := exclude[userID]; skip { + continue + } + out = append(out, userID) + if limit > 0 && len(out) >= limit { + break + } + } + return out +} + func (s *captureSessions) ChannelMembershipGeneration(_ [8]byte, _ int64) int64 { return 0 } func (s *captureSessions) SetSessionChannelMemberships(_ [8]byte, _ int64, userID int64, channelIDs []int64, _ int64) { diff --git a/internal/store/channel.go b/internal/store/channel.go index e68a32bc..519af694 100644 --- a/internal/store/channel.go +++ b/internal/store/channel.go @@ -183,6 +183,10 @@ type ChannelStore interface { ListActiveChannelMembers(ctx context.Context, viewerUserID, channelID int64, limit int) (domain.Channel, domain.ChannelMember, []domain.ChannelMember, error) ListChannelInviteAdminMemberIDs(ctx context.Context, channelID int64, limit int) ([]int64, error) FilterActiveChannelMemberIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) + // FilterChannelMessageAudienceIDs authoritatively intersects a bounded online + // candidate set with users allowed to receive channel message-box updates: + // active members plus non-banned public-channel preview subscribers. + FilterChannelMessageAudienceIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) MaxChannelPts(ctx context.Context, channelID int64) (int, error) // MaxChannelPtsBatch returns existing channel watermarks with one bounded store round trip. // Missing/deleted ids are omitted so a stale process-local membership key cannot poison the diff --git a/internal/store/memory/channel_members.go b/internal/store/memory/channel_members.go index 3aec3389..14e63cc0 100644 --- a/internal/store/memory/channel_members.go +++ b/internal/store/memory/channel_members.go @@ -892,6 +892,42 @@ func (s *ChannelStore) FilterActiveChannelMemberIDs(_ context.Context, channelID return out, nil } +func (s *ChannelStore) FilterChannelMessageAudienceIDs(_ context.Context, channelID int64, userIDs []int64) ([]int64, error) { + s.mu.RLock() + defer s.mu.RUnlock() + if channelID == 0 || len(userIDs) == 0 { + return nil, nil + } + channel, ok := s.channels[channelID] + if !ok || channel.Deleted { + return nil, nil + } + public := publicPreviewableChannel(channel) + members := s.members[channelID] + out := make([]int64, 0, len(userIDs)) + seen := make(map[int64]struct{}, len(userIDs)) + for _, userID := range userIDs { + if userID == 0 { + continue + } + if _, ok := seen[userID]; ok { + continue + } + seen[userID] = struct{}{} + member, found := members[userID] + if member.BannedRights.ViewMessages || + member.Status == domain.ChannelMemberKicked || + member.Status == domain.ChannelMemberBanned { + continue + } + if member.Status == domain.ChannelMemberActive || public && (!found || member.Status == domain.ChannelMemberLeft) { + out = append(out, userID) + } + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out, nil +} + func (s *ChannelStore) ListActiveChannelMembers(_ context.Context, viewerUserID, channelID int64, limit int) (domain.Channel, domain.ChannelMember, []domain.ChannelMember, error) { s.mu.RLock() defer s.mu.RUnlock() diff --git a/internal/store/memory/channel_updates.go b/internal/store/memory/channel_updates.go index 033daca0..2afdc3f9 100644 --- a/internal/store/memory/channel_updates.go +++ b/internal/store/memory/channel_updates.go @@ -31,16 +31,6 @@ func (s *ChannelStore) ListChannelDifference(_ context.Context, req domain.Chann if preview { dialog = previewChannelDialog(req.UserID, channel, member) } - if preview && member.Status != domain.ChannelMemberActive { - return domain.ChannelDifference{ - Channel: channel, - Self: member, - Pts: channel.Pts, - Final: true, - Timeout: 30, - Dialog: dialog, - }, nil - } checkpoint := s.channelUpdateCheckpointLocked(req.ChannelID, channel) if req.Pts < checkpoint.RetainedThroughPts || channel.Pts-req.Pts > limit { messages := make([]domain.ChannelMessage, 0, domain.MaxChannelDifferenceTooLongMessages) @@ -81,10 +71,15 @@ func (s *ChannelStore) ListChannelDifference(_ context.Context, req domain.Chann } } } + scanned := 0 for _, event := range s.events[req.ChannelID] { if event.Pts <= req.Pts { continue } + if scanned >= limit { + break + } + scanned++ lastPts = event.Pts visible, ok := domain.FilterChannelUpdateEventForAvailableMinID(cloneChannelEvent(event), member.AvailableMinID) if !ok { @@ -106,7 +101,7 @@ func (s *ChannelStore) ListChannelDifference(_ context.Context, req domain.Chann Channel: channel, Self: member, Pts: maxInt(lastPts, req.Pts), - Final: true, + Final: lastPts >= channel.Pts, Timeout: 30, Dialog: dialog, }, nil diff --git a/internal/store/postgres/channel_difference_integration_test.go b/internal/store/postgres/channel_difference_integration_test.go index 842949e9..8a0d8577 100644 --- a/internal/store/postgres/channel_difference_integration_test.go +++ b/internal/store/postgres/channel_difference_integration_test.go @@ -116,7 +116,7 @@ func TestChannelStoreDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { } } -func TestChannelStorePublicPreviewDifferenceSkipsNonMemberMessages(t *testing.T) { +func TestChannelStorePublicPreviewDifferenceReplaysVisibleMessages(t *testing.T) { pool := testPool(t) ctx := context.Background() suffix := randomSuffix(t) @@ -183,12 +183,57 @@ func TestChannelStorePublicPreviewDifferenceSkipsNonMemberMessages(t *testing.T) if err != nil { t.Fatalf("list public preview difference: %v", err) } - if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.Events) != 0 || len(diff.NewMessages) != 0 || len(diff.OtherUpdates) != 0 { - t.Fatalf("preview diff = %+v, want empty public preview difference at current pts", diff) + if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.Events) != 1 || len(diff.NewMessages) != 1 || len(diff.OtherUpdates) != 0 { + t.Fatalf("preview diff = %+v, want one durable public preview message at pts %d", diff, sent.Event.Pts) + } + if diff.NewMessages[0].ID != sent.Message.ID || diff.NewMessages[0].Body != sent.Message.Body { + t.Fatalf("preview diff message = %+v, want sent message %+v", diff.NewMessages[0], sent.Message) } if diff.Dialog.UnreadCount != 0 || diff.Dialog.ReadInboxMaxID < sent.Message.ID { t.Fatalf("preview diff dialog = %+v, want read-only public preview dialog", diff.Dialog) } + edited, err := channels.EditChannelMessage(ctx, domain.EditChannelMessageRequest{ + UserID: owner.ID, ChannelID: channelID, ID: sent.Message.ID, + Message: "public preview edited", EditDate: 1700000372, + }) + if err != nil { + t.Fatalf("edit public preview message: %v", err) + } + pinned, err := channels.UpdatePinnedMessage(ctx, domain.UpdateChannelPinnedMessageRequest{ + UserID: owner.ID, ChannelID: channelID, MessageID: sent.Message.ID, + Pinned: true, Date: 1700000373, + }) + if err != nil { + t.Fatalf("pin public preview message: %v", err) + } + deleted, err := channels.DeleteChannelMessages(ctx, domain.DeleteChannelMessagesRequest{ + UserID: owner.ID, ChannelID: channelID, IDs: []int{sent.Message.ID}, Date: 1700000374, + }) + if err != nil { + t.Fatalf("delete public preview message: %v", err) + } + mutations, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: viewer.ID, ChannelID: channelID, Pts: sent.Event.Pts, Limit: 10, + }) + if err != nil { + t.Fatalf("list public preview mutations: %v", err) + } + if !mutations.Final || mutations.Pts != deleted.Event.Pts || len(mutations.NewMessages) != 0 || len(mutations.OtherUpdates) != 3 { + t.Fatalf("public preview mutations = %+v, want edit/pin/delete through pts %d", mutations, deleted.Event.Pts) + } + wantTypes := []domain.ChannelUpdateEventType{ + domain.ChannelUpdateEditMessage, + domain.ChannelUpdatePinnedMessages, + domain.ChannelUpdateDeleteMessages, + } + for i, want := range wantTypes { + if mutations.OtherUpdates[i].Type != want { + t.Fatalf("public preview mutation[%d] = %+v, want %s", i, mutations.OtherUpdates[i], want) + } + } + if edited.Event.Pts >= pinned.Event.Pts || pinned.Event.Pts >= deleted.Event.Pts { + t.Fatalf("public preview mutation pts = edit %d pin %d delete %d, want strictly increasing", edited.Event.Pts, pinned.Event.Pts, deleted.Event.Pts) + } } func TestChannelStoreDifferenceUsesDurableMessageSnapshots(t *testing.T) { diff --git a/internal/store/postgres/channel_member_list.go b/internal/store/postgres/channel_member_list.go index 94188cf3..c2c4e280 100644 --- a/internal/store/postgres/channel_member_list.go +++ b/internal/store/postgres/channel_member_list.go @@ -360,3 +360,52 @@ ORDER BY user_id`, channelID, candidates[start:end]) sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) return out, nil } + +func (s *ChannelStore) FilterChannelMessageAudienceIDs(ctx context.Context, channelID int64, userIDs []int64) ([]int64, error) { + if channelID == 0 || len(userIDs) == 0 { + return nil, nil + } + candidates := uniqueChannelUserIDs(userIDs, 0) + if len(candidates) == 0 { + return nil, nil + } + out := make([]int64, 0, len(candidates)) + for start := 0; start < len(candidates); start += channelMemberFilterBatch { + end := start + channelMemberFilterBatch + if end > len(candidates) { + end = len(candidates) + } + rows, err := s.db.Query(ctx, ` +SELECT candidate.user_id +FROM channels c +CROSS JOIN unnest($2::bigint[]) AS candidate(user_id) +LEFT JOIN channel_members m + ON m.channel_id = c.id AND m.user_id = candidate.user_id +WHERE c.id = $1 + AND NOT c.deleted + AND NOT COALESCE((m.banned_rights->>'ViewMessages')::boolean, false) + AND COALESCE(m.status, '') NOT IN ('kicked', 'banned') + AND ( + m.status = 'active' + OR (COALESCE(c.username, '') <> '' AND COALESCE(m.status, 'left') = 'left') + ) +ORDER BY candidate.user_id`, channelID, candidates[start:end]) + if err != nil { + return nil, fmt.Errorf("filter channel message audience: %w", err) + } + for rows.Next() { + var userID int64 + if err := rows.Scan(&userID); err != nil { + rows.Close() + return nil, err + } + out = append(out, userID) + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, err + } + rows.Close() + } + return out, nil +} diff --git a/internal/store/postgres/channel_public_preview_integration_test.go b/internal/store/postgres/channel_public_preview_integration_test.go index 4a7657a5..169dae66 100644 --- a/internal/store/postgres/channel_public_preview_integration_test.go +++ b/internal/store/postgres/channel_public_preview_integration_test.go @@ -75,12 +75,42 @@ func TestPublicChannelAndMegagroupPreviewPostgres(t *testing.T) { if !found || history.Self.Status != domain.ChannelMemberLeft { t.Fatalf("preview history = %+v self=%+v", history.Messages, history.Self) } + audience, err := channels.FilterChannelMessageAudienceIDs(ctx, public.ID, []int64{viewer.ID, owner.ID, viewer.ID}) + if err != nil { + t.Fatalf("filter public message audience: %v", err) + } + if len(audience) != 2 || audience[0] != owner.ID || audience[1] != viewer.ID { + t.Fatalf("public message audience = %v, want owner/member and viewer/subscriber", audience) + } + diff, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: viewer.ID, ChannelID: public.ID, Pts: created.Event.Pts, Limit: 20, + }) + if err != nil { + t.Fatalf("public preview difference: %v", err) + } + if !diff.Final || diff.Pts != sent.Event.Pts || len(diff.NewMessages) != 1 || diff.NewMessages[0].ID != sent.Message.ID { + t.Fatalf("public preview difference = %+v, want sent message through pts %d", diff, sent.Event.Pts) + } if _, err := channels.GetParticipants(ctx, viewer.ID, public.ID, domain.ChannelParticipantsFilter{Kind: domain.ChannelParticipantsRecent}, 0, 20); err != nil { t.Fatalf("public preview participants: %v", err) } if _, err := channels.GetParticipant(ctx, viewer.ID, public.ID, viewer.ID); !errors.Is(err, domain.ErrUserNotParticipant) { t.Fatalf("public preview self participant err = %v, want ErrUserNotParticipant", err) } + dialogs := appdialogs.NewService(nil, channels) + peerDialogs, err := dialogs.GetPeerDialogs(ctx, viewer.ID, []domain.Peer{{Type: domain.PeerTypeChannel, ID: public.ID}}) + if err != nil { + t.Fatalf("public preview peer dialogs: %v", err) + } + if len(peerDialogs.Dialogs) != 1 || len(peerDialogs.ChannelMessages) != 0 || len(peerDialogs.Channels) != 1 { + t.Fatalf("public preview peer dialogs = %+v, want one zero-top bootstrap", peerDialogs) + } + previewDialog := peerDialogs.Dialogs[0] + if previewDialog.TopMessage != 0 || !previewDialog.ChannelLeft || + previewDialog.ReadInboxMaxID != 0 || previewDialog.ReadOutboxMaxID != 0 || + previewDialog.Pts != sent.Event.Pts { + t.Fatalf("public preview bootstrap dialog = %+v", previewDialog) + } var memberExists bool if err := pool.QueryRow(ctx, `SELECT EXISTS ( SELECT 1 FROM channel_members WHERE channel_id = $1 AND user_id = $2 @@ -96,6 +126,28 @@ SELECT 1 FROM channel_members WHERE channel_id = $1 AND user_id = $2 if _, err := channels.LeaveChannel(ctx, public.ID, viewer.ID, 1700009430+i); err != nil { t.Fatalf("leave public peer: %v", err) } + filtered, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: viewer.ID, ChannelID: public.ID, Pts: sent.Event.Pts, Limit: 20, + }) + if err != nil { + t.Fatalf("public difference across participant events: %v", err) + } + if tc.broadcast { + if !filtered.Final || filtered.Pts != sent.Event.Pts || len(filtered.Events) != 0 || + len(filtered.NewMessages) != 0 || len(filtered.OtherUpdates) != 0 { + t.Fatalf("broadcast difference after transient participant changes = %+v, want unchanged PTS", filtered) + } + } else { + if !filtered.Final || filtered.Pts <= sent.Event.Pts || len(filtered.NewMessages) != 2 || + len(filtered.OtherUpdates) != 0 { + t.Fatalf("megagroup join/leave difference = %+v, want two real service messages", filtered) + } + for _, message := range filtered.NewMessages { + if message.Action == nil { + t.Fatalf("megagroup join/leave difference message = %+v, want service action", message) + } + } + } if _, err := channels.GetParticipant(ctx, viewer.ID, public.ID, viewer.ID); !errors.Is(err, domain.ErrUserNotParticipant) { t.Fatalf("left self participant err = %v, want ErrUserNotParticipant", err) } @@ -112,6 +164,9 @@ SELECT 1 FROM channel_members WHERE channel_id = $1 AND user_id = $2 t.Fatalf("create private group: %v", err) } channelIDs = append(channelIDs, private.Channel.ID) + if audience, err := channels.FilterChannelMessageAudienceIDs(ctx, private.Channel.ID, []int64{viewer.ID}); err != nil || len(audience) != 0 { + t.Fatalf("private message audience = %v err %v, want empty", audience, err) + } if _, err := channels.ListChannelHistory(ctx, viewer.ID, domain.ChannelHistoryFilter{ChannelID: private.Channel.ID, Limit: 20}); !errors.Is(err, domain.ErrChannelPrivate) { t.Fatalf("private preview history err = %v, want ErrChannelPrivate", err) } diff --git a/internal/store/postgres/channel_updates.go b/internal/store/postgres/channel_updates.go index 24bef8ae..e18d84b9 100644 --- a/internal/store/postgres/channel_updates.go +++ b/internal/store/postgres/channel_updates.go @@ -27,16 +27,6 @@ func (s *ChannelStore) ListChannelDifference(ctx context.Context, req domain.Cha if limit <= 0 || limit > domain.MaxChannelDifferenceLimit { limit = domain.MaxChannelDifferenceLimit } - if preview && member.Status != domain.ChannelMemberActive { - return domain.ChannelDifference{ - Channel: channel, - Self: member, - Pts: channel.Pts, - Final: true, - Timeout: 30, - Dialog: previewChannelDialog(req.UserID, channel, member), - }, nil - } checkpoint, err := getChannelUpdateCheckpoint(ctx, s.db, req.ChannelID) if err != nil { return domain.ChannelDifference{}, err