astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: sync public channel preview updates

Browse source
This commit is contained in:
iamxvbaba 2026-07-24 14:50:17 +08:00
parent 2289a31f46
commit b4aaf57d6b
27 changed files with 877 additions and 128 deletions
Download patch file Download diff file Expand all files Collapse all files

94
internal/rpc/channel_fanout_dispatcher.go
View file

@ -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
}

21
internal/rpc/channel_interest.go
View file

@ -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

3
internal/rpc/channels_constants.go
View file

@ -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
)

2
internal/rpc/channels_legacy_chat.go
View file

@ -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)

10
internal/rpc/channels_messages.go
View file

@ -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

89
internal/rpc/channels_public_preview_rpc_test.go
View file

@ -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 {

32
internal/rpc/channels_stubs.go
View file

@ -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
}

2
internal/rpc/channels_topics.go
View file

@ -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

6
internal/rpc/channels_updates.go
View file

@ -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 {

17
internal/rpc/deps.go
View file

@ -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 {

2
internal/rpc/expiry_dispatcher.go
View file

@ -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)
})
}

2
internal/rpc/messages_delete.go
View file

@ -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

2
internal/rpc/messages_forum.go
View file

@ -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)},

70
internal/rpc/rpc_testkit_sessions_test.go
View file

@ -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) {