diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index 9fd61765..3e8dc394 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -1360,7 +1360,8 @@ func run(logger *zap.Logger) error { // 私聊端对端加密(Secret Chat)握手状态机 + qts 投递队列(盲中继)。 secretChatStore := postgres.NewSecretChatStore(pool) encryptedQueueStore := postgres.NewEncryptedQueueStore(pool) - secretChatService := secretchatapp.NewService(secretChatStore, encryptedQueueStore) + channelParticipantQueueStore := postgres.NewChannelParticipantQueueStore(pool) + secretChatService := secretchatapp.NewService(secretChatStore, encryptedQueueStore, channelParticipantQueueStore) // Passkey:凭据持久化走 postgres;一次性挑战走进程内内存(短 TTL,与 QR 登录 token // 同属进程内一次性凭据,不跨实例)。 passkeyStore := postgres.NewPasskeyStore(pool) diff --git a/deploy/migrations/20260915000001_channel_participant_event_queue.down.sql b/deploy/migrations/20260915000001_channel_participant_event_queue.down.sql new file mode 100644 index 00000000..e2783ef8 --- /dev/null +++ b/deploy/migrations/20260915000001_channel_participant_event_queue.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS public.channel_participant_event_queue; diff --git a/deploy/migrations/20260915000001_channel_participant_event_queue.up.sql b/deploy/migrations/20260915000001_channel_participant_event_queue.up.sql new file mode 100644 index 00000000..144e1697 --- /dev/null +++ b/deploy/migrations/20260915000001_channel_participant_event_queue.up.sql @@ -0,0 +1,21 @@ +-- Durable per-device qts queue for channel-membership self-notifications +-- (kick/ban/promote/demote/ownership transfer). Shares the qts number space +-- in secret_qts_watermarks with encrypted_message_queue: a device has one +-- qts sequence, and updateChannelParticipant rides it per the MTProto spec, +-- so a disconnected device can recover a missed kick/ban via +-- updates.getDifference instead of losing it once its live push is missed. +CREATE TABLE public.channel_participant_event_queue ( + receiver_auth_key_id bigint NOT NULL, + qts integer NOT NULL, + receiver_user_id bigint NOT NULL, + channel_id bigint NOT NULL, + actor_user_id bigint NOT NULL, + date integer NOT NULL, + previous_participant jsonb, + new_participant jsonb, + acked boolean DEFAULT false NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + PRIMARY KEY (receiver_auth_key_id, qts) +); + +CREATE INDEX idx_cpeq_unacked ON public.channel_participant_event_queue USING btree (receiver_auth_key_id, qts) WHERE (acked = false); diff --git a/internal/app/secretchat/service.go b/internal/app/secretchat/service.go index 747aef51..51b5f9d0 100644 --- a/internal/app/secretchat/service.go +++ b/internal/app/secretchat/service.go @@ -18,11 +18,19 @@ import ( type Service struct { store store.SecretChatStore queue store.EncryptedQueueStore + // channelParticipants is the per-device qts queue for channel-membership + // self-notifications (kick/ban/promote/transfer). It shares its qts number + // space with queue (one qts sequence per device, per the MTProto spec for + // updateChannelParticipant) but is otherwise unrelated to secret chats; + // it lives here only because this Service already owns the device qts + // watermark plumbing. + channelParticipants store.ChannelParticipantQueueStore } -// NewService 创建密聊服务。 -func NewService(st store.SecretChatStore, queue store.EncryptedQueueStore) *Service { - return &Service{store: st, queue: queue} +// NewService 创建密聊服务。channelParticipants may be nil in tests that don't +// exercise channel-membership self-notifications. +func NewService(st store.SecretChatStore, queue store.EncryptedQueueStore, channelParticipants store.ChannelParticipantQueueStore) *Service { + return &Service{store: st, queue: queue, channelParticipants: channelParticipants} } // RequestEncryption 受理 requestEncryption:校验 g_a → 校验 random_id/chat_id 全局唯一性 → 分配双 @@ -242,6 +250,34 @@ func (s *Service) AckQueue(ctx context.Context, deviceAuthKeyID int64, maxQts in return s.queue.AckEncryptedMessages(ctx, deviceAuthKeyID, maxQts) } +// AppendChannelParticipantEvent durably enqueues a channel-membership +// self-notification for one device, reserving that device's next qts. +func (s *Service) AppendChannelParticipantEvent(ctx context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) { + if s.channelParticipants == nil || ev.ReceiverAuthKeyID == 0 { + return domain.DeviceChannelParticipantEvent{}, nil + } + return s.channelParticipants.AppendChannelParticipantEvent(ctx, ev) +} + +// ListChannelParticipantEventsSince returns a device's channel-membership +// self-notifications with qts > sinceQts (getDifference catch-up). +func (s *Service) ListChannelParticipantEventsSince(ctx context.Context, deviceAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) { + if s.channelParticipants == nil || deviceAuthKeyID == 0 { + return nil, nil + } + return s.channelParticipants.ListChannelParticipantEventsSince(ctx, deviceAuthKeyID, sinceQts, limit) +} + +// AckChannelParticipantEvents marks a device's channel-membership +// self-notifications up to maxQts as delivered (GC only; confirmed_qts itself +// advances via AckQueue against the shared watermark). +func (s *Service) AckChannelParticipantEvents(ctx context.Context, deviceAuthKeyID int64, maxQts int) error { + if s.channelParticipants == nil || deviceAuthKeyID == 0 || maxQts <= 0 { + return nil + } + return s.channelParticipants.AckChannelParticipantEvents(ctx, deviceAuthKeyID, maxQts) +} + // RecordEncryptionEvent 写入 durable updateEncryption 状态事件(离线补偿)。 // targetAuthKeyID=0 表示账号级(建链前邀请/撤回对 target 所有设备可见),非 0 表示 // 绑定设备定向。投递时按 secret_chats 权威态重建(不固化快照)。 diff --git a/internal/app/secretchat/service_test.go b/internal/app/secretchat/service_test.go index eb52d067..d79ee873 100644 --- a/internal/app/secretchat/service_test.go +++ b/internal/app/secretchat/service_test.go @@ -23,7 +23,8 @@ func validGA() []byte { func newTestService() (*Service, *memory.SecretChatStore) { st := memory.NewSecretChatStore() - return NewService(st, memory.NewEncryptedQueueStore()), st + queue := memory.NewEncryptedQueueStore() + return NewService(st, queue, queue), st } const ( diff --git a/internal/domain/channel_participant_queue.go b/internal/domain/channel_participant_queue.go new file mode 100644 index 00000000..164f9595 --- /dev/null +++ b/internal/domain/channel_participant_queue.go @@ -0,0 +1,19 @@ +package domain + +// DeviceChannelParticipantEvent is a durable, per-device queued entry recording +// that the receiving account's participant status in a channel changed +// (kick/ban/promote/demote/ownership transfer). It rides the same qts sequence +// as SecretChatMessage (one counter per device, shared with secret chats per +// the MTProto spec for updateChannelParticipant), so a device that misses the +// live push can recover the event via updates.getDifference instead of never +// learning it was removed. +type DeviceChannelParticipantEvent struct { + ReceiverAuthKeyID int64 + ReceiverUserID int64 + Qts int + ChannelID int64 + ActorUserID int64 + Date int + Previous ChannelMember + Participant ChannelMember +} diff --git a/internal/rpc/channels_legacy_chat.go b/internal/rpc/channels_legacy_chat.go index 35e37622..52392cb2 100644 --- a/internal/rpc/channels_legacy_chat.go +++ b/internal/rpc/channels_legacy_chat.go @@ -383,6 +383,8 @@ func (r *Router) onMessagesEditChatCreator(ctx context.Context, req *tg.Messages } r.invalidateChannelFullBotInfoCacheForChannel(res.Channel.ID) r.addOnlineChannelMemberships(res.Channel.ID, res.OldOwner.UserID, res.NewOwner.UserID) + r.deliverChannelParticipantSelfEvent(ctx, res.OldOwner.UserID, userID, res.Channel, res.PreviousOwner, res.OldOwner, res.Date) + r.deliverChannelParticipantSelfEvent(ctx, res.NewOwner.UserID, userID, res.Channel, res.PreviousNewOwner, res.NewOwner, res.Date) cache := newViewerPeerCache(r) updates := r.channelOwnershipTransferUpdatesWithPeerCache(ctx, userID, userID, res, cache) r.pushChannelUpdates(ctx, userID, res.Channel.ID, res.Recipients, func(viewerUserID int64) *tg.Updates { diff --git a/internal/rpc/channels_members.go b/internal/rpc/channels_members.go index 087352ec..409af5e4 100644 --- a/internal/rpc/channels_members.go +++ b/internal/rpc/channels_members.go @@ -3,6 +3,7 @@ package rpc import ( "context" "errors" + "github.com/iamxvbaba/td/proto" "github.com/iamxvbaba/td/tg" "github.com/iamxvbaba/td/tgerr" "go.uber.org/zap" @@ -451,6 +452,58 @@ func (r *Router) recordChannelStateForUser(ctx context.Context, userID, channelI } } +// deliverChannelParticipantSelfEvent durably enqueues, and best-effort live +// pushes, a correctly qts-numbered updateChannelParticipant to every device of +// the affected (non-actor) user. updateChannelParticipant carries the account's +// qts per the MTProto spec; a client discards any update whose qts isn't a +// proper increment over its own, so the generic pts-based fanout in +// pushChannelUpdates -- which always sends qts=0 -- is silently ignored by +// real clients for this specific update type. This is the only path that +// actually reaches the affected user's own client, both live and (via +// AppendChannelParticipantEvent's durable queue, replayed by +// updates.getDifference) after being offline when it happened. +func (r *Router) deliverChannelParticipantSelfEvent(ctx context.Context, targetUserID, actorUserID int64, channel domain.Channel, previous, participant domain.ChannelMember, date int) { + if r.deps.SecretChats == nil || r.deps.Auth == nil || targetUserID == 0 || targetUserID == actorUserID { + return + } + auths, err := r.deps.Auth.ListAuthorizations(ctx, targetUserID) + if err != nil || len(auths) == 0 { + return + } + targeted, canTarget := r.deps.Sessions.(AuthKeyTargetedSessionBinder) + for _, a := range auths { + deviceAuthKeyID := businessAuthKeyInt64(a.AuthKeyID) + if deviceAuthKeyID == 0 { + continue + } + stored, err := r.deps.SecretChats.AppendChannelParticipantEvent(ctx, domain.DeviceChannelParticipantEvent{ + ReceiverAuthKeyID: deviceAuthKeyID, + ReceiverUserID: targetUserID, + ChannelID: channel.ID, + ActorUserID: actorUserID, + Date: date, + Previous: previous, + Participant: participant, + }) + if err != nil || stored.Qts == 0 { + continue + } + if !canTarget { + continue + } + upd := r.channelParticipantUpdates(ctx, targetUserID, actorUserID, channel, previous, participant, date) + if upd == nil { + continue + } + for _, u := range upd.Updates { + if uc, ok := u.(*tg.UpdateChannelParticipant); ok { + uc.Qts = stored.Qts + } + } + _, _ = targeted.PushToUserAuthKey(ctx, targetUserID, deviceAuthKeyBytes(deviceAuthKeyID), proto.MessageFromServer, upd) + } +} + func (r *Router) onChannelsEditAdmin(ctx context.Context, req *tg.ChannelsEditAdminRequest) (tg.UpdatesClass, error) { if r.deps.Channels == nil && r.deps.Communities == nil { return nil, notImplementedErr() @@ -511,6 +564,7 @@ func (r *Router) onChannelsEditAdmin(ctx context.Context, req *tg.ChannelsEditAd } else { r.removeOnlineChannelMemberships(res.Channel.ID, res.Participant.UserID) } + r.deliverChannelParticipantSelfEvent(ctx, res.Participant.UserID, userID, res.Channel, res.Previous, res.Participant, res.Date) cache := newViewerPeerCache(r) updates := r.channelParticipantUpdatesWithPeerCache(ctx, userID, userID, res.Channel, res.Previous, res.Participant, res.Date, cache) r.pushChannelUpdates(ctx, userID, res.Channel.ID, res.Recipients, func(viewerUserID int64) *tg.Updates { @@ -557,6 +611,7 @@ func (r *Router) onChannelsEditBanned(ctx context.Context, req *tg.ChannelsEditB if res.Participant.Status == domain.ChannelMemberKicked && res.Previous.Status == domain.ChannelMemberActive { r.recordChannelStateForUser(ctx, res.Participant.UserID, res.Channel.ID, false) } + r.deliverChannelParticipantSelfEvent(ctx, res.Participant.UserID, userID, res.Channel, res.Previous, res.Participant, res.Date) cache := newViewerPeerCache(r) build := func(viewerUserID int64) *tg.Updates { updates := r.channelParticipantUpdatesWithPeerCache(ctx, viewerUserID, userID, res.Channel, res.Previous, res.Participant, res.Date, cache) diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index b3bf58d4..03a2a111 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -1237,6 +1237,17 @@ type SecretChatService interface { MarkStateEventsDelivered(ctx context.Context, deviceAuthKeyID int64, eventIDs []int64) error PutEncryptedFile(ctx context.Context, ownerUserID int64, ref domain.EncryptedFileRef) error GetEncryptedFile(ctx context.Context, id, accessHash int64) (domain.EncryptedFileRef, bool, error) + + // AppendChannelParticipantEvent durably enqueues a channel-membership + // self-notification (kick/ban/promote/transfer) for one device, reserving + // that device's next qts from the same watermark as secret chat messages. + AppendChannelParticipantEvent(ctx context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) + // ListChannelParticipantEventsSince returns a device's channel-membership + // self-notifications with qts > sinceQts (getDifference catch-up). + ListChannelParticipantEventsSince(ctx context.Context, deviceAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) + // AckChannelParticipantEvents marks a device's channel-membership + // self-notifications up to maxQts as delivered. + AckChannelParticipantEvents(ctx context.Context, deviceAuthKeyID int64, maxQts int) error } // PhoneService 抽象私聊 1:1 通话信令状态机(app/phone)。所有返回值都是状态快照; diff --git a/internal/rpc/encrypted_chats_test.go b/internal/rpc/encrypted_chats_test.go index 7f1aaf73..b213fe56 100644 --- a/internal/rpc/encrypted_chats_test.go +++ b/internal/rpc/encrypted_chats_test.go @@ -61,7 +61,7 @@ func newEncryptedFixture(t *testing.T) *encryptedFixture { queueStore := memory.NewEncryptedQueueStore() router := New(Config{}, Deps{ Users: appusers.NewService(userStore), - SecretChats: appsecret.NewService(secretStore, queueStore), + SecretChats: appsecret.NewService(secretStore, queueStore, queueStore), Updates: appupdates.NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore()), Files: &fakeFiles{}, Sessions: sessions, diff --git a/internal/rpc/encrypted_messages.go b/internal/rpc/encrypted_messages.go index 962ec22f..f2e7a997 100644 --- a/internal/rpc/encrypted_messages.go +++ b/internal/rpc/encrypted_messages.go @@ -3,6 +3,7 @@ package rpc import ( "context" "fmt" + "sort" "github.com/iamxvbaba/td/proto" "github.com/iamxvbaba/td/tg" @@ -103,6 +104,10 @@ func (r *Router) onMessagesReceivedQueue(ctx context.Context, maxQts int) ([]int if err := r.deps.SecretChats.AckQueue(ctx, deviceKey, maxQts); err != nil { return nil, internalErr() } + // Best-effort GC mark for the sibling channel-participant-event queue, + // which shares this device's qts sequence; a failure here doesn't affect + // correctness (only retention), so it isn't fatal to the RPC. + _ = r.deps.SecretChats.AckChannelParticipantEvents(ctx, deviceKey, maxQts) return []int64{}, nil } @@ -143,38 +148,84 @@ func (r *Router) deviceEncryptedQts(ctx context.Context) int { return qts } -// encryptedDifference 返回当前设备 qts > sinceQts 的连续前缀、推进后的 qts 与是否还有 -// 下一页。存储错误或 qts gap 必须 fail-fast,禁止越过缺口推进客户端水位。 -func (r *Router) encryptedDifference(ctx context.Context, sinceQts int) ([]tg.EncryptedMessageClass, int, bool, error) { +// encryptedDifference 返回当前设备 qts > sinceQts 的连续前缀(加密消息 + +// channel 成员关系自通知,二者共用同一设备 qts 序列,按 qts 归并后统一做缺口 +// 检查)、推进后的 qts 与是否还有下一页。存储错误或 qts gap 必须 fail-fast, +// 禁止越过缺口推进客户端水位。 +func (r *Router) encryptedDifference(ctx context.Context, sinceQts int) ([]tg.EncryptedMessageClass, []tg.UpdateClass, []int64, []int64, int, bool, error) { if r.deps.SecretChats == nil { - return nil, sinceQts, false, nil + return nil, nil, nil, nil, sinceQts, false, nil } deviceKey, ok := businessAuthKeyIDFrom(ctx) if !ok { - return nil, sinceQts, false, nil + return nil, nil, nil, nil, sinceQts, false, nil } msgs, err := r.deps.SecretChats.ListNewMessages(ctx, deviceKey, sinceQts, encryptedDifferencePageSize+1) if err != nil { - return nil, sinceQts, false, err + return nil, nil, nil, nil, sinceQts, false, err } - if len(msgs) == 0 { - return nil, sinceQts, false, nil + participantEvents, err := r.deps.SecretChats.ListChannelParticipantEventsSince(ctx, deviceKey, sinceQts, encryptedDifferencePageSize+1) + if err != nil { + return nil, nil, nil, nil, sinceQts, false, err } - partial := len(msgs) > encryptedDifferencePageSize - if partial { - msgs = msgs[:encryptedDifferencePageSize] + if len(msgs) == 0 && len(participantEvents) == 0 { + return nil, nil, nil, nil, sinceQts, false, nil } - out := make([]tg.EncryptedMessageClass, 0, len(msgs)) + items := make([]deviceQtsItem, 0, len(msgs)+len(participantEvents)) + for i := range msgs { + items = append(items, deviceQtsItem{qts: msgs[i].Qts, msg: &msgs[i]}) + } + for i := range participantEvents { + items = append(items, deviceQtsItem{qts: participantEvents[i].Qts, participant: &participantEvents[i]}) + } + sort.Slice(items, func(i, j int) bool { return items[i].qts < items[j].qts }) + partial := len(msgs) > encryptedDifferencePageSize || len(participantEvents) > encryptedDifferencePageSize + if len(items) > encryptedDifferencePageSize { + items = items[:encryptedDifferencePageSize] + partial = true + } + encMsgs := make([]tg.EncryptedMessageClass, 0, len(items)) + participantUpdates := make([]tg.UpdateClass, 0, len(items)) + participantPeerIDs := make([]int64, 0, len(items)) + participantChannelIDs := make([]int64, 0, len(items)) newQts := sinceQts - for i, m := range msgs { + for i, item := range items { expected := newQts + 1 - if m.Qts != expected { - return nil, sinceQts, false, fmt.Errorf("secret chat qts gap at index %d: got %d want %d", i, m.Qts, expected) + if item.qts != expected { + return nil, nil, nil, nil, sinceQts, false, fmt.Errorf("device qts gap at index %d: got %d want %d", i, item.qts, expected) } - out = append(out, tgEncryptedMessage(m)) - newQts = m.Qts + if item.msg != nil { + encMsgs = append(encMsgs, tgEncryptedMessage(*item.msg)) + } else { + ev := item.participant + update := &tg.UpdateChannelParticipant{ + ChannelID: ev.ChannelID, + Date: ev.Date, + ActorID: ev.ActorUserID, + UserID: ev.Participant.UserID, + Qts: ev.Qts, + } + if ev.Previous.UserID != 0 { + update.SetPrevParticipant(tgChannelParticipantForUpdate(ev.ReceiverUserID, ev.Previous)) + } + if ev.Participant.UserID != 0 { + update.SetNewParticipant(tgChannelParticipantForUpdate(ev.ReceiverUserID, ev.Participant)) + } + participantUpdates = append(participantUpdates, update, &tg.UpdateChannel{ChannelID: ev.ChannelID}) + participantPeerIDs = append(participantPeerIDs, ev.ActorUserID, ev.Participant.UserID, ev.Participant.InviterUserID, ev.Previous.UserID, ev.Previous.InviterUserID) + participantChannelIDs = append(participantChannelIDs, ev.ChannelID) + } + newQts = item.qts } - return out, newQts, partial, nil + return encMsgs, participantUpdates, participantPeerIDs, participantChannelIDs, newQts, partial, nil +} + +// deviceQtsItem is one entry in a device's merged qts stream: either an +// encrypted message or a channel-participant self-notification, never both. +type deviceQtsItem struct { + qts int + msg *domain.SecretChatMessage + participant *domain.DeviceChannelParticipantEvent } // injectEncryptedMessages 把加密消息与推进后的 qts 注入差分响应(按类型分别写 State / diff --git a/internal/rpc/receives_updates_mark_test.go b/internal/rpc/receives_updates_mark_test.go index 2cbd9357..d71372c9 100644 --- a/internal/rpc/receives_updates_mark_test.go +++ b/internal/rpc/receives_updates_mark_test.go @@ -285,7 +285,7 @@ func TestInvokeWithoutUpdatesBaselineCommitsResultAndSecretEventsWithoutSubscrib authKeyID := [8]byte{21} deviceKey := businessAuthKeyInt64(authKeyID) queue := memory.NewEncryptedQueueStore() - secret := appsecret.NewService(memory.NewSecretChatStore(), queue) + secret := appsecret.NewService(memory.NewSecretChatStore(), queue, queue) eventID, err := queue.AppendStateEvent(context.Background(), domain.EncryptedStateEvent{ TargetUserID: userID, ChatID: 77, diff --git a/internal/rpc/updates.go b/internal/rpc/updates.go index 52e9aa52..11607340 100644 --- a/internal/rpc/updates.go +++ b/internal/rpc/updates.go @@ -118,8 +118,9 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD break } } - // 密聊设备级 qts 消息(独立于账号级 pts 事件):按当前设备 req.Qts 补回。 - encMsgs, newQts, encryptedPartial, err := r.encryptedDifference(ctx, req.Qts) + // 密聊设备级 qts 消息 + channel 成员关系自通知(共用同一设备 qts 序列,独立于 + // 账号级 pts 事件):按当前设备 req.Qts 补回。 + encMsgs, participantUpdates, participantPeerIDs, participantChannelIDs, newQts, encryptedPartial, err := r.encryptedDifference(ctx, req.Qts) if err != nil { r.log.Error("load secret chat qts difference", zap.Error(err)) return nil, internalErr() @@ -130,6 +131,8 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD r.log.Error("load secret chat state difference", zap.Error(err)) return nil, internalErr() } + stateUpdates = append(stateUpdates, participantUpdates...) + statePeerUserIDs = append(statePeerUserIDs, participantPeerIDs...) // 账号 pts、设备 qts 和无序号状态事件任一被截断,都必须返回 differenceSlice。 st.Partial = st.Partial || encryptedPartial || encryptedStatePartial if !st.Partial && len(st.Events) == 0 && len(st.ChannelNudges) == 0 && len(encMsgs) == 0 && len(stateUpdates) == 0 { @@ -158,12 +161,40 @@ func (r *Router) onUpdatesGetDifference(ctx context.Context, req *tg.UpdatesGetD zap.Error(err)) return nil, internalErr() } + diff = r.injectChannelParticipantEventChats(ctx, userID, diff, participantChannelIDs) returnedCursor := st.State returnedCursor.Qts = newQts r.stageUpdatesBaselineAfterDelivery(ctx, userID, &returnedCursor, domain.UpdateStateCommitDeliveredOnly, stateEventIDs, true) return diff, nil } +// injectChannelParticipantEventChats resolves each channel referenced by a +// recovered channel-participant self-notification through GetChannels (which +// already projects Forbidden for a kicked/banned viewer, see +// tgChannelChatForView) and appends the resulting chat objects to diff -- +// the bare updateChannel nudge alone carries no chat payload, and the +// receiving client needs the full projection to apply the removal. +func (r *Router) injectChannelParticipantEventChats(ctx context.Context, userID int64, diff tg.UpdatesDifferenceClass, channelIDs []int64) tg.UpdatesDifferenceClass { + if r.deps.Channels == nil || len(channelIDs) == 0 { + return diff + } + views, err := r.deps.Channels.GetChannels(ctx, userID, channelIDs) + if err != nil || len(views) == 0 { + return diff + } + chats := make([]tg.ChatClass, 0, len(views)) + for _, view := range views { + chats = append(chats, tgChannelChatForView(userID, view)) + } + switch v := diff.(type) { + case *tg.UpdatesDifference: + v.Chats = appendUniqueTGChats(v.Chats, chats...) + case *tg.UpdatesDifferenceSlice: + v.Chats = appendUniqueTGChats(v.Chats, chats...) + } + return diff +} + func (r *Router) accountChannelDifferenceNudges(ctx context.Context, userID int64, sinceDate int) []domain.ChannelDifferenceNudge { if r.deps.Channels == nil || userID == 0 || sinceDate <= 0 { return nil diff --git a/internal/store/memory/secretchat.go b/internal/store/memory/secretchat.go index 06fabdbe..da23bc6d 100644 --- a/internal/store/memory/secretchat.go +++ b/internal/store/memory/secretchat.go @@ -114,17 +114,20 @@ func (s *SecretChatStore) ListActiveSecretChatsByAuthKey(_ context.Context, auth return out, nil } -// EncryptedQueueStore 是 store.EncryptedQueueStore 的进程内实现。 +// EncryptedQueueStore 是 store.EncryptedQueueStore 的进程内实现。也实现 +// store.ChannelParticipantQueueStore:两者共用同一个 reserved 水位 map(一台 +// 设备一条 qts 序列),与 postgres 两张表共用一张 secret_qts_watermarks 对齐。 type EncryptedQueueStore struct { - mu sync.Mutex - byDevice map[int64][]domain.SecretChatMessage // receiverAuthKeyID → qts 升序消息 - reserved map[int64]int - confirmed map[int64]int - dedup map[emqDedupKey]int // → qts - stateEvents []domain.EncryptedStateEvent - delivered map[int64]map[int64]bool // eventID → deviceAuthKeyID → true - nextEventID int64 - files map[int64]domain.EncryptedFileRef // file id → 快照 + mu sync.Mutex + byDevice map[int64][]domain.SecretChatMessage // receiverAuthKeyID → qts 升序消息 + byDeviceParticipant map[int64][]domain.DeviceChannelParticipantEvent // receiverAuthKeyID → qts 升序事件 + reserved map[int64]int + confirmed map[int64]int + dedup map[emqDedupKey]int // → qts + stateEvents []domain.EncryptedStateEvent + delivered map[int64]map[int64]bool // eventID → deviceAuthKeyID → true + nextEventID int64 + files map[int64]domain.EncryptedFileRef // file id → 快照 } type emqDedupKey struct { @@ -136,14 +139,56 @@ type emqDedupKey struct { // NewEncryptedQueueStore 创建内存实现。 func NewEncryptedQueueStore() *EncryptedQueueStore { return &EncryptedQueueStore{ - byDevice: make(map[int64][]domain.SecretChatMessage), - reserved: make(map[int64]int), - confirmed: make(map[int64]int), - dedup: make(map[emqDedupKey]int), - delivered: make(map[int64]map[int64]bool), + byDevice: make(map[int64][]domain.SecretChatMessage), + byDeviceParticipant: make(map[int64][]domain.DeviceChannelParticipantEvent), + reserved: make(map[int64]int), + confirmed: make(map[int64]int), + dedup: make(map[emqDedupKey]int), + delivered: make(map[int64]map[int64]bool), } } +func cloneChannelParticipantEvent(ev domain.DeviceChannelParticipantEvent) domain.DeviceChannelParticipantEvent { + return ev +} + +// AppendChannelParticipantEvent implements store.ChannelParticipantQueueStore, +// reserving from the same per-device qts counter as AppendEncryptedMessage. +func (s *EncryptedQueueStore) AppendChannelParticipantEvent(_ context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.reserved[ev.ReceiverAuthKeyID]++ + ev.Qts = s.reserved[ev.ReceiverAuthKeyID] + stored := cloneChannelParticipantEvent(ev) + s.byDeviceParticipant[ev.ReceiverAuthKeyID] = append(s.byDeviceParticipant[ev.ReceiverAuthKeyID], stored) + return cloneChannelParticipantEvent(stored), nil +} + +func (s *EncryptedQueueStore) ListChannelParticipantEventsSince(_ context.Context, receiverAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) { + s.mu.Lock() + defer s.mu.Unlock() + if limit <= 0 { + limit = 1000 + } + var out []domain.DeviceChannelParticipantEvent + for _, ev := range s.byDeviceParticipant[receiverAuthKeyID] { + if ev.Qts > sinceQts { + out = append(out, cloneChannelParticipantEvent(ev)) + if len(out) >= limit { + break + } + } + } + return out, nil +} + +// AckChannelParticipantEvents is a no-op in the memory store: unlike postgres +// it does no row-level GC, and confirmed_qts is already advanced by +// AckEncryptedMessages against the shared reserved/confirmed watermark. +func (s *EncryptedQueueStore) AckChannelParticipantEvents(_ context.Context, _ int64, _ int) error { + return nil +} + func cloneSecretMessage(m domain.SecretChatMessage) domain.SecretChatMessage { m.Bytes = append([]byte(nil), m.Bytes...) if m.File != nil { diff --git a/internal/store/postgres/channel_participant_queue.go b/internal/store/postgres/channel_participant_queue.go new file mode 100644 index 00000000..701e1f30 --- /dev/null +++ b/internal/store/postgres/channel_participant_queue.go @@ -0,0 +1,149 @@ +package postgres + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/jackc/pgx/v5" + + "telesrv/internal/domain" + "telesrv/internal/store/postgres/sqlcgen" +) + +// ChannelParticipantQueueStore is the PostgreSQL implementation of +// store.ChannelParticipantQueueStore. It reserves qts from the same +// secret_qts_watermarks table as EncryptedQueueStore (one qts sequence per +// device, shared across event kinds), keeping this table's own schema and +// queries entirely separate from the encrypted-message hot path. +type ChannelParticipantQueueStore struct { + db sqlcgen.DBTX +} + +// NewChannelParticipantQueueStore builds the store on a pgx pool or transaction. +func NewChannelParticipantQueueStore(db sqlcgen.DBTX) *ChannelParticipantQueueStore { + return &ChannelParticipantQueueStore{db: db} +} + +const channelParticipantEventColumns = `receiver_auth_key_id, qts, receiver_user_id, channel_id, actor_user_id, +date, previous_participant, new_participant` + +func scanChannelParticipantEvent(row rowScanner) (domain.DeviceChannelParticipantEvent, error) { + var ev domain.DeviceChannelParticipantEvent + var previousRaw, newRaw []byte + if err := row.Scan( + &ev.ReceiverAuthKeyID, &ev.Qts, &ev.ReceiverUserID, &ev.ChannelID, &ev.ActorUserID, + &ev.Date, &previousRaw, &newRaw, + ); err != nil { + return domain.DeviceChannelParticipantEvent{}, err + } + if len(previousRaw) > 0 { + if err := json.Unmarshal(previousRaw, &ev.Previous); err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("unmarshal previous participant: %w", err) + } + } + if len(newRaw) > 0 { + if err := json.Unmarshal(newRaw, &ev.Participant); err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("unmarshal new participant: %w", err) + } + } + return ev, nil +} + +func (s *ChannelParticipantQueueStore) begin(ctx context.Context, op string) (pgx.Tx, error) { + beginner, ok := s.db.(txBeginner) + if !ok { + return nil, fmt.Errorf("%s: db does not support transactions", op) + } + tx, err := beginner.Begin(ctx) + if err != nil { + return nil, fmt.Errorf("begin %s: %w", op, err) + } + return tx, nil +} + +// AppendChannelParticipantEvent reserves the receiving device's next qts +// (shared secret_qts_watermarks sequence) and writes the event, in one +// transaction so the device's qts never has a hole. +func (s *ChannelParticipantQueueStore) AppendChannelParticipantEvent(ctx context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) { + tx, err := s.begin(ctx, "append channel participant event") + if err != nil { + return domain.DeviceChannelParticipantEvent{}, err + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback(ctx) + } + }() + + var qts int + if err := tx.QueryRow(ctx, ` +INSERT INTO secret_qts_watermarks (auth_key_id, reserved_qts) +VALUES ($1, 1) +ON CONFLICT (auth_key_id) DO UPDATE SET reserved_qts = secret_qts_watermarks.reserved_qts + 1, updated_at = now() +RETURNING reserved_qts`, ev.ReceiverAuthKeyID).Scan(&qts); err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("reserve device qts: %w", err) + } + ev.Qts = qts + + previousRaw, err := marshalJSON(ev.Previous, "null") + if err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("marshal previous participant: %w", err) + } + newRaw, err := marshalJSON(ev.Participant, "null") + if err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("marshal new participant: %w", err) + } + if _, err := tx.Exec(ctx, ` +INSERT INTO channel_participant_event_queue (receiver_auth_key_id, qts, receiver_user_id, channel_id, actor_user_id, + date, previous_participant, new_participant) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + ev.ReceiverAuthKeyID, ev.Qts, ev.ReceiverUserID, ev.ChannelID, ev.ActorUserID, + ev.Date, previousRaw, newRaw); err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("insert channel participant event: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("commit append channel participant event: %w", err) + } + committed = true + return ev, nil +} + +func (s *ChannelParticipantQueueStore) ListChannelParticipantEventsSince(ctx context.Context, receiverAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) { + if limit <= 0 || limit > 1001 { + limit = 1000 + } + rows, err := s.db.Query(ctx, + `SELECT `+channelParticipantEventColumns+` FROM channel_participant_event_queue +WHERE receiver_auth_key_id = $1 AND qts > $2 ORDER BY qts ASC LIMIT $3`, + receiverAuthKeyID, sinceQts, limit) + if err != nil { + return nil, fmt.Errorf("list channel participant events: %w", err) + } + defer rows.Close() + var out []domain.DeviceChannelParticipantEvent + for rows.Next() { + ev, err := scanChannelParticipantEvent(rows) + if err != nil { + return nil, err + } + out = append(out, ev) + } + if err := rows.Err(); err != nil { + return nil, err + } + return out, nil +} + +func (s *ChannelParticipantQueueStore) AckChannelParticipantEvents(ctx context.Context, receiverAuthKeyID int64, maxQts int) error { + if maxQts <= 0 { + return nil + } + if _, err := s.db.Exec(ctx, ` +UPDATE channel_participant_event_queue SET acked = true +WHERE receiver_auth_key_id = $1 AND qts <= $2 AND NOT acked`, receiverAuthKeyID, maxQts); err != nil { + return fmt.Errorf("ack channel participant events: %w", err) + } + return nil +} diff --git a/internal/store/secretchat.go b/internal/store/secretchat.go index 480f19ca..bfcd40fc 100644 --- a/internal/store/secretchat.go +++ b/internal/store/secretchat.go @@ -59,3 +59,20 @@ type EncryptedQueueStore interface { // GetEncryptedFile 按 id + access_hash 回查文件快照(inputEncryptedFile 复用路径)。 GetEncryptedFile(ctx context.Context, id, accessHash int64) (domain.EncryptedFileRef, bool, error) } + +// ChannelParticipantQueueStore 持久化 channel 成员关系自通知(kick/ban/promote/ +// transfer)的设备级 qts 投递队列。updateChannelParticipant 走 qts 序列(MTProto +// 规范),与 EncryptedQueueStore 共用同一张 secret_qts_watermarks 水位表——每设备 +// 只有一条 qts 序列,两类事件按 qts 交错,通过各自独立的表分别落盘,读侧 +// (updates.getDifference)按 qts 归并成连续序列。 +type ChannelParticipantQueueStore interface { + // AppendChannelParticipantEvent 为接收设备分配下一个 qts 并写入队列,与 + // reserved_qts 推进同事务(与 AppendEncryptedMessage 共用同一水位表)。 + AppendChannelParticipantEvent(ctx context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) + // ListChannelParticipantEventsSince 返回接收设备 qts > sinceQts 的连续事件 + // (qts 升序,最多 limit)。 + ListChannelParticipantEventsSince(ctx context.Context, receiverAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) + // AckChannelParticipantEvents 标记 qts<=maxQts 行为 acked(confirmed_qts 由 + // AckEncryptedMessages 在同一张水位表上推进,这里只做本表的 GC 标记)。 + AckChannelParticipantEvents(ctx context.Context, receiverAuthKeyID int64, maxQts int) error +}