diff --git a/deploy/migrations/0149_channel_live_pinned_index.down.sql b/deploy/migrations/0149_channel_live_pinned_index.down.sql new file mode 100644 index 00000000..17ee0796 --- /dev/null +++ b/deploy/migrations/0149_channel_live_pinned_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS public.channel_messages_live_pinned_idx; diff --git a/deploy/migrations/0149_channel_live_pinned_index.up.sql b/deploy/migrations/0149_channel_live_pinned_index.up.sql new file mode 100644 index 00000000..cb53a292 --- /dev/null +++ b/deploy/migrations/0149_channel_live_pinned_index.up.sql @@ -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; diff --git a/internal/domain/channel.go b/internal/domain/channel.go index a43c2971..3f8bedd2 100644 --- a/internal/domain/channel.go +++ b/internal/domain/channel.go @@ -2175,15 +2175,22 @@ type ChannelHistoryFilter struct { SenderUserID int64 PinnedOnly bool MusicOnly bool - OffsetID int - OffsetDate int - AddOffset int - Limit int - MinDate int - MaxDate int - MaxID int - MinID int - Hash int64 + // NeedTotalCount requests the exact number of messages matching the static + // filters before offset/add_offset pagination. Ordinary history pages leave + // this false and keep the bounded len(page)+has-more hint. + NeedTotalCount bool + // CountOnly skips message hydration and returns only the exact Count plus + // the viewer-scoped channel metadata needed for access validation. + CountOnly bool + OffsetID int + OffsetDate int + AddOffset int + Limit int + MinDate int + MaxDate int + MaxID int + MinID int + Hash int64 } // ChannelSearchPostsRequest describes a bounded global public post search. diff --git a/internal/rpc/channels_multi_pin_android_rpc_test.go b/internal/rpc/channels_multi_pin_android_rpc_test.go index 4ec6efa4..e36f7071 100644 --- a/internal/rpc/channels_multi_pin_android_rpc_test.go +++ b/internal/rpc/channels_multi_pin_android_rpc_test.go @@ -65,8 +65,8 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) { } ids = append(ids, sent.Message.ID) } - // 三条置顶：早期、中间、最新（Android 置顶栏循环跳转需要全部三条都可跳）。 - pins := []int{ids[4], ids[14], ids[27]} + // 五条置顶：覆盖用户反馈中的真实规模；Android 置顶栏循环跳转需要全部可跳。 + pins := []int{ids[4], ids[9], ids[14], ids[20], ids[27]} for _, id := range pins { if _, err := channelSvc.UpdatePinnedMessage(ctx, owner.ID, domain.UpdateChannelPinnedMessageRequest{ ChannelID: channelID, @@ -113,7 +113,10 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) { if 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 { msg, ok := raw.(*tg.Message) 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 histEnc := dispatch(&tg.MessagesGetHistoryRequest{ 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]) } - // ③ 本地缺对象补拉：MessagesStorage.loadChatInfo → channels.getMessages。 + // ④ 本地缺对象补拉：MessagesStorage.loadChatInfo → channels.getMessages。 // DrKLO 发的是 pre-InputMessage 构造器 #93d7b347（id:Vector）， // 该请求 500 会让客户端把这批 pin 按「已取消置顶」从本地缓存删除。 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{ Channel: &tg.InputChannel{ChannelID: channelID, AccessHash: memberView.Channel.AccessHash}, }) @@ -220,7 +265,7 @@ func TestChannelMultiPinAndroidOpenAndJump(t *testing.T) { if !ok { t.Fatalf("full chat = %T, want channelFull", full.FullChat) } - if pinnedID, _ := channelFull.GetPinnedMsgID(); pinnedID != pins[2] { - t.Fatalf("channelFull pinned_msg_id = %d, want latest pin %d", 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[len(pins)-1]) } } diff --git a/internal/rpc/messages_history.go b/internal/rpc/messages_history.go index eb5234e4..e1bffd3c 100644 --- a/internal/rpc/messages_history.go +++ b/internal/rpc/messages_history.go @@ -197,35 +197,50 @@ func (r *Router) onMessagesGetSearchCounters(ctx context.Context, req *tg.Messag if err != nil { 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 mediaCounts := domain.MediaCategoryCounts{} - if peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil { - // 只读 PinnedMessageID（Channel 字段）：走轻量 ResolveChannel，省 dialog/读态/boost 查询。 - view, err := r.deps.Channels.ResolveChannel(ctx, userID, peer.ID) - if err != nil { - return nil, err + if needsPinned { + switch { + case peer.Type == domain.PeerTypeChannel && r.deps.Channels != nil: + history, err := r.deps.Channels.GetHistory(ctx, userID, domain.ChannelHistoryFilter{ + 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 { - 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 needsMedia { counts, err := r.mediaCountsForPeer(ctx, userID, peer) if err != nil { 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) { limit := req.Limit + countOnly := limit == 0 if limit <= 0 || limit > 100 { limit = 100 } @@ -836,14 +852,17 @@ func (r *Router) channelHistoryFilterFromSearchRequest(userID int64, req *tg.Mes Query: req.Q, PinnedOnly: messagesSearchFilterPinned(req.Filter), MusicOnly: messagesSearchFilterMusic(req.Filter), - OffsetID: req.OffsetID, - AddOffset: domain.ClampMessageHistoryAddOffset(req.AddOffset), - Limit: limit, - MinDate: req.MinDate, - MaxDate: req.MaxDate, - MaxID: req.MaxID, - MinID: req.MinID, - Hash: req.Hash, + NeedTotalCount: countOnly || + (req.OffsetID == 0 && req.MinDate == 0 && req.MaxDate == 0 && req.AddOffset >= 0 && req.Hash == 0), + CountOnly: countOnly, + OffsetID: req.OffsetID, + AddOffset: domain.ClampMessageHistoryAddOffset(req.AddOffset), + Limit: limit, + MinDate: req.MinDate, + MaxDate: req.MaxDate, + MaxID: req.MaxID, + MinID: req.MinID, + Hash: req.Hash, } if req.FromID != nil { from, ok := r.domainPeerFromInputPeer(userID, req.FromID) diff --git a/internal/store/memory/channel_message_history.go b/internal/store/memory/channel_message_history.go index d62bfd7c..5a29e800 100644 --- a/internal/store/memory/channel_message_history.go +++ b/internal/store/memory/channel_message_history.go @@ -66,6 +66,20 @@ func (s *ChannelStore) ListChannelHistory(_ context.Context, viewerUserID int64, } 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）： // >= 0 backward：锚点更旧方向（不含锚点），先跳过 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 { count = len(out) + 1 } + if filter.NeedTotalCount { + count = len(matched) + } s.populateChannelMessageRepliesLocked(viewerUserID, filter.ChannelID, 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{ Channel: channel, Self: member, diff --git a/internal/store/postgres/channel_message_history.go b/internal/store/postgres/channel_message_history.go index 61971770..f5f80a97 100644 --- a/internal/store/postgres/channel_message_history.go +++ b/internal/store/postgres/channel_message_history.go @@ -79,6 +79,24 @@ AND EXISTS ( baseArgs = append(baseArgs, filter.MinID) 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) { rows, err := s.db.Query(ctx, sql, queryArgs...) if err != nil { @@ -102,7 +120,6 @@ AND EXISTS ( // store 层二次钳制 add_offset 到 [-100,100]（与私聊 ListMessagesByUser 对齐）： // 即便某个 caller 漏在 RPC 层钳制，也不会把客户端巨大值变成大 SQL OFFSET 跳扫。 addOffset := domain.ClampMessageHistoryAddOffset(filter.AddOffset) - out := domain.ChannelHistory{Channel: channel, Self: member, Channels: extraChannels} hasMoreOlder := false // 锚点条件：offset_date 优先按日期、否则按消息 id（对齐私聊）； // 二者皆空时向更新方向退化为空、向更旧方向退化为全部（取最新）。 @@ -199,9 +216,11 @@ AND EXISTS ( } out.Messages = older } - out.Count = len(out.Messages) - if hasMoreOlder { - out.Count = len(out.Messages) + 1 + if !needExactTotal { + out.Count = len(out.Messages) + if hasMoreOlder { + out.Count = len(out.Messages) + 1 + } } if err := s.populateChannelMessageReplies(ctx, s.db, viewerUserID, channel, out.Messages); err != nil { return domain.ChannelHistory{}, err diff --git a/internal/store/postgres/channel_multi_pin_integration_test.go b/internal/store/postgres/channel_multi_pin_integration_test.go index 2edee1e3..a6dfd7ca 100644 --- a/internal/store/postgres/channel_multi_pin_integration_test.go +++ b/internal/store/postgres/channel_multi_pin_integration_test.go @@ -3,6 +3,7 @@ package postgres import ( "context" "errors" + "strings" "testing" "telesrv/internal/domain" @@ -98,6 +99,70 @@ func TestChannelMultiPin(t *testing.T) { 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 标志（多置顶都标，不只最新）。 page, err := channels.ListChannelHistory(ctx, member.ID, domain.ChannelHistoryFilter{ChannelID: channelID, Limit: 50}) if err != nil { diff --git a/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go b/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go index 9436fae2..0114de3b 100644 --- a/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go +++ b/internal/store/postgres/star_gift_lifecycle_migration_integration_test.go @@ -14,7 +14,7 @@ func TestStarGiftLifecycleMigrationsApply(t *testing.T) { if err != nil { t.Fatalf("migrate star gift lifecycle schema: %v", err) } - if status.Dirty || status.Empty || status.Version != 148 { - t.Fatalf("migration status = %+v, want clean version 148", status) + if status.Dirty || status.Empty || status.Version != 149 { + t.Fatalf("migration status = %+v, want clean version 149", status) } }