fix(messages): sync exact channel pinned counts

This commit is contained in:
iamxvbaba 2026-07-27 00:17:21 +08:00
parent b742450938
commit e75fd04a28
9 changed files with 232 additions and 62 deletions

View file

@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.channel_messages_live_pinned_idx;

View file

@ -0,0 +1,3 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS channel_messages_live_pinned_idx
ON public.channel_messages (channel_id, id DESC)
WHERE pinned AND NOT deleted;

View file

@ -2175,15 +2175,22 @@ type ChannelHistoryFilter struct {
SenderUserID int64 SenderUserID int64
PinnedOnly bool PinnedOnly bool
MusicOnly bool MusicOnly bool
OffsetID int // NeedTotalCount requests the exact number of messages matching the static
OffsetDate int // filters before offset/add_offset pagination. Ordinary history pages leave
AddOffset int // this false and keep the bounded len(page)+has-more hint.
Limit int NeedTotalCount bool
MinDate int // CountOnly skips message hydration and returns only the exact Count plus
MaxDate int // the viewer-scoped channel metadata needed for access validation.
MaxID int CountOnly bool
MinID int OffsetID int
Hash int64 OffsetDate int
AddOffset int
Limit int
MinDate int
MaxDate int
MaxID int
MinID int
Hash int64
} }
// ChannelSearchPostsRequest describes a bounded global public post search. // ChannelSearchPostsRequest describes a bounded global public post search.

View file

