From 6fd690a06ed569a61546cd2131e53b8ce20c65ff Mon Sep 17 00:00:00 2001 From: A Date: Sun, 7 Jun 2026 20:57:31 +0800 Subject: [PATCH] fix: keep participant changes out of channel pts (cherry picked from commit 07b2497664bd108dec84f6cfe43715540faf2688) --- docs/channel-module.md | 18 +++--- docs/compatibility-matrix.md | 10 +-- docs/persistence-layer.md | 6 +- internal/app/channels/service_test.go | 38 +++++++---- internal/rpc/channels.go | 18 ++---- internal/rpc/router_test.go | 25 ++++---- internal/store/memory/channel.go | 37 +++++------ internal/store/memory/channel_test.go | 63 +++++++++++++++++++ internal/store/postgres/channel.go | 35 ++--------- .../postgres/channel_integration_test.go | 49 ++++++++++++--- 10 files changed, 184 insertions(+), 115 deletions(-) diff --git a/docs/channel-module.md b/docs/channel-module.md index b2f2096e..0bef22c5 100644 --- a/docs/channel-module.md +++ b/docs/channel-module.md @@ -92,7 +92,7 @@ gotd 已提供所有需要的 Layer 225 类型和 dispatcher： - `UpdatesGetChannelDifferenceRequest` 返回 `UpdatesChannelDifferenceClass`，请求 limit 对普通用户建议 10-100，服务端必须 cap。 - `UpdateNewChannelMessage` 携带 `message/pts/pts_count`。 - `UpdateEditChannelMessage`、`UpdateDeleteChannelMessages`、`UpdatePinnedChannelMessages` 都走 channel pts；delete 的 `pts_count` 必须等于本次删除 id 数，pin/edit 为 1。 -- `UpdateChannelParticipant` 携带 actor、prev/new participant，适合 editAdmin/editBanned 的在线更新。 +- `UpdateChannelParticipant` 携带 actor、prev/new participant，适合 editAdmin/editBanned 的在线瞬时更新；Layer 225 该 update 不带 `pts/pts_count`，不能作为 channel pts durable event。 - `ChatInviteExported`、`ChatInvite`、`ChatInviteAlready` 覆盖邀请链接导出、预览和已加入状态。 - `Channel` 和 `ChannelFull` 有 TDesktop 最小必需字段：`AccessHash`、`Broadcast`、`Megagroup`、`HasLink`、`ParticipantsCount`、`AdminRights`、`BannedRights`、`DefaultBannedRights`、`LinkedChatID`、`ReadInboxMaxID`、`ReadOutboxMaxID`、`UnreadCount`、`NotifySettings`、`ExportedInvite`、`Pts`。 - `channels.getGroupsForDiscussion#f5dad378` 无入参，返回 `messages.Chats`；`channels.setDiscussionGroup#40582bb2` 接 `broadcast:InputChannel group:InputChannel`，返回 Bool，并显式定义 `LINK_NOT_MODIFIED/BROADCAST_ID_INVALID/MEGAGROUP_ID_INVALID/MEGAGROUP_PREHISTORY_HIDDEN` 等错误。`channelAdminLogEventActionChangeLinkedChat` 可记录管理日志。 @@ -128,7 +128,7 @@ username 与管理项方面，参考实现 的 `channels.checkUsername` 只校 参考实现 的 send/forward 请求转换器保留 Layer225 `InputReplyTo`，message mapper 再把业务 `ReplyTo/InputReplyTo` 转回 `MessageReplyHeader`；它还用 `MessageForwardedEvent/MessageReplyUpdatedEvent` 维护原消息回复统计。参考实现 的 message/dialog app service 会把 `FwdHeader.FromId/SavedFromPeer/SavedFromId` 放进额外 peer 集合，确保客户端 apply update 前能解析 forward 来源。telesrv 保留 `reply_to_msg_id/top_id/quote` 与 forward header 的客户端可见语义，在响应、durable difference、outbox 投递里补齐可解析的 user/channel peer 上下文，并用 `reply_to_top_id` + linked discussion root 维护首版 channel replies 统计/已读。 -参考实现 的 admin/ban 流程先校验 `add_admins/ban_users`，再通过 saga 改成员 read model 并把 `updateChannelParticipant` 放入可恢复的 updates 流；TDesktop 对 `updateChannelParticipant` 本身不走 channel pts 检查，但会在 `updateChannelTooLong(pts)` 后主动拉 `updates.getChannelDifference`。telesrv 因此把 editAdmin/editBanned 写入 `channel_update_events(channel_participant)`，payload 保存 prev/new participant，在线响应额外携带 `updateChannelTooLong(pts)` 触发客户端补齐 channel pts。 +参考实现 的 admin/ban 流程先校验 `add_admins/ban_users`，再更新成员 read model，并通过 `updateChannel`/`updateChannelParticipant` 让在线客户端刷新状态；TDesktop 对 `updateChannelParticipant` 本身不走 channel pts 检查，且基线 `applyUpdateNoPtsCheck()` 不处理它。telesrv 因此不把 editAdmin/editBanned 写入 `channel_update_events`，也不为纯权限/封禁状态变化分配 channel pts；在线响应/推送只携带 `updateChannelParticipant + updateChannel`，离线客户端通过 full channel、participants 列表或后续可见消息的 channel pts 路径恢复状态。若操作产生可见 service message（例如加人/踢人消息），那条 service message 作为 `updateNewChannelMessage` 单独占 channel pts。 ### 参考实现 B @@ -148,7 +148,7 @@ username 与管理项方面，参考实现 的 `channels.checkUsername` 只校 参考实现 的 channel outbox 在写入 reply 消息前会按 `channel_id + reply_to_msg_id` 读取被回复消息：如果目标消息已有 `reply_to_top_id` 就继承，否则把当前 `reply_to_msg_id` 作为 top，最终把 `ReplyTo/ReplyToTopId` 一起落库；forum topic 发送后会更新 topic 的 top message。TDesktop 在 topic 输入框里会发送 `reply_to_msg_id=0 + top_msg_id=topicRootId`，参考实现 的 `InputReplyToMessage.TopMsgId` 也按 topic/thread 维度保存并在 header 上打 `forum_topic`。telesrv 借鉴该语义，但改成事务内校验目标或 topic root 未删除且对当前成员可见，非法目标返回 `REPLY_MESSAGE_ID_INVALID`，topic 内普通消息返回 `messageReplyHeader{forum_topic, reply_to_top_id}` 并更新 topic top message。 -参考实现 的 `updates.getChannelDifference` 还会在当前 participant 的 `AvailableMinPts > req.pts` 时把请求 pts 抬到 `AvailableMinPts`，再去读 `channel_pts_updates`。这是避免新成员用 `pts=0` 拉到入群前 participant/admin/settings 事件的关键边界；telesrv 因此在 `channel_members` 中同时保存 `available_min_pts`，加入/导入/受邀/重新加入时设为加入前 `channels.pts`，消息历史可见性仍由 `available_min_id` 独立控制。 +参考实现 的 `updates.getChannelDifference` 还会在当前 participant 的 `AvailableMinPts > req.pts` 时把请求 pts 抬到 `AvailableMinPts`，再去读 `channel_pts_updates`。这是避免新成员用 `pts=0` 拉到入群前消息类 durable 事件的关键边界；telesrv 因此在 `channel_members` 中同时保存 `available_min_pts`，加入/导入/受邀/重新加入时设为加入前 `channels.pts`，消息历史可见性仍由 `available_min_id` 独立控制。 参考实现 的 `channels.deleteHistory` 本地清空路径不写 channel pts，而是返回并同步 `updateChannelAvailableMessages{channel_id, available_min_id}`；TDesktop 在 `api_updates.cpp` 收到后设置 channel `available_min_id` 并对已加载 history 执行 `clearUpTill`，`ChannelData::setAvailableMinId` 本身不会做 max-clamp。telesrv 采用同一客户端语义：本地清空只更新当前账号成员/dialog 水位，同时写账号级 durable update 供其它设备在线推送或 `updates.getDifference` 离线恢复；返回和推送的 `available_min_id` 必须是实际应用后的单调水位 `max(old_available_min_id, requested_max_id)`，避免多设备乱序或 stale 请求把 TDesktop 本地可见下界回退。 @@ -162,7 +162,7 @@ username 与管理项方面，参考实现 的 `channels.checkUsername` 只校 - `importInvite` 必须在同一事务内锁定对应 invite row 后检查并递增 `usage_count`，避免多个客户端同时导入一次性链接时突破 `usage_limit`。 - `ChannelMember`：`ChannelID`、`UserID`、`InviterUserID`、`Role`、`Status`、`JoinedAt`、`LeftAt`、`AdminRights`、`BannedRights`、`Rank`、`AvailableMinID`、`AvailableMinPts`、`ReadInboxMaxID`、`ReadInboxDate`、`ReadOutboxMaxID`、`UnreadMark`。 - `ChannelMember.AvailableMinID`：当前成员可见历史下界；开启 prehistory hidden 后，新加入/导入/受邀成员初始化为加入前 `channel.top_message_id`，只看后续消息和自己的加入服务消息。 -- `ChannelMember.AvailableMinPts`：当前成员可恢复 channel difference 的 pts 下界；新加入/导入/受邀/重新加入成员初始化为加入前 `channels.pts`，`updates.getChannelDifference(pts=0)` 也会先抬到该值，避免入群前 participant/admin/settings durable event 泄漏。 +- `ChannelMember.AvailableMinPts`：当前成员可恢复 channel difference 的 pts 下界；新加入/导入/受邀/重新加入成员初始化为加入前 `channels.pts`，`updates.getChannelDifference(pts=0)` 也会先抬到该值，避免入群前消息类 durable event 泄漏。 - `ChannelMember.ReadInboxDate`：当前成员最后一次推进 `read_inbox_max_id` 的时间，用于 `messages.getMessageReadParticipants` 返回 `readParticipantDate.date`；不参与 channel pts。 - `ChannelMember.SlowmodeLastSendDate`：普通成员最近一次成功发言时间，用于服务端按 channel 维度返回 `SLOWMODE_WAIT_X`；creator/admin 不受首批 slowmode 限制。 - `ChannelMessage`：`ChannelID`、`ID`、`RandomID`、`SenderUserID`、`From`、`SendAs`、`Date`、`EditDate`、`Post`、`Silent`、`NoForwards`、`Body`、`Entities`、`ReplyTo`、`Forward`、`Action`、`Pts`、`Deleted`。 @@ -314,10 +314,10 @@ Redis miss 恢复来源： - `limit` cap：普通用户 `1..100`，超过按 100；内部硬上限 1000，拒绝负数和超大值。 - 若 `pts < 0` 或 `pts > current_channel_pts` 返回 `PERSISTENT_TIMESTAMP_INVALID`，避免客户端用未来水位跳过 durable log。 - 从 `channel_update_events` 读取 `pts > req.pts ORDER BY pts ASC LIMIT cap+1`。 -- `channel_update_events.payload` 对 new/edit/pin/admin participant 等事件保存 domain 快照；`updates.getChannelDifference` 必须优先使用事件时刻的 message/participant snapshot，不能回读当前 `channel_messages` 覆盖旧事件，否则连续编辑、删除后的离线补偿会丢失中间状态。 +- `channel_update_events.payload` 对 new/edit/pin 等消息类事件保存 domain 快照；`updates.getChannelDifference` 必须优先使用事件时刻的 message snapshot，不能回读当前 `channel_messages` 覆盖旧事件，否则连续编辑、删除后的离线补偿会丢失中间状态。 - 没有事件返回 `channelDifferenceEmpty{final=true, pts=current_channel_pts, timeout=30}`。 - 事件数 `<= cap` 返回 `channelDifference{final=true, pts=max_pts, new_messages, other_updates, chats, users}`。 -- 如果当前 member 的 `available_min_pts > req.pts`，先把请求 pts 抬到 `available_min_pts`，从源头跳过入群/重新加入前的 participant/admin/settings 事件。 +- 如果当前 member 的 `available_min_pts > req.pts`，先把请求 pts 抬到 `available_min_pts`，从源头跳过入群/重新加入前的消息类 durable 事件。 - 对 `available_min_id` 之后才可见的成员，普通差量仍扫描 durable log 并推进返回 `pts`，但会过滤 `new/edit/delete/pin` 中 `message_id <= available_min_id` 的消息内容和 id；若本页全被过滤，返回 `channelDifferenceEmpty{pts=max_scanned_pts}`，避免隐藏历史或本地清空后的旧消息通过差量恢复泄露。部分可见的 delete/pin 事件只裁剪 `messages` 向量，保留原始 `pts_count`；TDesktop 在线 update 用 `pts_count` 推进 channel PTS，差量响应最终用 `channelDifference.pts` 初始化，不要求 `len(messages)==pts_count`。 - 若 `current_channel_pts - req.pts > cap`，返回 `channelDifferenceTooLong`，包含带当前 channel pts 的 dialog、最新一页有界消息、channel、相关 users，避免大频道旧 pts 客户端循环拉取大量差量页。 - 否则事件数 `<= cap` 返回 `channelDifference{final=true, pts=max_pts, new_messages, other_updates, chats, users}`；事件数达到 cap 但仍未追上当前 pts 时返回 `final=false`，客户端会继续拉下一页。 @@ -375,7 +375,7 @@ Redis miss 恢复来源： - 已实现 `messages.getMessagesViews(InputPeerChannel)`：TDesktop 每秒最多 100 条批量增量，服务端按 `(channel_id,message_id,viewer_user_id)` 持久去重并维护 `channel_messages.views_count` 聚合列；本地清历史前不可见、已删除或不存在的 id 不递增且返回空 view，replies/comment 信息继续从 discussion/thread model 回填。 - 已实现 `channels.exportMessageLink`：复制频道/超级群消息链接前会校验 channel message 对当前成员真实存在且未被删除/本地清历史隐藏；公开 username 走 `t.me/{username}/{msg_id}`，私有 channel 走 `t.me/c/{channel_id}/{msg_id}`，普通 reply/thread 链接支持 `?thread={root_id}`。`grouped/html` 与 linked discussion 的 `?comment=` 细分链接留后续。 - 已实现管理面最小真实能力：`channels.editAdmin/editBanned/editTitle/deleteChannel`、`channels.getParticipants` 的 admins/kicked/banned/search 等过滤、`messages.updatePinnedMessage/unpinAllMessages`、`messages.exportChatInvite/checkChatInvite/importChatInvite`。`channels.deleteChannel` 按 TDesktop/参考实现预期返回并推送 `updateChannel + channelForbidden`，同时 dialog 列表过滤 deleted channel。`channels.editPhoto/messages.editChatPhoto` 在头像 media store 接入前只接受 `inputChatPhotoEmpty` no-op 删除，uploaded/existing photo 明确返回 `PHOTO_INVALID`，不伪造 `messageActionChatEditPhoto`。 -- `channels.editAdmin/editBanned` 会写 channel-scoped `channel_participant` durable event，保存 prev/new participant；在线响应/推送包含 `updateChannelParticipant + updateChannel + updateChannelTooLong(pts)`，离线或 pts gap 通过 `updates.getChannelDifference` 补偿。 +- `channels.editAdmin/editBanned` 只更新成员状态/计数并写 admin log，不占 channel pts、不写 `channel_update_events`；在线响应/推送包含 `updateChannelParticipant + updateChannel`。离线客户端通过 `channels.getFullChannel/getParticipants/getParticipant` 或后续可见消息触发的 channel state 刷新补偿；如果操作另行产生可见 service message，则由该 service message 进入 channel pts。 - 已实现公开 username 管理：`channels.checkUsername/updateUsername/getAdminedPublicChannels`，PG 用 `channel_usernames(username_lower)` 与 users username 查询避免跨 peer 占用；主 username 的清除只走 `channels.updateUsername("")`。 - 已实现 `channels.toggleSignatures`：权限校验后持久化 `channels.signatures`，返回/在线推 `updateChannel`。 - 已实现 `channels.updateColor/updateEmojiStatus`：颜色分别持久化 `color/profile_color` 与 background emoji id，并保留 color flag 显式 0；普通 emoji status 保存 document id/until，`emojiStatusEmpty` 清空，collectible gift 状态因缺少 gift/read model 先返回 `EMOJI_STATUS_INVALID`。响应、在线推送、`channels.getChannels` 都回填 `Channel.color/profile_color/emoji_status`。 @@ -394,7 +394,7 @@ Redis miss 恢复来源： - 已实现 invite management：TDesktop 管理 invite links、admins with invites、importers、join requests 时会调用 `messages.getExportedChatInvites/getExportedChatInvite/editExportedChatInvite/deleteExportedChatInvite/deleteRevokedExportedChatInvites/getAdminsWithInvites/getChatInviteImporters/hideChatJoinRequest/hideAllChatJoinRequests`；当前已持久化 invite 列表、detail/edit/revoke/delete、按 admin 统计、importer/read model、`request_needed` pending join request、单个 approve/dismiss 与 bounded `hideAll`。`getExportedChatInvites` 使用 `offset_date + offset_link` seek，`getChatInviteImporters` limit cap=100，`hideAll` 单批最多 1000，避免按客户端超大参数生成无界更新；subscription/chatlist/paid invite 与 join-request service notification 仍留后续。 - 已实现 public join settings：`channels.toggleJoinToSend`/`channels.toggleJoinRequest` 持久化 `channels.join_to_send/join_request` 并返回带 flags 28/29 的 `tg.Channel`；`join_request` 仅 public megagroup 可开启，非成员 `channels.joinChannel` 会写入 `channel_invite_importers(invite_id=0, requested=true)` 并返回 `INVITE_REQUEST_SENT`，之后可通过 `messages.getChatInviteImporters(requested=true)` 查询和 `messages.hideChatJoinRequest` approve/dismiss。admin 侧 `channels.getFullChannel` 会回填 `requests_pending/recent_requesters`，request-needed import/public join 以及 approve/dismiss 会向有界管理员集合推 `updatePendingJoinRequests`；该状态不写入无界 durable update log，离线管理员重新打开 full channel 时补偿。 - 已修正当前普通成员的 `channels.getParticipant(inputPeerSelf)` 与 participants 列表 TL constructor：TDesktop `requestSelf` 期望普通本人是 `channelParticipantSelf`，creator/admin 仍分别返回 creator/admin self 语义；避免客户端记录 `Got self regular participant`，且不改变 domain/store 成员模型。 -- PG channel pts 已补失败保护：现有 channel 的 send/edit/deleteHistory/deleteMessages/pin 和成员/标题服务消息在事务失败或权限失败后写 `noop` 占位，避免 Redis 分配过的 channel pts 形成 TDesktop `PtsWaiter` 永久 gap。 +- PG channel pts 已补失败保护：现有 channel 的 send/edit/deleteHistory/deleteMessages/pin 和加入/退出/标题等可见 service message 在事务失败或权限失败后写 `noop` 占位，避免 Redis 分配过的 channel pts 形成 TDesktop `PtsWaiter` 永久 gap。 - 2026-06-01 双 TDesktop 在线/离线实测已覆盖超级群发送、reply、forward、edit 与离线恢复：Bob 对 Alice 消息 reply，Bob 将回复 forward 回同一超级群，随后编辑该频道消息；Bob 关闭期间 Alice 发送 channel 消息，Bob 重启后 dialog 未读数=1，打开群后看到离线消息；Alice/Bob 双窗口均实时显示，server 日志无新增 `NOT_IMPLEMENTED` / `Unhandled RPC` / `bad_msg` / panic。删除、清历史、踢/禁言仍需用户行动时确认后做 UI 实测。 - 2026-06-02 Computer Use 双 TDesktop 复测已覆盖当前非破坏性频道/超级群 UI 路径：Alice/Bob 在 `E2E Super 0307` 中双向发送 `cu-round-alice-*` / `cu-round-bob-*` 并实时互见，成员栏显示 2 members/online，Alice 全局搜索 Bob 的新消息返回 `Found 1 message`；Alice 在 `CU Public Search 44238` 频道发布 `cu-channel-round-*` 后频道消息流和左侧 preview 同步更新。server 日志无新增 `NOT_IMPLEMENTED` / `Unhandled RPC` / `bad_msg` / panic，客户端本轮无新增 `Bad participant` / `Got self regular participant`；清空搜索框产生的 `SEARCH_QUERY_EMPTY` 保持可解释。 - 2026-06-02 09:55 Computer Use reaction/sticker 启动复测：Debug/Alice 与 DebugBob/Bob 同时打开 `E2E Super 0307`，互发 `cu-stubfix-alice-*` / `cu-stubfix-bob-*` 后双方消息列表和左侧 preview 均可见；打开 emoji 面板触发 `messages.getAvailableReactions` / `messages.getStickerSet` / `messages.getAvailableEffects`，server 日志无新增 `NOT_IMPLEMENTED` / `Unhandled RPC` / `bad_msg` / panic，Debug 当前 `log.txt` 无新增 `Unexpected messages.stickerSetNotModified` / participant 告警。右键消息菜单可打开但本轮未显示 reaction 快捷项，真实 reaction sticker animations/custom UI 仍留后续。 @@ -468,4 +468,4 @@ TDesktop 最小入口： - `channel_messages` 增 `media` JSONB 快照列（与私聊同构），`SendChannelMessage` 透传 `req.Media`，讨论组联动消息一并带 media；`scanChannel*`/`channelMessageColumns` 统一加 `media::text`，所有 history/getMessages/replies/difference 读取路径自动带出。`tgChannelMessage` 在 media 非空时 `SetMedia`。放宽 `channel_messages` content CHECK 为 `body<>'' OR action<>'{}' OR media<>'{}'`。 - 频道头像：`channels` 表反范式 `photo_id/photo_dc_id/photo_stripped`（migration `0059`），`channelColumns` + 全部 5 处 channel scanner 同步；`channels.editPhoto`/`messages.editChatPhoto` 经 `resolveInputChatPhoto` 上传或引用照片，admin（change_info）校验后落列并返回 `updateChannel` + 推 channel state；`tgChannel.Photo`(ChatPhoto)/`tgChannelFull.ChatPhoto` 渲染真实头像（`getFile` 按 `photo::` 解析忽略 access_hash，合成 a/c 尺寸即可下载）。 - 2026-06-03 接手审计修正：`SendChannelMessage` 的空内容校验已把 `req.Media` 纳入，允许超级群/频道发送无 caption 的 photo/document/sticker；新增 PG 集成测试覆盖 channel media 经 `ListChannelDifference` 恢复，防止离线 TDesktop 拉差分丢媒体。 -- 范围外：in-history `MessageActionChatEditPhoto` service 消息留 todo。 \ No newline at end of file +- 范围外：in-history `MessageActionChatEditPhoto` service 消息留 todo。 diff --git a/docs/compatibility-matrix.md b/docs/compatibility-matrix.md index 343de35c..f38116dc 100644 --- a/docs/compatibility-matrix.md +++ b/docs/compatibility-matrix.md @@ -127,7 +127,7 @@ status 取值：done(真实实现) / stub(兼容响应) / todo(已发现未实 |---|---|---|---| | updates.getState | done | real | auth_key+user 维度持久化 update_states；账号级当前 pts 报告**最大连续已提交 pts**（MaxContiguousPts，非 allocator 最大已分配值），避免同设备换号/多账号串差分，也避免越过在途空洞 | | updates.getDifference | done | real | 按 user_id 从 user_update_events 拉取 pts 后增量；只返回从客户端 pts 起**连续**的事件（遇在途空洞即截断），超 100 条置 differenceSlice；支持 new_message、read_history_inbox/read_history_outbox（私聊与 channel peer，channel read 映射 updateReadChannelInbox）、edit_message、message_reactions（输出带最新 `message.reactions` 的 affected message + `updateMessageReactions`，并按 viewer 重取最新聚合，避免 TDesktop 离线恢复时本地 message cache 不刷新）、delete_messages、contacts_reset、dialog pinned/order/manual unread、peer_settings、dialog filters/folder peers 与 noop gap；消息事件携带 fwd/reply 所需 users/chats，payload 来自 durable log。**账号级绝不返回 `differenceTooLong`**：已核对 TDesktop 基线 `api_updates.cpp:516`——对账号级 differenceTooLong 只打一行日志、不读 pts 且漏 `setRequesting(false)`，会永久锁死 update 引擎（`:689` 早退，重连/新 session 不可恢复）；落后客户端改用 `differenceSlice` 续传。因此 `user_update_events` **永久保留、不做 retention 裁剪**（对齐 参考实现），详见 docs/performance-audit.md 附录 C | -| updates.getChannelDifference | done | real-channel | 超级群/频道使用 channel 维度 durable log：`channel_update_events(channel_id, pts, pts_count, ...)`；按 channel pts 返回 `channelDifferenceEmpty/channelDifference/channelDifferenceTooLong`，limit cap=100；`pts < 0` 或 `pts > current_channel_pts` 返回 `PERSISTENT_TIMESTAMP_INVALID`；当前 member 的 `available_min_pts` 会抬高请求 pts，避免新加入/重新加入成员拉到入群前 participant/admin/settings 事件；公共 username 频道允许非成员以只读预览身份拉可见差分并返回 synthetic read dialog，私有频道和禁看用户仍返回权限错误；当 `current_channel_pts-pts > cap` 时返回带当前 dialog pts 和最新有界消息快照的 `channelDifferenceTooLong`，避免大频道旧 pts 客户端循环拉大量页；普通 difference 优先使用事件 payload 中的 message/participant 快照，连续编辑/删除后不会被当前消息状态污染；本页消息的 sender/send_as/fwd_from/reply_to/action peers 会随 users/chats 返回，禁止复用 user_update_events | +| updates.getChannelDifference | done | real-channel | 超级群/频道使用 channel 维度 durable log：`channel_update_events(channel_id, pts, pts_count, ...)`；按 channel pts 返回 `channelDifferenceEmpty/channelDifference/channelDifferenceTooLong`，limit cap=100；`pts < 0` 或 `pts > current_channel_pts` 返回 `PERSISTENT_TIMESTAMP_INVALID`；当前 member 的 `available_min_pts` 会抬高请求 pts，避免新加入/重新加入成员拉到入群前消息类 durable 事件；公共 username 频道允许非成员以只读预览身份拉可见差分并返回 synthetic read dialog，私有频道和禁看用户仍返回权限错误；当 `current_channel_pts-pts > cap` 时返回带当前 dialog pts 和最新有界消息快照的 `channelDifferenceTooLong`，避免大频道旧 pts 客户端循环拉大量页；普通 difference 优先使用事件 payload 中的 message 快照，连续编辑/删除后不会被当前消息状态污染；`updateChannelParticipant` 不进入 channel pts log，成员权限/封禁状态靠在线 `updateChannelParticipant/updateChannel` 与 `channels.getFullChannel/getParticipants/getParticipant` 刷新；本页消息的 sender/send_as/fwd_from/reply_to/action peers 会随 users/chats 返回，禁止复用 user_update_events | ### 兼容硬约束：pts / pts_count（跨所有产生 update 的 RPC） @@ -279,7 +279,7 @@ outbox 多 worker 并发 + 发送事务乱序提交 → **主动推送可能乱 | channels.updateUsername | done | real | creator 可设置/清除频道或超级群主 username；PG 事务内更新 `channels.username` 与 `channel_usernames`，大小写不敏感唯一约束兜底，成功后向在线成员推 `updateChannel` | | channels.getAdminedPublicChannels | done | real | 返回当前用户 active creator/admin 且带主 username 的公开 channel/supergroup，用于 TDesktop public username limit/选择入口 | | channels.toggleSignatures | done | real-minimal | creator/change_info admin 可切换 `channels.signatures` 并返回/推送 `updateChannel`；profiles_enabled 先不单独持久化 | -| channels.togglePreHistoryHidden | done | real-minimal | creator 可切换 `channels.pre_history_hidden`，返回/在线推 `updateChannel`，`getFullChannel` 暴露 `hidden_prehistory`；新加入/导入/受邀成员按当前 `top_message_id` 初始化 `available_min_id/read_inbox`，并按加入前 channel pts 初始化 `available_min_pts`，避免看到旧历史或补到入群前成员事件 | +| channels.togglePreHistoryHidden | done | real-minimal | creator 可切换 `channels.pre_history_hidden`，返回/在线推 `updateChannel`，`getFullChannel` 暴露 `hidden_prehistory`；新加入/导入/受邀成员按当前 `top_message_id` 初始化 `available_min_id/read_inbox`，并按加入前 channel pts 初始化 `available_min_pts`，避免看到旧历史或补到入群前消息类 durable 事件 | | channels.toggleSlowMode | done | real-minimal | change_info 管理员可持久化 `channels.slowmode_seconds`，`Channel.slowmode_enabled/ChannelFull.slowmode_seconds` 可见；普通成员发送按 `channel_members.slowmode_last_send_date` 返回 `SLOWMODE_WAIT_X`，管理员/creator 豁免 | | channels.setStickers | stub | empty-only | 当前无群贴纸集 store；仅 megagroup + `inputStickerSetEmpty` 清空入口做权限校验 no-op 成功，非空 sticker set 返回 `STICKERSET_INVALID`，避免 TDesktop 误以为贴纸集已落库 | | channels.reorderUsernames | stub | permission-ok | 当前无 Fragment/多 username 模型，限制 order<=32、校验 change_info 后返回 BoolTrue | @@ -323,8 +323,8 @@ outbox 多 worker 并发 + 发送事务乱序提交 → **主动推送可能乱 | channels.readHistory | done | real | TDesktop 可能直接调用 channels.readHistory；语义同 messages.readHistory(InputPeerChannel)，含发送方 `updateReadChannelOutbox` 在线通知 | | channels.deleteMessages | done | real-channel | 管理员/作者权限校验后软删单份 channel messages，生成有界 `updateDeleteChannelMessages`，`pts_count=len(ids)`，单次 id cap=1000 | | channels.deleteHistory | done | real-partial | `for_everyone` 执行一个有界管理员删除 page 并推 `updateDeleteChannelMessages`，单批 cap=1000；该 TL 返回 `Updates` 且 TDesktop 不读取 offset，禁止在同步 RPC 内循环构造超大 id/update；非 for_everyone 只清当前用户可见历史/read/dialog，不写扩散、不生成超大 update | -| channels.editAdmin | done | real | creator 或具备 add_admins 的 admin 可更新 admin rights/rank；非 creator 只能授予自己拥有的权限；禁止改 creator；写 `channel_participant` durable event，返回/在线推 `updateChannelParticipant/updateChannel/updateChannelTooLong(pts)` | -| channels.editBanned | done | real | creator 或具备 ban_users 的 admin 可更新 banned rights/kicked 状态，刷新 participants/admin/banned/kicked 计数；写 `channel_participant` durable event，返回/在线推 `updateChannelParticipant/updateChannel/updateChannelTooLong(pts)` | +| channels.editAdmin | done | real | creator 或具备 add_admins 的 admin 可更新 admin rights/rank；非 creator 只能授予自己拥有的权限；禁止改 creator；写 admin log，不占 channel pts、不写 `channel_update_events`；返回/在线推 `updateChannelParticipant + updateChannel`，离线设备通过 full channel/participants 刷新状态 | +| channels.editBanned | done | real | creator 或具备 ban_users 的 admin 可更新 banned rights/kicked 状态，刷新 participants/admin/banned/kicked 计数；写 admin log，不占 channel pts、不写 `channel_update_events`；返回/在线推 `updateChannelParticipant + updateChannel`，若后续产生可见踢人/加人 service message，则 service message 单独占 channel pts | | channels.editTitle | done | real | creator/change_info admin 可改标题；写 megagroup service message `messageActionChatEditTitle` 与 channel pts，返回 `updateChannel + updateNewChannelMessage` | | channels.editPhoto | done | real-photo | change_info 权限校验后解析 `inputChatPhoto`：`inputChatUploadedPhoto` 组装上传→建 Photo，`inputChatPhoto{inputPhoto}` 引用已存在照片，`inputChatPhotoEmpty`/`inputPhotoEmpty` 清除；落 `channels.photo_id/photo_dc_id/photo_stripped` 反范式列，`tgChannel.Photo`(ChatPhoto)/`tgChannelFull.ChatPhoto` 渲染真实头像，返回 `updateChannel` + 推 channel state；服务端无 Files 时按 `PHOTO_INVALID` 处理；in-history `MessageActionChatEditPhoto` service 消息留 todo | | channels.deleteChannel | done | real | creator 权限，标记 channel deleted，返回/推送 `updateChannel + channelForbidden`，dialog 列表过滤 deleted channel；后续 admin log 细化 | @@ -417,4 +417,4 @@ outbox 多 worker 并发 + 发送事务乱序提交 → **主动推送可能乱 | method/type_id | first_seen | raw_note | |---|---|---| -| — | — | 当前无未实现 RPC trace | \ No newline at end of file +| — | — | 当前无未实现 RPC trace | diff --git a/docs/persistence-layer.md b/docs/persistence-layer.md index 77fd4f35..dcde1809 100644 --- a/docs/persistence-layer.md +++ b/docs/persistence-layer.md @@ -63,8 +63,8 @@ DDL 见 [`deploy/migrations/0001_init.up.sql`](../deploy/migrations/0001_init.up - **`private_messages`** —— 共享私聊消息主体，按 `sender_user_id` HASH 分区；`sender_user_id + random_id` 唯一保证 `messages.sendMessage/forwardMessages` 幂等；文本编辑更新共享 body/entities/edit_date；silent/noforwards/reply_to/fwd_from 元数据随消息持久化。 - **`message_boxes`** —— owner 视角消息盒，按 `owner_user_id` HASH 分区；每个账号看到自己的 `box_id`、peer、outgoing、pts、edit_date 与删除状态，历史/搜索走该表索引。删除只软删 owner 视角 message_box，`revoke` 通过 `(message_sender_id, private_message_id)` 定位其它 owner 视角并软删；编辑会同步所有可见 owner 视角盒子。该反向定位与分区键不一致，规划会展开全部 owner 分区，后续应增加 unpartitioned box 映射或先推导 owner_user_id 后再按 owner 分区点查。 - **`dialogs`** —— 当前账号会话摘要，按 `user_id` HASH 分区；只允许 `user` peer，支持 top message、置顶过滤、folder_id=0/1 主列表/归档与 offset 分页，并保存当前 owner 的 `pinned_order`、manual `unread_mark` 与 `hidden_peer_settings_bar`。列表查询 join `contacts` 时优先返回当前 owner 保存的联系人姓名/电话，避免不同账号看同一 peer 串备注；相关状态变化会写入账号级 durable update log，离线设备可通过 `updates.getDifference` 恢复。 -- **`channels` / `channel_messages` / `channel_message_viewers` / `channel_update_events`** —— 超级群/频道单份消息模型，按 `channel_id` HASH 分区；`channels.pts` 是 channel-scoped durable log 水位，`channels.forum/forum_tabs` 持久化 megagroup topics 开关与 TDesktop tabs/list 布局，`channels.participants_hidden` 持久化隐藏成员设置并由 `ChannelFull.participants_hidden` 恢复 TDesktop UI，`channels.antispam` 持久化 native anti-spam 开关并由 `ChannelFull.antispam` 恢复管理入口，`channels.color_set/color/color_background_emoji_id/profile_color_set/profile_color/profile_color_background_emoji_id/emoji_status_document_id/emoji_status_until` 持久化频道外观并回填 `Channel.color/profile_color/emoji_status`，`channels.linked_chat_id` 维护 broadcast 与 discussion megagroup 的双向链接并走 `channels_linked_chat_idx` 反查，`channel_update_events(channel_id, pts)` 存 new/edit/delete/pin/participant/noop 的恢复负载，`channel_messages(channel_id, id)` 走 seek pagination 并保存 `views_count` 聚合列，`channel_message_viewers(channel_id,message_id,viewer_user_id)` 用主键完成 views 去重递增，`reply_to_msg_id/reply_to_top_id` 支撑 thread/comment 分页，`discussion_channel_id/discussion_message_id` 把 broadcast post 映射到 linked megagroup root，禁止按成员写扩散。 -- **`channel_members` / `channel_dialogs` / `channel_unread_mentions`** —— 成员权限、读水位、owner 视角 channel dialog 与未读提及索引，分别按 `channel_id` / `user_id` / `user_id` HASH 分区；`available_min_id` 限制成员可见历史消息，`available_min_pts` 限制 `updates.getChannelDifference` 起点，避免新成员或重新加入成员恢复到入群前的 participant/admin/settings 事件。`channel_dialogs.unread_count` 是小超级群普通未读缓存字段，不是 broadcast/大超级群真值；大频道读取 dialog/full channel 时按 `channel_members.read_inbox_max_id`、`available_min_id`、`channels.top_message_id` 与未删除消息动态派生普通未读。`channel_dialogs.default_send_as_peer_type/default_send_as_peer_id` 保存当前 owner 的默认发送身份，由 `channels.getFullChannel` 输出为 `channelFull.default_send_as`，不参与历史分页或 dialog 排序；`channel_dialogs.view_forum_as_messages` 是当前账号本地 forum 展示模式，由 `Dialog/ChannelFull.view_forum_as_messages` 恢复 UI，并通过账号级 durable update 同步多 session。`channel_unread_mentions(user_id,channel_id,message_id)` 不复制消息正文，发送时只写解析出的 active/可见/未读成员，清除后重算 `channel_dialogs.unread_mentions_count`。共同超级群查询已迁到 `user_channel_member_index(user_id, channel_id)`，排除 broadcast 与非 active/deleted 成员；后续 `channels.getLeftChannels`、joined/admined channel 列表和启动 dialog 聚合也必须从 user 维度 read model 或两步 channel_id 列表读取，不能直接用 `channel_members WHERE user_id=...` 反向扫 `channel_id` 分区。 +- **`channels` / `channel_messages` / `channel_message_viewers` / `channel_update_events`** —— 超级群/频道单份消息模型，按 `channel_id` HASH 分区；`channels.pts` 是 channel-scoped durable log 水位，`channels.forum/forum_tabs` 持久化 megagroup topics 开关与 TDesktop tabs/list 布局，`channels.participants_hidden` 持久化隐藏成员设置并由 `ChannelFull.participants_hidden` 恢复 TDesktop UI，`channels.antispam` 持久化 native anti-spam 开关并由 `ChannelFull.antispam` 恢复管理入口，`channels.color_set/color/color_background_emoji_id/profile_color_set/profile_color/profile_color_background_emoji_id/emoji_status_document_id/emoji_status_until` 持久化频道外观并回填 `Channel.color/profile_color/emoji_status`，`channels.linked_chat_id` 维护 broadcast 与 discussion megagroup 的双向链接并走 `channels_linked_chat_idx` 反查，`channel_update_events(channel_id, pts)` 存 new/edit/delete/pin/noop 的恢复负载，成员权限/封禁变化不进入该 log；`channel_messages(channel_id, id)` 走 seek pagination 并保存 `views_count` 聚合列，`channel_message_viewers(channel_id,message_id,viewer_user_id)` 用主键完成 views 去重递增，`reply_to_msg_id/reply_to_top_id` 支撑 thread/comment 分页，`discussion_channel_id/discussion_message_id` 把 broadcast post 映射到 linked megagroup root，禁止按成员写扩散。 +- **`channel_members` / `channel_dialogs` / `channel_unread_mentions`** —— 成员权限、读水位、owner 视角 channel dialog 与未读提及索引，分别按 `channel_id` / `user_id` / `user_id` HASH 分区；`available_min_id` 限制成员可见历史消息，`available_min_pts` 限制 `updates.getChannelDifference` 起点，避免新成员或重新加入成员恢复到入群前的消息类 durable 事件。`channel_dialogs.unread_count` 是小超级群普通未读缓存字段，不是 broadcast/大超级群真值；大频道读取 dialog/full channel 时按 `channel_members.read_inbox_max_id`、`available_min_id`、`channels.top_message_id` 与未删除消息动态派生普通未读。`channel_dialogs.default_send_as_peer_type/default_send_as_peer_id` 保存当前 owner 的默认发送身份，由 `channels.getFullChannel` 输出为 `channelFull.default_send_as`，不参与历史分页或 dialog 排序；`channel_dialogs.view_forum_as_messages` 是当前账号本地 forum 展示模式，由 `Dialog/ChannelFull.view_forum_as_messages` 恢复 UI，并通过账号级 durable update 同步多 session。`channel_unread_mentions(user_id,channel_id,message_id)` 不复制消息正文，发送时只写解析出的 active/可见/未读成员，清除后重算 `channel_dialogs.unread_mentions_count`。共同超级群查询已迁到 `user_channel_member_index(user_id, channel_id)`，排除 broadcast 与非 active/deleted 成员；后续 `channels.getLeftChannels`、joined/admined channel 列表和启动 dialog 聚合也必须从 user 维度 read model 或两步 channel_id 列表读取，不能直接用 `channel_members WHERE user_id=...` 反向扫 `channel_id` 分区。 - **`channel_invites` / `channel_invite_importers`** —— 邀请链接、导入者与 join request read model，均按 `channel_id` HASH 分区；invite 保存 `usage_count/requested_count`，importer 以 `(channel_id,user_id)` 保证同一用户只有一个 pending/approved 状态，管理页查询走 `admin/revoked/offset_link` 与 `requested/link/date/user_id` seek 索引，禁止按超大 limit 或 hash 反查做全表扫。public `channels.toggleJoinRequest` 使用 `channels.join_request` 与 `channel_invite_importers(invite_id=0, requested=true)` 表达非 invite-link pending request；管理员实时提醒用 bounded `updatePendingJoinRequests` + full channel 回填，不为每条 pending 状态生成无界 durable updates。 - **`dialog_filters` / `dialog_filter_settings`** —— 当前账号自定义 dialog filter、filter 顺序与 folder tags 开关，按 `user_id` HASH 分区；自定义 filter 从 ID 2 开始，归档只由 `dialogs.folder_id=1` 表达，避免一列同时承担归档状态和任意筛选规则。 - **`dispatch_outbox`** —— 按 `target_user_id` HASH 分区的 transactional outbox。发送事务内写入，RPC outbox worker 用 `FOR UPDATE SKIP LOCKED` 批量 claim，成功标记 delivered，失败退避重试；排除当前设备使用 `exclude_auth_key_id + exclude_session_id`，避免一个设备换号或多账号登录时误过滤。按 target 分区适合投递完成/失败按用户更新，但全局 claim/cleanup 与分区键不一致，规划会展开所有分区；上量前需引入 ready queue 或 worker shard read model。 @@ -171,4 +171,4 @@ blob backend：`internal/app/files`（`BlobBackend` 接口 + `LocalFS` 本地磁 种子导入：启动时 `files.Service.SeedMedia` 从 `TELESRV_STICKER_SEED_DIR`（真实 Telegram 导出，含 `available_reactions_raw.json` + 各集 `set_info.json` + `.tgs/.webp/缩略图`）幂等导入：JSON 元数据→表，二进制→blob，`dc_id` 重写为本 server DC。实测 74 reactions / 24 sets / ~1.5k documents / ~3k blob 索引 / ~2.8k 去重磁盘文件。 -类型边界：`domain.Document/Photo/MessageMedia/StickerSet/AvailableReaction` 带 json tag（store 直接 marshal JSONB），完全不依赖 `tg.*`；domain↔tg 转换集中在 `internal/rpc/convert_media.go`。 \ No newline at end of file +类型边界：`domain.Document/Photo/MessageMedia/StickerSet/AvailableReaction` 带 json tag（store 直接 marshal JSONB），完全不依赖 `tg.*`；domain↔tg 转换集中在 `internal/rpc/convert_media.go`。 diff --git a/internal/app/channels/service_test.go b/internal/app/channels/service_test.go index 73b10e5f..9e34a532 100644 --- a/internal/app/channels/service_test.go +++ b/internal/app/channels/service_test.go @@ -784,6 +784,7 @@ func TestChannelAdminTitlePinAndInvite(t *testing.T) { if err != nil { t.Fatalf("CreateMegagroupFromCreateChat: %v", err) } + ptsBeforeAdmin := created.Channel.Pts admin, err := service.EditAdmin(ctx, 1001, domain.EditChannelAdminRequest{ ChannelID: created.Channel.ID, @@ -802,15 +803,18 @@ func TestChannelAdminTitlePinAndInvite(t *testing.T) { if admin.Participant.Role != domain.ChannelRoleAdmin || !admin.Participant.AdminRights.PinMessages || admin.Channel.AdminsCount != 2 { t.Fatalf("admin result = %+v, want promoted admin with counts", admin) } - if admin.Event.Type != domain.ChannelUpdateParticipant || admin.Event.PtsCount != 1 || admin.Event.Participant.UserID != 1002 || admin.Event.Previous.UserID != 1002 { - t.Fatalf("admin participant event = %+v, want durable participant transition", admin.Event) + if admin.Channel.Pts != ptsBeforeAdmin { + t.Fatalf("admin channel pts = %d, want unchanged %d", admin.Channel.Pts, ptsBeforeAdmin) } - diffAfterAdmin, err := service.GetDifference(ctx, 1002, domain.ChannelDifferenceRequest{ChannelID: created.Channel.ID, Pts: 1, Limit: 10}) + if admin.Event.Type != domain.ChannelUpdateParticipant || admin.Event.Pts != 0 || admin.Event.PtsCount != 0 || admin.Event.Participant.UserID != 1002 || admin.Event.Previous.UserID != 1002 { + t.Fatalf("admin participant event = %+v, want transient participant transition", admin.Event) + } + diffAfterAdmin, err := service.GetDifference(ctx, 1002, domain.ChannelDifferenceRequest{ChannelID: created.Channel.ID, Pts: ptsBeforeAdmin, Limit: 10}) if err != nil { t.Fatalf("GetDifference after admin: %v", err) } - if len(diffAfterAdmin.OtherUpdates) != 1 || diffAfterAdmin.OtherUpdates[0].Type != domain.ChannelUpdateParticipant { - t.Fatalf("diff after admin = %+v, want participant update in channel difference", diffAfterAdmin) + if len(diffAfterAdmin.OtherUpdates) != 0 || diffAfterAdmin.Pts != ptsBeforeAdmin { + t.Fatalf("diff after admin = %+v, want no durable participant update", diffAfterAdmin) } admins, err := service.GetParticipants(ctx, 1001, created.Channel.ID, domain.ChannelParticipantsFilter{Kind: domain.ChannelParticipantsAdmins}, 0, 10) if err != nil { @@ -995,6 +999,7 @@ func TestChannelBanAndDeletePermissions(t *testing.T) { if err != nil { t.Fatalf("CreateMegagroupFromCreateChat: %v", err) } + ptsBeforeBan := created.Channel.Pts if _, err := service.DeleteChannel(ctx, 1002, domain.DeleteChannelRequest{ChannelID: created.Channel.ID, Date: 11}); !errors.Is(err, domain.ErrChannelAdminRequired) { t.Fatalf("member DeleteChannel err = %v, want ErrChannelAdminRequired", err) } @@ -1013,8 +1018,11 @@ func TestChannelBanAndDeletePermissions(t *testing.T) { if banned.Participant.Status != domain.ChannelMemberKicked || banned.Channel.ParticipantsCount != 1 || banned.Channel.KickedCount != 1 { t.Fatalf("banned = %+v, want kicked participant and counts", banned) } - if banned.Event.Type != domain.ChannelUpdateParticipant || banned.Event.Participant.Status != domain.ChannelMemberKicked || banned.Event.PtsCount != 1 { - t.Fatalf("ban participant event = %+v, want durable kicked transition", banned.Event) + if banned.Channel.Pts != ptsBeforeBan { + t.Fatalf("banned channel pts = %d, want unchanged %d", banned.Channel.Pts, ptsBeforeBan) + } + if banned.Event.Type != domain.ChannelUpdateParticipant || banned.Event.Participant.Status != domain.ChannelMemberKicked || banned.Event.Pts != 0 || banned.Event.PtsCount != 0 { + t.Fatalf("ban participant event = %+v, want transient kicked transition", banned.Event) } kicked, err := service.GetParticipants(ctx, 1001, created.Channel.ID, domain.ChannelParticipantsFilter{Kind: domain.ChannelParticipantsKicked}, 0, 10) if err != nil { @@ -1348,6 +1356,7 @@ func TestChannelDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { if err != nil { t.Fatalf("CreateMegagroupFromCreateChat: %v", err) } + ptsFloor := created.Channel.Pts promoted, err := service.EditAdmin(ctx, 1001, domain.EditChannelAdminRequest{ ChannelID: created.Channel.ID, MemberID: 1002, @@ -1359,12 +1368,15 @@ func TestChannelDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { if err != nil { t.Fatalf("EditAdmin: %v", err) } + if promoted.Event.Pts != 0 || promoted.Channel.Pts != ptsFloor { + t.Fatalf("promoted = %+v, want transient admin event and unchanged pts %d", promoted, ptsFloor) + } joined, err := service.JoinChannel(ctx, 1003, created.Channel.ID, 12) if err != nil { t.Fatalf("JoinChannel: %v", err) } - if joined.Members[0].AvailableMinPts != promoted.Event.Pts { - t.Fatalf("joined available_min_pts = %d, want pre-join channel pts %d", joined.Members[0].AvailableMinPts, promoted.Event.Pts) + if joined.Members[0].AvailableMinPts != ptsFloor { + t.Fatalf("joined available_min_pts = %d, want pre-join channel pts %d", joined.Members[0].AvailableMinPts, ptsFloor) } diff, err := service.GetDifference(ctx, 1003, domain.ChannelDifferenceRequest{ChannelID: created.Channel.ID, Pts: 0, Limit: 100}) if err != nil { @@ -1374,13 +1386,13 @@ func TestChannelDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { t.Fatalf("diff pts = %d, want current channel pts %d", diff.Pts, joined.Channel.Pts) } for _, msg := range diff.NewMessages { - if msg.Pts <= promoted.Event.Pts { - t.Fatalf("diff leaks pre-join message %+v at or before available_min_pts %d", msg, promoted.Event.Pts) + if msg.Pts <= ptsFloor { + t.Fatalf("diff leaks pre-join message %+v at or before available_min_pts %d", msg, ptsFloor) } } for _, event := range diff.OtherUpdates { - if event.Pts <= promoted.Event.Pts { - t.Fatalf("diff leaks pre-join event %+v at or before available_min_pts %d", event, promoted.Event.Pts) + if event.Pts <= ptsFloor { + t.Fatalf("diff leaks pre-join event %+v at or before available_min_pts %d", event, ptsFloor) } } } diff --git a/internal/rpc/channels.go b/internal/rpc/channels.go index 5a56c7f4..0afff93e 100644 --- a/internal/rpc/channels.go +++ b/internal/rpc/channels.go @@ -1861,9 +1861,9 @@ func (r *Router) onChannelsEditAdmin(ctx context.Context, req *tg.ChannelsEditAd } else { r.removeOnlineChannelMemberships(res.Channel.ID, res.Participant.UserID) } - updates := r.channelParticipantUpdates(ctx, userID, userID, res.Channel, res.Previous, res.Participant, res.Event, res.Date) + updates := r.channelParticipantUpdates(ctx, userID, userID, res.Channel, res.Previous, res.Participant, res.Date) r.pushChannelUpdates(ctx, userID, res.Channel.ID, res.Recipients, func(viewerUserID int64) *tg.Updates { - return r.channelParticipantUpdates(ctx, viewerUserID, userID, res.Channel, res.Previous, res.Participant, res.Event, res.Date) + return r.channelParticipantUpdates(ctx, viewerUserID, userID, res.Channel, res.Previous, res.Participant, res.Date) }) return updates, nil } @@ -1894,9 +1894,9 @@ func (r *Router) onChannelsEditBanned(ctx context.Context, req *tg.ChannelsEditB if err != nil { return nil, channelAdminErr(err) } - updates := r.channelParticipantUpdates(ctx, userID, userID, res.Channel, res.Previous, res.Participant, res.Event, res.Date) + updates := r.channelParticipantUpdates(ctx, userID, userID, res.Channel, res.Previous, res.Participant, res.Date) r.pushChannelUpdates(ctx, userID, res.Channel.ID, res.Recipients, func(viewerUserID int64) *tg.Updates { - return r.channelParticipantUpdates(ctx, viewerUserID, userID, res.Channel, res.Previous, res.Participant, res.Event, res.Date) + return r.channelParticipantUpdates(ctx, viewerUserID, userID, res.Channel, res.Previous, res.Participant, res.Date) }) return updates, nil } @@ -2904,7 +2904,7 @@ func (r *Router) channelTitleUpdates(ctx context.Context, viewerUserID int64, re } } -func (r *Router) channelParticipantUpdates(ctx context.Context, viewerUserID, actorUserID int64, channel domain.Channel, previous, participant domain.ChannelMember, event domain.ChannelUpdateEvent, date int) *tg.Updates { +func (r *Router) channelParticipantUpdates(ctx context.Context, viewerUserID, actorUserID int64, channel domain.Channel, previous, participant domain.ChannelMember, date int) *tg.Updates { update := &tg.UpdateChannelParticipant{ ChannelID: channel.ID, Date: date, @@ -2920,14 +2920,8 @@ func (r *Router) channelParticipantUpdates(ctx context.Context, viewerUserID, ac if participant.UserID != 0 { update.SetNewParticipant(tgChannelParticipantForUpdate(viewerUserID, participant)) } - updates := []tg.UpdateClass{update, &tg.UpdateChannel{ChannelID: channel.ID}} - if event.Pts > 0 { - tooLong := &tg.UpdateChannelTooLong{ChannelID: channel.ID} - tooLong.SetPts(event.Pts) - updates = append(updates, tooLong) - } return &tg.Updates{ - Updates: updates, + Updates: []tg.UpdateClass{update, &tg.UpdateChannel{ChannelID: channel.ID}}, Users: r.tgUsersForIDs(ctx, viewerUserID, []int64{participant.UserID, participant.InviterUserID, previous.UserID, previous.InviterUserID, update.ActorID}), Chats: []tg.ChatClass{tgChannelChat(viewerUserID, channel, nil)}, Date: int(r.clock.Now().Unix()), diff --git a/internal/rpc/router_test.go b/internal/rpc/router_test.go index 8121d6b8..7e87270c 100644 --- a/internal/rpc/router_test.go +++ b/internal/rpc/router_test.go @@ -4186,6 +4186,11 @@ func TestChannelAdminPinInviteRPC(t *testing.T) { t.Fatalf("create chat: %v", err) } channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel) + createdChannel, err := channelStore.GetChannelByID(ctx, channel.ID) + if err != nil { + t.Fatalf("get created channel: %v", err) + } + initialChannelPts := createdChannel.Pts selfParticipant, err := r.onChannelsGetParticipant(WithUserID(ctx, friend.ID), &tg.ChannelsGetParticipantRequest{ Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, @@ -4229,29 +4234,25 @@ func TestChannelAdminPinInviteRPC(t *testing.T) { if err != nil { t.Fatalf("edit admin: %v", err) } - if updates := adminUpdates.(*tg.Updates); len(updates.Updates) < 3 { - t.Fatalf("admin updates empty, want participant update") + if updates := adminUpdates.(*tg.Updates); len(updates.Updates) != 2 { + t.Fatalf("admin updates = %+v, want participant update and channel refresh", updates.Updates) } else if _, ok := updates.Updates[0].(*tg.UpdateChannelParticipant); !ok { t.Fatalf("admin update[0] = %T, want updateChannelParticipant", updates.Updates[0]) - } else if tooLong, ok := updates.Updates[2].(*tg.UpdateChannelTooLong); !ok { - t.Fatalf("admin update[2] = %T, want updateChannelTooLong", updates.Updates[2]) - } else if pts, ok := tooLong.GetPts(); !ok || pts == 0 { - t.Fatalf("admin updateChannelTooLong pts = %d ok=%v, want set pts", pts, ok) + } else if _, ok := updates.Updates[1].(*tg.UpdateChannel); !ok { + t.Fatalf("admin update[1] = %T, want updateChannel", updates.Updates[1]) } adminDiff, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetChannelDifferenceRequest{ Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, Filter: &tg.ChannelMessagesFilterEmpty{}, - Pts: 1, + Pts: initialChannelPts, Limit: 10, }) if err != nil { t.Fatalf("channel difference after admin: %v", err) } - adminFullDiff, ok := adminDiff.(*tg.UpdatesChannelDifference) - if !ok || len(adminFullDiff.OtherUpdates) == 0 { - t.Fatalf("admin diff = %T %+v, want participant other update", adminDiff, adminDiff) - } else if _, ok := adminFullDiff.OtherUpdates[0].(*tg.UpdateChannelParticipant); !ok { - t.Fatalf("admin diff update[0] = %T, want updateChannelParticipant", adminFullDiff.OtherUpdates[0]) + adminEmptyDiff, ok := adminDiff.(*tg.UpdatesChannelDifferenceEmpty) + if !ok || !adminEmptyDiff.Final || adminEmptyDiff.Pts != initialChannelPts { + t.Fatalf("admin diff = %T %+v, want empty difference at unchanged pts %d", adminDiff, adminDiff, initialChannelPts) } admins, err := r.onChannelsGetParticipants(WithUserID(ctx, owner.ID), &tg.ChannelsGetParticipantsRequest{ Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash}, diff --git a/internal/store/memory/channel.go b/internal/store/memory/channel.go index 0be16f9b..b80c401f 100644 --- a/internal/store/memory/channel.go +++ b/internal/store/memory/channel.go @@ -627,8 +627,7 @@ func (s *ChannelStore) EditChannelAdmin(_ context.Context, req domain.EditChanne }) s.refreshChannelCountsLocked(req.ChannelID) channel = s.channels[req.ChannelID] - event := s.appendParticipantEventLocked(channel, req.UserID, previous, member, req.Date) - channel = s.channels[req.ChannelID] + event := transientChannelParticipantEvent(channel.ID, req.UserID, previous, member, req.Date) if msg, ok := s.findMessageLocked(req.ChannelID, channel.TopMessageID); ok { s.upsertChannelDialogLocked(member.UserID, channel, msg, false) } @@ -702,8 +701,7 @@ func (s *ChannelStore) EditChannelBanned(_ context.Context, req domain.EditChann }) s.refreshChannelCountsLocked(req.ChannelID) channel = s.channels[req.ChannelID] - event := s.appendParticipantEventLocked(channel, req.UserID, previous, member, req.Date) - channel = s.channels[req.ChannelID] + event := transientChannelParticipantEvent(channel.ID, req.UserID, previous, member, req.Date) if member.Status == domain.ChannelMemberActive { if msg, ok := s.findMessageLocked(req.ChannelID, channel.TopMessageID); ok { s.upsertChannelDialogLocked(member.UserID, channel, msg, false) @@ -5142,25 +5140,6 @@ func (s *ChannelStore) nextChannelPtsNLocked(channelID int64, count int) int { return s.ptsSeq[channelID] } -func (s *ChannelStore) appendParticipantEventLocked(channel domain.Channel, actorUserID int64, previous, participant domain.ChannelMember, date int) domain.ChannelUpdateEvent { - pts := s.nextChannelPtsLocked(channel.ID) - channel.Pts = pts - s.channels[channel.ID] = channel - event := domain.ChannelUpdateEvent{ - ChannelID: channel.ID, - Type: domain.ChannelUpdateParticipant, - Pts: pts, - PtsCount: 1, - Date: date, - SenderUserID: actorUserID, - UserIDs: uniqueNonZeroInt64s(actorUserID, previous.UserID, previous.InviterUserID, participant.UserID, participant.InviterUserID), - Previous: previous, - Participant: participant, - } - s.events[channel.ID] = append(s.events[channel.ID], event) - return cloneChannelEvent(event) -} - func (s *ChannelStore) appendChannelServiceMessageLocked(channelID, senderUserID int64, date int, action domain.ChannelMessageAction) (domain.ChannelMessage, domain.ChannelUpdateEvent) { channel := s.channels[channelID] pts := s.nextChannelPtsLocked(channelID) @@ -5189,6 +5168,18 @@ func (s *ChannelStore) appendChannelServiceMessageLocked(channelID, senderUserID return msg, event } +func transientChannelParticipantEvent(channelID, actorUserID int64, previous, participant domain.ChannelMember, date int) domain.ChannelUpdateEvent { + return domain.ChannelUpdateEvent{ + ChannelID: channelID, + Type: domain.ChannelUpdateParticipant, + Date: date, + SenderUserID: actorUserID, + UserIDs: uniqueNonZeroInt64s(actorUserID, previous.UserID, previous.InviterUserID, participant.UserID, participant.InviterUserID), + Previous: previous, + Participant: participant, + } +} + func (s *ChannelStore) channelForMemberLocked(userID, channelID int64) (domain.Channel, error) { channel, _, err := s.channelAndMemberLocked(userID, channelID) return channel, err diff --git a/internal/store/memory/channel_test.go b/internal/store/memory/channel_test.go index 7e2fae39..2e90eb93 100644 --- a/internal/store/memory/channel_test.go +++ b/internal/store/memory/channel_test.go @@ -39,6 +39,69 @@ func TestChannelRealtimeRecipientsAreCapped(t *testing.T) { } } +func TestChannelAdminAndBanDoNotAdvanceChannelPts(t *testing.T) { + ctx := context.Background() + store := NewChannelStore() + created, err := store.CreateChannel(ctx, domain.CreateChannelRequest{ + CreatorUserID: 1, + Title: "participant state no pts", + Megagroup: true, + MemberUserIDs: []int64{2}, + Date: 1_700_000_120, + }) + if err != nil { + t.Fatalf("create channel: %v", err) + } + channelID := created.Channel.ID + ptsFloor := created.Channel.Pts + + promoted, err := store.EditChannelAdmin(ctx, domain.EditChannelAdminRequest{ + UserID: 1, + ChannelID: channelID, + MemberID: 2, + AdminRights: domain.ChannelAdminRights{ + InviteUsers: true, + }, + Date: 1_700_000_121, + }) + if err != nil { + t.Fatalf("edit admin: %v", err) + } + if promoted.Event.Pts != 0 || promoted.Event.PtsCount != 0 || promoted.Channel.Pts != ptsFloor { + t.Fatalf("edit admin pts = event(%d,%d) channel %d, want unchanged %d", promoted.Event.Pts, promoted.Event.PtsCount, promoted.Channel.Pts, ptsFloor) + } + + banned, err := store.EditChannelBanned(ctx, domain.EditChannelBannedRequest{ + UserID: 1, + ChannelID: channelID, + Participant: domain.Peer{Type: domain.PeerTypeUser, ID: 2}, + BannedRights: domain.ChannelBannedRights{ + ViewMessages: true, + UntilDate: 1_700_001_121, + }, + Date: 1_700_000_122, + }) + if err != nil { + t.Fatalf("edit banned: %v", err) + } + if banned.Event.Pts != 0 || banned.Event.PtsCount != 0 || banned.Channel.Pts != ptsFloor { + t.Fatalf("edit banned pts = event(%d,%d) channel %d, want unchanged %d", banned.Event.Pts, banned.Event.PtsCount, banned.Channel.Pts, ptsFloor) + } + + diff, err := store.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: 1, + ChannelID: channelID, + Pts: ptsFloor, + Limit: 10, + }) + if err != nil { + t.Fatalf("list difference: %v", err) + } + if len(diff.Events) != 0 || diff.Pts != ptsFloor { + t.Fatalf("difference after participant state change = %+v, want no durable events at pts %d", diff, ptsFloor) + } +} + func TestPendingJoinRequestsSummaryAndInviteAdmins(t *testing.T) { ctx := context.Background() store := NewChannelStore() diff --git a/internal/store/postgres/channel.go b/internal/store/postgres/channel.go index c33104ec..500499b3 100644 --- a/internal/store/postgres/channel.go +++ b/internal/store/postgres/channel.go @@ -798,11 +798,9 @@ func (s *ChannelStore) EditChannelAdmin(ctx context.Context, req domain.EditChan return domain.EditChannelAdminResult{}, fmt.Errorf("begin edit channel admin: %w", err) } committed := false - var reserved []reservedChannelPts defer func() { if !committed { _ = tx.Rollback(ctx) - s.recordChannelPtsGaps(ctx, reserved, req.Date) } }() channel, actor, err := s.getChannelForMember(ctx, tx, req.UserID, req.ChannelID) @@ -873,10 +871,7 @@ func (s *ChannelStore) EditChannelAdmin(ctx context.Context, req domain.EditChan if err != nil { return domain.EditChannelAdminResult{}, err } - event, channel, err := s.insertParticipantEventTx(ctx, tx, channel, req.UserID, previous, member, req.Date, &reserved) - if err != nil { - return domain.EditChannelAdminResult{}, err - } + event := transientChannelParticipantEvent(channel.ID, req.UserID, previous, member, req.Date) msg, _ := s.getChannelMessage(ctx, tx, req.ChannelID, channel.TopMessageID) if err := upsertChannelDialogTx(ctx, tx, member.UserID, channel, msg, member.ReadInboxMaxID, member.ReadOutboxMaxID); err != nil { return domain.EditChannelAdminResult{}, err @@ -906,11 +901,9 @@ func (s *ChannelStore) EditChannelBanned(ctx context.Context, req domain.EditCha return domain.EditChannelBannedResult{}, fmt.Errorf("begin edit channel banned: %w", err) } committed := false - var reserved []reservedChannelPts defer func() { if !committed { _ = tx.Rollback(ctx) - s.recordChannelPtsGaps(ctx, reserved, req.Date) } }() channel, actor, err := s.getChannelForMember(ctx, tx, req.UserID, req.ChannelID) @@ -979,10 +972,7 @@ func (s *ChannelStore) EditChannelBanned(ctx context.Context, req domain.EditCha if err != nil { return domain.EditChannelBannedResult{}, err } - event, channel, err := s.insertParticipantEventTx(ctx, tx, channel, req.UserID, previous, member, req.Date, &reserved) - if err != nil { - return domain.EditChannelBannedResult{}, err - } + event := transientChannelParticipantEvent(channel.ID, req.UserID, previous, member, req.Date) if member.Status == domain.ChannelMemberActive { msg, _ := s.getChannelMessage(ctx, tx, req.ChannelID, channel.TopMessageID) if err := upsertChannelDialogTx(ctx, tx, member.UserID, channel, msg, member.ReadInboxMaxID, member.ReadOutboxMaxID); err != nil { @@ -8025,31 +8015,16 @@ func (s *ChannelStore) insertServiceMessage(ctx context.Context, tx pgx.Tx, chan return msg, event, nil } -func (s *ChannelStore) insertParticipantEventTx(ctx context.Context, tx pgx.Tx, channel domain.Channel, actorUserID int64, previous, participant domain.ChannelMember, date int, reserved *[]reservedChannelPts) (domain.ChannelUpdateEvent, domain.Channel, error) { - pts, err := s.pts.NextChannelPts(ctx, channel.ID) - if err != nil { - return domain.ChannelUpdateEvent{}, channel, fmt.Errorf("allocate channel participant pts: %w", err) - } - reserveChannelPts(reserved, channel.ID, pts, 1) - event := domain.ChannelUpdateEvent{ - ChannelID: channel.ID, +func transientChannelParticipantEvent(channelID, actorUserID int64, previous, participant domain.ChannelMember, date int) domain.ChannelUpdateEvent { + return domain.ChannelUpdateEvent{ + ChannelID: channelID, Type: domain.ChannelUpdateParticipant, - Pts: pts, - PtsCount: 1, Date: date, SenderUserID: actorUserID, UserIDs: uniqueNonZeroInt64s(actorUserID, previous.UserID, previous.InviterUserID, participant.UserID, participant.InviterUserID), Previous: previous, Participant: participant, } - if err := insertChannelEventTx(ctx, tx, event); err != nil { - return domain.ChannelUpdateEvent{}, channel, err - } - if _, err := tx.Exec(ctx, `UPDATE channels SET pts = $2, updated_at = now() WHERE id = $1`, channel.ID, pts); err != nil { - return domain.ChannelUpdateEvent{}, channel, fmt.Errorf("update channel participant pts: %w", err) - } - channel.Pts = pts - return event, channel, nil } func (s *ChannelStore) deleteChannelMessagesTx(ctx context.Context, tx pgx.Tx, channel domain.Channel, member domain.ChannelMember, ids []int, actorUserID int64, date int, reserved *[]reservedChannelPts) ([]int, domain.ChannelUpdateEvent, domain.Channel, error) { diff --git a/internal/store/postgres/channel_integration_test.go b/internal/store/postgres/channel_integration_test.go index 785a628a..90388029 100644 --- a/internal/store/postgres/channel_integration_test.go +++ b/internal/store/postgres/channel_integration_test.go @@ -689,7 +689,8 @@ func TestChannelStoreJoinRejectsKickedMember(t *testing.T) { t.Fatalf("create channel: %v", err) } channelID = created.Channel.ID - if _, err := channels.EditChannelBanned(ctx, domain.EditChannelBannedRequest{ + ptsFloor := created.Channel.Pts + banned, err := channels.EditChannelBanned(ctx, domain.EditChannelBannedRequest{ UserID: owner.ID, ChannelID: channelID, Participant: domain.Peer{Type: domain.PeerTypeUser, ID: member.ID}, @@ -698,9 +699,25 @@ func TestChannelStoreJoinRejectsKickedMember(t *testing.T) { UntilDate: 1700001300, }, Date: 1700000306, - }); err != nil { + }) + if err != nil { t.Fatalf("kick member: %v", err) } + if banned.Event.Pts != 0 || banned.Event.PtsCount != 0 || banned.Channel.Pts != ptsFloor { + t.Fatalf("kick affected channel pts = event(%d,%d) channel %d, want no pts advance from %d", banned.Event.Pts, banned.Event.PtsCount, banned.Channel.Pts, ptsFloor) + } + banDiff, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: owner.ID, + ChannelID: channelID, + Pts: ptsFloor, + Limit: 10, + }) + if err != nil { + t.Fatalf("difference after kick: %v", err) + } + if len(banDiff.Events) != 0 || banDiff.Pts != ptsFloor { + t.Fatalf("difference after kick = %+v, want no durable participant event at pts %d", banDiff, ptsFloor) + } if _, err := channels.JoinChannel(ctx, channelID, member.ID, 1700000307); !errors.Is(err, domain.ErrChannelUserBanned) { t.Fatalf("kicked JoinChannel err = %v, want ErrChannelUserBanned", err) } @@ -1292,6 +1309,7 @@ func TestChannelStoreDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { t.Fatalf("create channel: %v", err) } channelID = created.Channel.ID + ptsFloor := created.Channel.Pts promoted, err := channels.EditChannelAdmin(ctx, domain.EditChannelAdminRequest{ UserID: owner.ID, ChannelID: channelID, @@ -1304,12 +1322,27 @@ func TestChannelStoreDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { if err != nil { t.Fatalf("edit admin: %v", err) } + if promoted.Event.Pts != 0 || promoted.Event.PtsCount != 0 || promoted.Channel.Pts != ptsFloor { + t.Fatalf("promote affected channel pts = event(%d,%d) channel %d, want no pts advance from %d", promoted.Event.Pts, promoted.Event.PtsCount, promoted.Channel.Pts, ptsFloor) + } + adminDiff, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ + UserID: member.ID, + ChannelID: channelID, + Pts: ptsFloor, + Limit: 10, + }) + if err != nil { + t.Fatalf("difference after promote: %v", err) + } + if len(adminDiff.Events) != 0 || adminDiff.Pts != ptsFloor { + t.Fatalf("difference after promote = %+v, want no durable participant event at pts %d", adminDiff, ptsFloor) + } joined, err := channels.JoinChannel(ctx, channelID, joiner.ID, 1700000352) if err != nil { t.Fatalf("join channel: %v", err) } - if len(joined.Members) != 1 || joined.Members[0].AvailableMinPts != promoted.Event.Pts { - t.Fatalf("joined members = %+v, want available_min_pts %d", joined.Members, promoted.Event.Pts) + if len(joined.Members) != 1 || joined.Members[0].AvailableMinPts != ptsFloor { + t.Fatalf("joined members = %+v, want available_min_pts %d", joined.Members, ptsFloor) } diff, err := channels.ListChannelDifference(ctx, domain.ChannelDifferenceRequest{ UserID: joiner.ID, @@ -1324,13 +1357,13 @@ func TestChannelStoreDifferenceStartsAtMemberAvailableMinPts(t *testing.T) { t.Fatalf("diff pts = %d, want current channel pts %d", diff.Pts, joined.Channel.Pts) } for _, msg := range diff.NewMessages { - if msg.Pts <= promoted.Event.Pts { - t.Fatalf("diff leaks pre-join message %+v at or before available_min_pts %d", msg, promoted.Event.Pts) + if msg.Pts <= ptsFloor { + t.Fatalf("diff leaks pre-join message %+v at or before available_min_pts %d", msg, ptsFloor) } } for _, event := range diff.OtherUpdates { - if event.Pts <= promoted.Event.Pts { - t.Fatalf("diff leaks pre-join event %+v at or before available_min_pts %d", event, promoted.Event.Pts) + if event.Pts <= ptsFloor { + t.Fatalf("diff leaks pre-join event %+v at or before available_min_pts %d", event, ptsFloor) } } }