@ -65,8 +65,8 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
} }
ids = append(ids, sent.Message.ID) ids = append(ids, sent.Message.ID)
} }
// 三条置顶:早期、中间、最新(Android 置顶栏循环跳转需要全部三条都可跳)。 // 五条置顶:覆盖用户反馈中的真实规模;Android 置顶栏循环跳转需要全部可跳。
pins := []int{ids[4], ids[14], ids[27]} pins := []int{ids[4], ids[9], ids[14], ids[20], ids[27]}
for _, id := range pins { for _, id := range pins {
if _, err := channelSvc.UpdatePinnedMessage(ctx, owner.ID, domain.UpdateChannelPinnedMessageRequest{ if _, err := channelSvc.UpdatePinnedMessage(ctx, owner.ID, domain.UpdateChannelPinnedMessageRequest{
ChannelID: channelID, ChannelID: channelID,
@ -113,7 +113,10 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
if channelMessages.Count != len(pins) { if channelMessages.Count != len(pins) {
t.Fatalf("pinned search count = %d, want %d", channelMessages.Count, len(pins)) t.Fatalf("pinned search count = %d, want %d", channelMessages.Count, len(pins))
} }
wantDesc := []int{pins[2], pins[1], pins[0]} wantDesc := make([]int, len(pins))
for i := range pins {
wantDesc[i] = pins[len(pins)-1-i]
}
for i, raw := range channelMessages.Messages { for i, raw := range channelMessages.Messages {
msg, ok := raw.(*tg.Message) msg, ok := raw.(*tg.Message)
if !ok { if !ok {
@ -128,7 +131,49 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
} }
} }
// ② 点置顶栏跳最旧 pin:scrollToMessageId → getHistory AROUND(手机 count=20)。 // ② tweb 冷加载:先用 limit=1 取最新 pin,并把 messages.channelMessages.count
// 当作完整置顶数;旧实现把 count 错算成 len(page)+hasMore,即五条只报两条。
twebSearchEnc := dispatch(&tg.MessagesSearchRequest{
Peer: peer,
Q: "",
Filter: &tg.InputMessagesFilterPinned{},
Limit: 1,
})
twebSearch, ok := twebSearchEnc.(*tg.MessagesChannelMessages)
if !ok {
t.Fatalf("tweb pinned search response = %T, want messages.channelMessages", twebSearchEnc)
}
if len(twebSearch.Messages) != 1 || twebSearch.Count != len(pins) {
t.Fatalf("tweb pinned search messages/count = %d/%d, want 1/%d", len(twebSearch.Messages), twebSearch.Count, len(pins))
}
// limit=0 是官方 count-only 入口:不得为了计数反序列化/返回消息页。
countOnlyEnc := dispatch(&tg.MessagesSearchRequest{
Peer: peer,
Q: "",
Filter: &tg.InputMessagesFilterPinned{},
Limit: 0,
})
countOnly, ok := countOnlyEnc.(*tg.MessagesChannelMessages)
if !ok {
t.Fatalf("count-only pinned search response = %T, want messages.channelMessages", countOnlyEnc)
}
if len(countOnly.Messages) != 0 || countOnly.Count != len(pins) {
t.Fatalf("count-only pinned search messages/count = %d/%d, want 0/%d", len(countOnly.Messages), countOnly.Count, len(pins))
}
counters, err := r.onMessagesGetSearchCounters(WithUserID(androidClientContext(), member.ID), &tg.MessagesGetSearchCountersRequest{
Peer: peer,
Filters: []tg.MessagesFilterClass{&tg.InputMessagesFilterPinned{}},
})
if err != nil {
t.Fatalf("messages.getSearchCounters(filterPinned): %v", err)
}
if len(counters) != 1 || counters[0].Count != len(pins) {
t.Fatalf("pinned search counters = %+v, want count %d", counters, len(pins))
}
// ③ 点置顶栏跳最旧 pin:scrollToMessageId → getHistory AROUND(手机 count=20)。
const aroundCount = 20 const aroundCount = 20
histEnc := dispatch(&tg.MessagesGetHistoryRequest{ histEnc := dispatch(&tg.MessagesGetHistoryRequest{
Peer: peer, Peer: peer,
@ -160,7 +205,7 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
t.Fatalf("around history lacks anchor %d: jump shows MessageNotFound on Android", pins[0]) t.Fatalf("around history lacks anchor %d: jump shows MessageNotFound on Android", pins[0])
} }
// ③ 本地缺对象补拉:MessagesStorage.loadChatInfo → channels.getMessages。 // ④ 本地缺对象补拉:MessagesStorage.loadChatInfo → channels.getMessages。
// DrKLO 发的是 pre-InputMessage 构造器 #93d7b347(id:Vector<int>), // DrKLO 发的是 pre-InputMessage 构造器 #93d7b347(id:Vector<int>),
// 该请求 500 会让客户端把这批 pin 按「已取消置顶」从本地缓存删除。 // 该请求 500 会让客户端把这批 pin 按「已取消置顶」从本地缓存删除。
var legacy bin.Buffer var legacy bin.Buffer
@ -208,7 +253,7 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
} }
} }
// ④ chatFull 降级缓存:pinned_msg_id 必须是最新置顶(Android 以它判断是否重拉列表)。 // ⑤ chatFull 降级缓存:pinned_msg_id 必须是最新置顶(Android 以它判断是否重拉列表)。
fullEnc := dispatch(&tg.ChannelsGetFullChannelRequest{ fullEnc := dispatch(&tg.ChannelsGetFullChannelRequest{
Channel: &tg.InputChannel{ChannelID: channelID, AccessHash: memberView.Channel.AccessHash}, Channel: &tg.InputChannel{ChannelID: channelID, AccessHash: memberView.Channel.AccessHash},
}) })
@ -220,7 +265,7 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) {
if !ok { if !ok {
t.Fatalf("full chat = %T, want channelFull", full.FullChat) t.Fatalf("full chat = %T, want channelFull", full.FullChat)
} }
if pinnedID, _ := channelFull.GetPinnedMsgID(); pinnedID != pins[2] { if pinnedID, _ := channelFull.GetPinnedMsgID(); pinnedID != pins[len(pins)-1] {
t.Fatalf("channelFull pinned_msg_id = %d, want latest pin %d", pinnedID, pins[2]) t.Fatalf("channelFull pinned_msg_id = %d, want latest pin %d", pinnedID, pins[len(pins)-1])
} }
} }

View file

@ -197,35 +197,50 @@ func (r *Router) onMessagesGetSearchCounters(ctx context.Context, req *tg.Messag
if err != nil { if err != nil {
return nil, err return nil, err
} }
needsPinned := false
needsMedia := false
for _, filter := range req.Filters {
if filter == nil {
continue
}
if _, ok := filter.(*tg.InputMessagesFilterPinned); ok {
needsPinned = true
continue
}
if len(mediaCategoriesForFilter(filter)) > 0 {
needsMedia = true
}
}
pinnedCount := 0 pinnedCount := 0
mediaCounts := domain.MediaCategoryCounts{} mediaCounts := domain.MediaCategoryCounts{}
if peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil { if needsPinned {
// 只读 PinnedMessageID(Channel 字段):走轻量 ResolveChannel,省 dialog/读态/boost 查询。 switch {
view, err := r.deps.Channels.ResolveChannel(ctx, userID, peer.ID) case peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil:
if err != nil { history, err := r.deps.Channels.GetHistory(ctx, userID, domain.ChannelHistoryFilter{
return nil, err ChannelID: peer.ID,
PinnedOnly: true,
NeedTotalCount: true,
CountOnly: true,
})
if err != nil {
return nil, err
}
pinnedCount = history.Count
case peer.Type == domain.PeerTypeUser && r.deps.Messages != nil:
list, err := r.deps.Messages.Search(ctx, userID, domain.MessageFilter{
HasPeer: true,
Peer: peer,
PinnedOnly: true,
Limit: 1,
NeedTotalCount: true,
})
if err != nil {
return nil, err
}
pinnedCount = list.Count
} }
if view.Channel.PinnedMessageID > 0 {
pinnedCount = 1
}
counts, err := r.mediaCountsForPeer(ctx, userID, peer)
if err != nil {
return nil, err
}
mediaCounts = counts
} }
if peer.Type == domain.PeerTypeUser && r.deps.Messages != nil { if needsMedia {
list, err := r.deps.Messages.Search(ctx, userID, domain.MessageFilter{
HasPeer: true,
Peer: peer,
PinnedOnly: true,
Limit: 1,
NeedTotalCount: true,
})
if err != nil {
return nil, err
}
pinnedCount = list.Count
counts, err := r.mediaCountsForPeer(ctx, userID, peer) counts, err := r.mediaCountsForPeer(ctx, userID, peer)
if err != nil { if err != nil {
return nil, err return nil, err
@ -828,6 +843,7 @@ func (r *Router) messageFilterFromSearchRequest(ctx context.Context, userID int6
func (r *Router) channelHistoryFilterFromSearchRequest(userID int64, req *tg.MessagesSearchRequest, channelID int64) (domain.ChannelHistoryFilter, bool) { func (r *Router) channelHistoryFilterFromSearchRequest(userID int64, req *tg.MessagesSearchRequest, channelID int64) (domain.ChannelHistoryFilter, bool) {
limit := req.Limit limit := req.Limit
countOnly := limit == 0
if limit <= 0 || limit > 100 { if limit <= 0 || limit > 100 {
limit = 100 limit = 100
} }
@ -836,14 +852,17 @@ func (r *Router) channelHistoryFilterFromSearchRequest(userID int64, req *tg.Mes
Query: req.Q, Query: req.Q,
PinnedOnly: messagesSearchFilterPinned(req.Filter), PinnedOnly: messagesSearchFilterPinned(req.Filter),
MusicOnly: messagesSearchFilterMusic(req.Filter), MusicOnly: messagesSearchFilterMusic(req.Filter),
OffsetID: req.OffsetID, NeedTotalCount: countOnly ||
AddOffset: domain.ClampMessageHistoryAddOffset(req.AddOffset), (req.OffsetID == 0 && req.MinDate == 0 && req.MaxDate == 0 && req.AddOffset >= 0 && req.Hash == 0),
Limit: limit, CountOnly: countOnly,
MinDate: req.MinDate, OffsetID: req.OffsetID,
MaxDate: req.MaxDate, AddOffset: domain.ClampMessageHistoryAddOffset(req.AddOffset),
MaxID: req.MaxID, Limit: limit,
MinID: req.MinID, MinDate: req.MinDate,
Hash: req.Hash, MaxDate: req.MaxDate,
MaxID: req.MaxID,
MinID: req.MinID,
Hash: req.Hash,
} }
if req.FromID != nil { if req.FromID != nil {
from, ok := r.domainPeerFromInputPeer(userID, req.FromID) from, ok := r.domainPeerFromInputPeer(userID, req.FromID)

View file

@ -66,6 +66,20 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64,
} }
matched = append(matched, msg) matched = append(matched, msg)
} }
extraChannels := []domain.Channel(nil)
if channel.Monoforum && channel.LinkedMonoforumID != 0 {
if parent, ok := s.channels[channel.LinkedMonoforumID]; ok && !parent.Deleted {
extraChannels = append(extraChannels, cloneChannel(parent))
}
}
if filter.CountOnly {
return domain.ChannelHistory{
Channel: channel,
Self: member,
Channels: extraChannels,
Count: len(matched),
}, nil
}
// add_offset 决定加载方向(对齐 postgres ListChannelHistory): // add_offset 决定加载方向(对齐 postgres ListChannelHistory):
// >= 0 backward:锚点更旧方向(不含锚点),先跳过 add_offset 条 // >= 0 backward:锚点更旧方向(不含锚点),先跳过 add_offset 条
// < 0 且 +limit>0 around:以锚点为中心,向更新取 -add_offset 条 + 向更旧(含锚点)取 limit+add_offset 条 // < 0 且 +limit>0 around:以锚点为中心,向更新取 -add_offset 条 + 向更旧(含锚点)取 limit+add_offset 条
@ -160,14 +174,11 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64,
if hasMoreOlder { if hasMoreOlder {
count = len(out) + 1 count = len(out) + 1
} }
if filter.NeedTotalCount {
count = len(matched)
}
s.populateChannelMessageRepliesLocked(viewerUserID, filter.ChannelID, out) s.populateChannelMessageRepliesLocked(viewerUserID, filter.ChannelID, out)
s.populateChannelMessageReactionsLocked(viewerUserID, channel, out) s.populateChannelMessageReactionsLocked(viewerUserID, channel, out)
extraChannels := []domain.Channel(nil)
if channel.Monoforum && channel.LinkedMonoforumID != 0 {
if parent, ok := s.channels[channel.LinkedMonoforumID]; ok && !parent.Deleted {
extraChannels = append(extraChannels, cloneChannel(parent))
}
}
return domain.ChannelHistory{ return domain.ChannelHistory{
Channel: channel, Channel: channel,
Self: member, Self: member,

View file

@ -79,6 +79,24 @@ AND EXISTS (
baseArgs = append(baseArgs, filter.MinID) baseArgs = append(baseArgs, filter.MinID)
base += fmt.Sprintf(" AND id > $%d", len(baseArgs)) base += fmt.Sprintf(" AND id > $%d", len(baseArgs))
} }
out := domain.ChannelHistory{Channel: channel, Self: member, Channels: extraChannels}
needExactTotal := filter.NeedTotalCount || filter.CountOnly
exactTotal := 0
if needExactTotal {
// Exact totals are opt-in. messages.search first/count-only pages and
// messages.getSearchCounters need protocol-exact Count, while ordinary
// getHistory pages must stay on the single bounded page query.
if err := s.db.QueryRow(ctx,
"SELECT count(*)::int FROM channel_messages WHERE "+base,
baseArgs...,
).Scan(&exactTotal); err != nil {
return domain.ChannelHistory{}, fmt.Errorf("count channel history: %w", err)
}
out.Count = exactTotal
}
if filter.CountOnly {
return out, nil
}
scanList := func(sql string, queryArgs []any) ([]domain.ChannelMessage, error) { scanList := func(sql string, queryArgs []any) ([]domain.ChannelMessage, error) {
rows, err := s.db.Query(ctx, sql, queryArgs...) rows, err := s.db.Query(ctx, sql, queryArgs...)
if err != nil { if err != nil {
@ -102,7 +120,6 @@ AND EXISTS (
// store 层二次钳制 add_offset 到 [-100,100](与私聊 ListMessagesByUser 对齐): // store 层二次钳制 add_offset 到 [-100,100](与私聊 ListMessagesByUser 对齐):
// 即便某个 caller 漏在 RPC 层钳制,也不会把客户端巨大值变成大 SQL OFFSET 跳扫。 // 即便某个 caller 漏在 RPC 层钳制,也不会把客户端巨大值变成大 SQL OFFSET 跳扫。
addOffset := domain.ClampMessageHistoryAddOffset(filter.AddOffset) addOffset := domain.ClampMessageHistoryAddOffset(filter.AddOffset)
out := domain.ChannelHistory{Channel: channel, Self: member, Channels: extraChannels}
hasMoreOlder := false hasMoreOlder := false
// 锚点条件:offset_date 优先按日期、否则按消息 id(对齐私聊); // 锚点条件:offset_date 优先按日期、否则按消息 id(对齐私聊);
// 二者皆空时向更新方向退化为空、向更旧方向退化为全部(取最新)。 // 二者皆空时向更新方向退化为空、向更旧方向退化为全部(取最新)。
@ -199,9 +216,11 @@ AND EXISTS (
} }
out.Messages = older out.Messages = older
} }
out.Count = len(out.Messages) if !needExactTotal {
if hasMoreOlder { out.Count = len(out.Messages)
out.Count = len(out.Messages) + 1 if hasMoreOlder {
out.Count = len(out.Messages) + 1
}
} }
if err := s.populateChannelMessageReplies(ctx, s.db, viewerUserID, channel, out.Messages); err != nil { if err := s.populateChannelMessageReplies(ctx, s.db, viewerUserID, channel, out.Messages); err != nil {
return domain.ChannelHistory{}, err return domain.ChannelHistory{}, err

View file

@ -3,6 +3,7 @@ package postgres
import ( import (
"context" "context"
"errors" "errors"
"strings"
"testing" "testing"
"telesrv/internal/domain" "telesrv/internal/domain"
@ -98,6 +99,70 @@ func TestChannelMultiPin(t *testing.T) {
t.Fatalf("filterPinned message %d lacks pinned flag", msg.ID) t.Fatalf("filterPinned message %d lacks pinned flag", msg.ID)
} }
} }
// 普通历史热路径保留 bounded has-more hint,不为每页额外 COUNT。
hint, err := channels.ListChannelHistory(ctx, owner.ID, domain.ChannelHistoryFilter{
ChannelID: channelID, PinnedOnly: true, Limit: 1,
})
if err != nil {
t.Fatalf("filterPinned hint page: %v", err)
}
if len(hint.Messages) != 1 || hint.Count != 2 {
t.Fatalf("filterPinned hint messages/count = %d/%d, want 1/2", len(hint.Messages), hint.Count)
}
// messages.search 首屏显式请求精确总数;页大小不能污染 Count。
exact, err := channels.ListChannelHistory(ctx, owner.ID, domain.ChannelHistoryFilter{
ChannelID: channelID, PinnedOnly: true, Limit: 1, NeedTotalCount: true,
})
if err != nil {
t.Fatalf("filterPinned exact page: %v", err)
}
if len(exact.Messages) != 1 || exact.Count != 3 {
t.Fatalf("filterPinned exact messages/count = %d/%d, want 1/3", len(exact.Messages), exact.Count)
}
// messages.getSearchCounters/limit=0 只计数,不加载消息及 reply/reaction companion。
countOnly, err := channels.ListChannelHistory(ctx, owner.ID, domain.ChannelHistoryFilter{
ChannelID: channelID, PinnedOnly: true, NeedTotalCount: true, CountOnly: true,
})
if err != nil {
t.Fatalf("filterPinned count-only: %v", err)
}
if len(countOnly.Messages) != 0 || countOnly.Count != 3 {
t.Fatalf("filterPinned count-only messages/count = %d/%d, want 0/3", len(countOnly.Messages), countOnly.Count)
}
// 精确计数只扫描该频道的 pinned 部分索引,不能退化为全频道消息扫描。
tx, err := pool.Begin(ctx)
if err != nil {
t.Fatalf("begin pinned count explain: %v", err)
}
defer tx.Rollback(ctx)
if _, err := tx.Exec(ctx, "SET LOCAL enable_seqscan = off"); err != nil {
t.Fatalf("disable seqscan for pinned count explain: %v", err)
}
rows, err := tx.Query(ctx, `EXPLAIN (COSTS OFF)
SELECT count(*)::int
FROM channel_messages
WHERE channel_id = $1 AND NOT deleted AND pinned`, channelID)
if err != nil {
t.Fatalf("explain pinned count: %v", err)
}
var plan strings.Builder
for rows.Next() {
var line string
if err := rows.Scan(&line); err != nil {
rows.Close()
t.Fatalf("scan pinned count plan: %v", err)
}
plan.WriteString(line)
plan.WriteByte('\n')
}
if err := rows.Err(); err != nil {
rows.Close()
t.Fatalf("read pinned count plan: %v", err)
}
rows.Close()
if !strings.Contains(plan.String(), "channel_messages_live_pinned_idx") {
t.Fatalf("pinned count plan misses partial index:\n%s", plan.String())
}
// 普通历史页的消息行直接携带 pinned 标志(多置顶都标,不只最新)。 // 普通历史页的消息行直接携带 pinned 标志(多置顶都标,不只最新)。
page, err := channels.ListChannelHistory(ctx, member.ID, domain.ChannelHistoryFilter{ChannelID: channelID, Limit: 50}) page, err := channels.ListChannelHistory(ctx, member.ID, domain.ChannelHistoryFilter{ChannelID: channelID, Limit: 50})
if err != nil { if err != nil {

View file

@ -14,7 +14,7 @@ func TestStarGiftLifecycleMigrationsApply(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("migrate star gift lifecycle schema: %v", err) t.Fatalf("migrate star gift lifecycle schema: %v", err)
} }
if status.Dirty || status.Empty || status.Version != 148 { if status.Dirty || status.Empty || status.Version != 149 {
t.Fatalf("migration status = %+v, want clean version 148", status) t.Fatalf("migration status = %+v, want clean version 149", status)
} }
} }