From 87cebec9bd0dd366d282263f43f62c35fd4bdd48 Mon Sep 17 00:00:00 2001 From: onysd Date: Mon, 7 Sep 2026 09:31:06 +0300 Subject: [PATCH] fix for muting an account --- ...ser_update_events_notify_settings.down.sql | 16 ++ ..._user_update_events_notify_settings.up.sql | 21 +++ internal/app/updates/service.go | 13 ++ internal/domain/update_event.go | 41 +++-- internal/rpc/account_notify.go | 17 +- internal/rpc/convert_updates.go | 9 + internal/rpc/deps.go | 1 + ...notify_settings_cross_device_repro_test.go | 101 +++++++++++ internal/rpc/rpc_testkit_updates_test.go | 6 + ...first_contact_getpeerdialogs_repro_test.go | 96 +++++++++++ .../notify_settings_event_roundtrip_test.go | 85 +++++++++ internal/store/postgres/update_event.go | 162 +++++++++++++----- 12 files changed, 503 insertions(+), 65 deletions(-) create mode 100644 deploy/migrations/20260907000001_user_update_events_notify_settings.down.sql create mode 100644 deploy/migrations/20260907000001_user_update_events_notify_settings.up.sql create mode 100644 internal/rpc/notify_settings_cross_device_repro_test.go create mode 100644 internal/store/postgres/first_contact_getpeerdialogs_repro_test.go create mode 100644 internal/store/postgres/notify_settings_event_roundtrip_test.go diff --git a/deploy/migrations/20260907000001_user_update_events_notify_settings.down.sql b/deploy/migrations/20260907000001_user_update_events_notify_settings.down.sql new file mode 100644 index 00000000..e7c7d77f --- /dev/null +++ b/deploy/migrations/20260907000001_user_update_events_notify_settings.down.sql @@ -0,0 +1,16 @@ +-- Revert to the constraint without 'notify_settings'. +ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; +ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( + (event_type)::text = ANY (ARRAY[ + 'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents', + 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', + 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', + 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', + 'peer_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', 'delete_messages', + 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', + 'channel_view_forum_as_messages', 'channel_state', + 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', + 'sent_story_reaction', 'new_story_reaction', 'noop', + 'read_channel_discussion_inbox', 'read_channel_discussion_outbox' + ]::text[]) +); diff --git a/deploy/migrations/20260907000001_user_update_events_notify_settings.up.sql b/deploy/migrations/20260907000001_user_update_events_notify_settings.up.sql new file mode 100644 index 00000000..a60f850e --- /dev/null +++ b/deploy/migrations/20260907000001_user_update_events_notify_settings.up.sql @@ -0,0 +1,21 @@ +-- Adds 'notify_settings' to the allow-list: account.updateNotifySettings for a specific +-- peer (the common "mute this chat" action) now records a durable user_update_events row +-- (see RecordNotifySettings) so another of the account's sessions picks up the mute via +-- getDifference/live dispatch instead of only the account's own best-effort push, which is +-- lost if that other session isn't connected at that exact instant. Without this entry the +-- INSERT fails the CHECK constraint and the whole RPC returns 500 INTERNAL_SERVER_ERROR. +ALTER TABLE public.user_update_events DROP CONSTRAINT IF EXISTS user_update_events_type_check; +ALTER TABLE public.user_update_events ADD CONSTRAINT user_update_events_type_check CHECK ( + (event_type)::text = ANY (ARRAY[ + 'new_message', 'read_history_inbox', 'read_history_outbox', 'read_message_contents', + 'edit_message', 'web_page', 'message_reactions', 'message_poll', 'draft_message', 'quick_replies', + 'new_quick_reply', 'delete_quick_reply', 'quick_reply_message', 'delete_quick_reply_messages', + 'contacts_reset', 'dialog_pinned', 'pinned_dialogs', 'pinned_messages', 'dialog_unread_mark', + 'peer_settings', 'notify_settings', 'peer_story_blocked', 'user_phone', 'user_emoji_status', + 'delete_messages', 'dialog_filter', 'dialog_filter_order', 'dialog_filters', 'folder_peers', + 'channel_view_forum_as_messages', 'channel_state', + 'saved_dialog_pinned', 'pinned_saved_dialogs', 'story', 'read_stories', + 'sent_story_reaction', 'new_story_reaction', 'noop', + 'read_channel_discussion_inbox', 'read_channel_discussion_outbox' + ]::text[]) +); diff --git a/internal/app/updates/service.go b/internal/app/updates/service.go index d18f820e..1ff6e881 100644 --- a/internal/app/updates/service.go +++ b/internal/app/updates/service.go @@ -680,6 +680,19 @@ func (s *Service) RecordPeerSettings(ctx context.Context, stateAuthKeyID [8]byte }, true, excludeSessionID) } +// RecordNotifySettings 记录 per-peer 通知设置变化（updateNotifySettings），走 durable +// outbox 而非旧的 best-effort 推送。 +func (s *Service) RecordNotifySettings(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, topicID int, settings domain.PeerNotifySettings, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { + sc := settings.Clone() + return s.recordEvent(ctx, stateAuthKeyID, excludeAuthKeyID, userID, domain.UpdateEvent{ + Type: domain.UpdateEventNotifySettings, + Peer: peer, + TopMsgID: topicID, + NotifyPeerSettings: &sc, + PtsCount: 1, + }, true, excludeSessionID) +} + // RecordPeerStoryBlocked 记录当前账号 story blocklist 对某个 peer 的可见状态变化。 func (s *Service) RecordPeerStoryBlocked(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, blocked bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { return s.recordEvent(ctx, stateAuthKeyID, excludeAuthKeyID, userID, domain.UpdateEvent{ diff --git a/internal/domain/update_event.go b/internal/domain/update_event.go index c3621fa8..1ccc6899 100644 --- a/internal/domain/update_event.go +++ b/internal/domain/update_event.go @@ -28,6 +28,11 @@ const ( UpdateEventPinnedDialogs UpdateEventType = "pinned_dialogs" UpdateEventDialogUnreadMark UpdateEventType = "dialog_unread_mark" UpdateEventPeerSettings UpdateEventType = "peer_settings" + // UpdateEventNotifySettings 映射 updateNotifySettings（静音/预览等设置变化）。历史上 + // account.updateNotifySettings 只走 best-effort 推送，对方另一台离线设备错过就永远 + // 追不上，继续用旧设置弹通知直到重启。这里改走 durable dispatch，可经 getDifference + // 恢复。仅覆盖 NotifyScopePeer（静音某个具体会话，最常见场景）。 + UpdateEventNotifySettings UpdateEventType = "notify_settings" UpdateEventPeerStoryBlocked UpdateEventType = "peer_story_blocked" // UpdateEventUserPhone 只用于读取历史版本已落库的 updateUserPhone 事件。 // TL 构造器不携 pts，当前写路径禁止再产生该 event。 @@ -74,23 +79,24 @@ const ( // UpdateEvent 是账号视角的增量事件，按 user_id + pts 顺序持久化。 type UpdateEvent struct { - UserID int64 - Type UpdateEventType - Pts int - PtsCount int - Date int - Message Message - Story Story - Peer Peer - Peers []Peer - Bool bool - Phone string - EmojiStatus UserEmojiStatus - Settings PeerSettings - MessageIDs []int - MaxID int - StillUnreadCount int - ChannelPts int + UserID int64 + Type UpdateEventType + Pts int + PtsCount int + Date int + Message Message + Story Story + Peer Peer + Peers []Peer + Bool bool + Phone string + EmojiStatus UserEmojiStatus + Settings PeerSettings + NotifyPeerSettings *PeerNotifySettings + MessageIDs []int + MaxID int + StillUnreadCount int + ChannelPts int // TopMsgID 仅 forum per-topic 已读事件（read_channel_discussion_*）使用：承载话题 id // （General=1），与 MaxID(=read_max_id) 一起映射 updateReadChannelDiscussionInbox/Outbox。 TopMsgID int @@ -134,6 +140,7 @@ func (e UpdateEvent) LacksWirePts() bool { UpdateEventPinnedSavedDialogs, UpdateEventDialogUnreadMark, UpdateEventPeerSettings, + UpdateEventNotifySettings, UpdateEventPeerStoryBlocked, UpdateEventUserPhone, UpdateEventUserEmojiStatus, diff --git a/internal/rpc/account_notify.go b/internal/rpc/account_notify.go index b6c7abda..3555b6a7 100644 --- a/internal/rpc/account_notify.go +++ b/internal/rpc/account_notify.go @@ -94,7 +94,22 @@ func (r *Router) onAccountUpdateNotifySettings(ctx context.Context, req *tg.Acco } r.notifySettings.Delete(userID) } - // 推 updateNotifySettings 给本人其它在线设备（多设备静音同步）。 + // 推 updateNotifySettings 给本人其它在线设备（多设备静音同步）。NotifyScopePeer（静音/ + // 取消静音某一具体会话，最常见场景）走 durable outbox：对方另一台设备当时不在线也能 + // 在重连后经 getDifference 追上最新静音状态，而不是停留在旧设置继续弹通知直到重启 + // app 才刷新。其余全局作用域（NotifyUsers/Chats/Broadcasts）仍走旧的 best-effort 推送。 + if scope.Kind == domain.NotifyScopePeer && r.deps.Updates != nil { + authKeyID, _ := AuthKeyIDFrom(ctx) + sessionID, _ := SessionIDFrom(ctx) + event, _, err := r.deps.Updates.RecordNotifySettings(ctx, authKeyID, userID, scope.Peer, scope.TopicID, settings, rawAuthKeyIDForOrigin(ctx), sessionID) + if err != nil { + return false, internalErr() + } + if sessionID != 0 { + r.bookkeepAuxPtsForCurrentSession(ctx, event) + } + return true, nil + } r.pushUserUpdates(ctx, userID, &tg.Updates{ Updates: []tg.UpdateClass{&tg.UpdateNotifySettings{ Peer: tgNotifyPeer(scope), diff --git a/internal/rpc/convert_updates.go b/internal/rpc/convert_updates.go index 213767c8..4e6cc0af 100644 --- a/internal/rpc/convert_updates.go +++ b/internal/rpc/convert_updates.go @@ -406,6 +406,15 @@ func tgOtherUpdateFromEvent(event domain.UpdateEvent) tg.UpdateClass { return nil } return &tg.UpdatePeerSettings{Peer: peer, Settings: tgPeerSettings(event.Settings)} + case domain.UpdateEventNotifySettings: + if event.Peer.Type != domain.PeerTypeCommunity && tgPeer(event.Peer) == nil { + return nil + } + if event.NotifyPeerSettings == nil { + return nil + } + scope := domain.NotifyScope{Kind: domain.NotifyScopePeer, Peer: event.Peer, TopicID: event.TopMsgID} + return &tg.UpdateNotifySettings{Peer: tgNotifyPeer(scope), NotifySettings: *tgPeerNotifySettings(event.NotifyPeerSettings)} case domain.UpdateEventPeerStoryBlocked: peer := tgPeer(event.Peer) if peer == nil { diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index 6642b5dd..b3bf58d4 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -585,6 +585,7 @@ type UpdatesService interface { RecordPinnedSavedDialogs(ctx context.Context, stateAuthKeyID [8]byte, userID int64, order []domain.Peer, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordDialogUnreadMark(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, unread bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordPeerSettings(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, settings domain.PeerSettings, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) + RecordNotifySettings(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, topicID int, settings domain.PeerNotifySettings, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordPeerStoryBlocked(ctx context.Context, stateAuthKeyID [8]byte, userID int64, peer domain.Peer, blocked bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordDialogFilter(ctx context.Context, stateAuthKeyID [8]byte, userID int64, folderID int, folder *domain.DialogFolder, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) RecordDialogFilterOrder(ctx context.Context, stateAuthKeyID [8]byte, userID int64, order []int, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) diff --git a/internal/rpc/notify_settings_cross_device_repro_test.go b/internal/rpc/notify_settings_cross_device_repro_test.go new file mode 100644 index 00000000..0fce8105 --- /dev/null +++ b/internal/rpc/notify_settings_cross_device_repro_test.go @@ -0,0 +1,101 @@ +package rpc + +import ( + "context" + "testing" + + "github.com/iamxvbaba/td/clock" + "github.com/iamxvbaba/td/tg" + "go.uber.org/zap/zaptest" + + appaccount "telesrv/internal/app/account" + appupdates "telesrv/internal/app/updates" + appusers "telesrv/internal/app/users" + "telesrv/internal/domain" + "telesrv/internal/store/memory" +) + +// TestNotifySettingsMuteFromOneDeviceReachesAnotherViaDifference reproduces the reported +// symptom: muting a chat from one device (PC) does not apply on another device of the SAME +// account (phone), even after a full app restart (which does updates.getDifference, not a +// live push). This test drives the real durable pipeline end to end (real appupdates.Service + +// appaccount.Service, in-memory stores) instead of the captureUpdates fake, to catch anything +// the fake's simplistic recording hides. +func TestNotifySettingsMuteFromOneDeviceReachesAnotherViaDifference(t *testing.T) { + passwordStore := memory.NewPasswordStore() + updateStateStore := memory.NewUpdateStateStore() + updateEventStore := memory.NewUpdateEventStore() + userStore := memory.NewUserStore() + r := New(Config{}, Deps{ + Account: appaccount.NewService(passwordStore, appaccount.WithNotifySettings(passwordStore)), + Updates: appupdates.NewService(updateStateStore, updateEventStore), + Users: appusers.NewService(userStore), + }, zaptest.NewLogger(t), clock.System) + + peerUser, err := userStore.Create(context.Background(), domain.User{AccessHash: 44, Phone: "15550009001", FirstName: "Peer"}) + if err != nil { + t.Fatalf("create peer user: %v", err) + } + ownerUser, err := userStore.Create(context.Background(), domain.User{AccessHash: 45, Phone: "15550009002", FirstName: "Owner"}) + if err != nil { + t.Fatalf("create owner user: %v", err) + } + peerID := peerUser.ID + owner := ownerUser.ID + + // Phone is "at" pts=0 (fresh install / just opened), before the PC mutes anything. + phoneCtx := WithAuthKeyID(WithUserID(context.Background(), owner), [8]byte{2, 2, 2}) + baseline, err := r.onUpdatesGetDifference(phoneCtx, &tg.UpdatesGetDifferenceRequest{Pts: 0}) + if err != nil { + t.Fatalf("phone baseline getDifference: %v", err) + } + t.Logf("phone baseline = %#v", baseline) + var baselinePts int + switch d := baseline.(type) { + case *tg.UpdatesDifference: + baselinePts = d.State.Pts + case *tg.UpdatesDifferenceEmpty: + baselinePts = 0 + case *tg.UpdatesDifferenceSlice: + baselinePts = d.IntermediateState.Pts + default: + t.Fatalf("unexpected baseline type %T", baseline) + } + + // PC (different session, same user) mutes a specific peer. + pcCtx := WithSessionID(WithAuthKeyID(WithUserID(context.Background(), owner), [8]byte{1, 1, 1}), 77) + in := tg.InputPeerNotifySettings{} + in.SetMuteUntil(2000000000) + peerInput := &tg.InputNotifyPeer{Peer: &tg.InputPeerUser{UserID: peerID}} + if ok, err := r.onAccountUpdateNotifySettings(pcCtx, &tg.AccountUpdateNotifySettingsRequest{Peer: peerInput, Settings: in}); err != nil || !ok { + t.Fatalf("PC mute = ok %v err %v", ok, err) + } + + // Phone reconnects / restarts and asks for what changed since its baseline. + diff, err := r.onUpdatesGetDifference(phoneCtx, &tg.UpdatesGetDifferenceRequest{Pts: baselinePts}) + if err != nil { + t.Fatalf("phone catch-up getDifference: %v", err) + } + t.Logf("phone catch-up diff = %#v", diff) + d, ok := diff.(*tg.UpdatesDifference) + if !ok { + t.Fatalf("phone catch-up diff type = %T, want *tg.UpdatesDifference containing the mute", diff) + } + t.Logf("other updates = %#v", d.OtherUpdates) + found := false + for _, u := range d.OtherUpdates { + if ns, ok := u.(*tg.UpdateNotifySettings); ok { + t.Logf("found UpdateNotifySettings: %#v peer=%#v settings=%#v", ns, ns.Peer, ns.NotifySettings) + if np, ok := ns.Peer.(*tg.NotifyPeer); ok { + if pu, ok := np.Peer.(*tg.PeerUser); ok && pu.UserID == peerID { + if mu, hasMu := ns.NotifySettings.GetMuteUntil(); hasMu && mu == 2000000000 { + found = true + } + } + } + } + } + if !found { + t.Fatalf("phone's getDifference does not contain the PC's mute of peer %d -- this is the reported bug", peerID) + } +} diff --git a/internal/rpc/rpc_testkit_updates_test.go b/internal/rpc/rpc_testkit_updates_test.go index d1aff8e4..c71bc033 100644 --- a/internal/rpc/rpc_testkit_updates_test.go +++ b/internal/rpc/rpc_testkit_updates_test.go @@ -259,6 +259,12 @@ func (s *captureUpdates) RecordPeerSettings(_ context.Context, authKeyID [8]byte return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{Type: domain.UpdateEventPeerSettings, Peer: peer, Settings: settings}) } +func (s *captureUpdates) RecordNotifySettings(_ context.Context, authKeyID [8]byte, userID int64, peer domain.Peer, topicID int, settings domain.PeerNotifySettings, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { + s.captureExclude(excludeAuthKeyID, excludeSessionID) + sc := settings.Clone() + return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{Type: domain.UpdateEventNotifySettings, Peer: peer, TopMsgID: topicID, NotifyPeerSettings: &sc}) +} + func (s *captureUpdates) RecordPeerStoryBlocked(_ context.Context, authKeyID [8]byte, userID int64, peer domain.Peer, blocked bool, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, domain.UpdateState, error) { s.captureExclude(excludeAuthKeyID, excludeSessionID) return s.recordCapturedEvent(authKeyID, userID, domain.UpdateEvent{Type: domain.UpdateEventPeerStoryBlocked, Peer: peer, Bool: blocked}) diff --git a/internal/store/postgres/first_contact_getpeerdialogs_repro_test.go b/internal/store/postgres/first_contact_getpeerdialogs_repro_test.go new file mode 100644 index 00000000..a270e53b --- /dev/null +++ b/internal/store/postgres/first_contact_getpeerdialogs_repro_test.go @@ -0,0 +1,96 @@ +package postgres + +import ( + "context" + "testing" + + "telesrv/internal/domain" +) + +// TestFirstContactMessageThenGetPeerDialogsReturnsUsableDialog reproduces the reported +// client symptom (Android + desktop, both stock-derived): a first-ever message from a +// sender with no prior dialog fires a notification, but the new dialog does not appear +// in the recipient's dialog list until app restart. The client's live-update path +// (updateInterfaceWithMessages) builds a dialog in-memory then confirms it via +// messages.getPeerDialogs (DialogStore.ListByPeers server-side). This test drives that +// exact server-side path end to end against real Postgres and dumps every field a stock +// client needs to accept and render the dialog, to catch anything subtly missing/wrong +// that unit tests against fakes wouldn't. +func TestFirstContactMessageThenGetPeerDialogsReturnsUsableDialog(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + suffix := randomSuffix(t) + + users := NewUserStore(pool) + sender, err := users.Create(ctx, domain.User{AccessHash: 31, Phone: "+1667" + suffix + "01", FirstName: "Sender"}) + if err != nil { + t.Fatalf("create sender: %v", err) + } + recipient, err := users.Create(ctx, domain.User{AccessHash: 32, Phone: "+1667" + suffix + "02", FirstName: "Recipient"}) + if err != nil { + t.Fatalf("create recipient: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", []int64{sender.ID, recipient.ID}) + }) + + messages := NewMessageStore(pool) + var originAuthKeyID [8]byte + originAuthKeyID[0] = 9 + sendRes, err := messages.SendPrivateText(ctx, domain.SendPrivateTextRequest{ + SenderUserID: sender.ID, + RecipientUserID: recipient.ID, + RandomID: 555111, + Message: "hey, first message ever", + Date: 1700005000, + OriginAuthKeyID: originAuthKeyID, + OriginSessionID: 41, + }) + if err != nil { + t.Fatalf("SendPrivateText: %v", err) + } + t.Logf("sender box: %+v", sendRes.SenderMessage) + t.Logf("recipient box: %+v", sendRes.RecipientMessage) + + dialogs := NewDialogStore(pool) + peer := domain.Peer{Type: domain.PeerTypeUser, ID: sender.ID} + list, err := dialogs.ListByPeers(ctx, recipient.ID, []domain.Peer{peer}) + if err != nil { + t.Fatalf("ListByPeers (recipient's view of sender): %v", err) + } + t.Logf("dialog list: %+v", list) + if len(list.Dialogs) != 1 { + t.Fatalf("dialogs = %d, want exactly 1 (the freshly-created dialog with sender)", len(list.Dialogs)) + } + d := list.Dialogs[0] + t.Logf("dialog: %+v", d) + if d.Peer != peer { + t.Fatalf("dialog peer = %+v, want %+v", d.Peer, peer) + } + if d.TopMessage == 0 { + t.Fatalf("dialog.TopMessage = 0, want the recipient's box id for the new message") + } + if d.TopMessage != sendRes.RecipientMessage.ID { + t.Fatalf("dialog.TopMessage = %d, want recipient box id %d", d.TopMessage, sendRes.RecipientMessage.ID) + } + if len(list.Messages) != 1 { + t.Fatalf("messages returned = %d, want 1 (top message content, required for stock client to accept the dialog)", len(list.Messages)) + } + msg := list.Messages[0] + t.Logf("message: %+v", msg) + if msg.ID != d.TopMessage { + t.Fatalf("message.ID = %d, does not match dialog.TopMessage = %d -- stock client discards a dialog whose top message it can't resolve", msg.ID, d.TopMessage) + } + if len(list.Users) == 0 { + t.Fatalf("users returned = 0, want at least the sender's User object (client needs it to render the dialog title/avatar)") + } + foundSender := false + for _, u := range list.Users { + if u.ID == sender.ID { + foundSender = true + } + } + if !foundSender { + t.Fatalf("users = %+v, want sender %d present", list.Users, sender.ID) + } +} diff --git a/internal/store/postgres/notify_settings_event_roundtrip_test.go b/internal/store/postgres/notify_settings_event_roundtrip_test.go new file mode 100644 index 00000000..52e383c6 --- /dev/null +++ b/internal/store/postgres/notify_settings_event_roundtrip_test.go @@ -0,0 +1,85 @@ +package postgres + +import ( + "context" + "testing" + + "telesrv/internal/domain" +) + +// TestNotifySettingsEventSurvivesPostgresRoundTrip is a regression test for a real production +// incident: RecordNotifySettings set domain.UpdateEvent.NotifyPeerSettings only in memory, but +// appendUserUpdateEvent/ListAfter never serialized that field to/from Postgres. The write +// itself succeeded, but every later read (outbox dispatch batching, updates.getDifference) +// got NotifyPeerSettings == nil back, so convert_updates.go's UpdateEventNotifySettings case +// silently produced no TL update ("non-noop outbox event produced no update"). Worse: the +// stuck outbox row blocked that account's entire dispatch lane, so unrelated updates (new +// messages) stopped reaching other sessions until a full resync. This test appends a real +// notify_settings event and reads it back through the same store, asserting the settings +// survive the round trip. +func TestNotifySettingsEventSurvivesPostgresRoundTrip(t *testing.T) { + pool := testPool(t) + ctx := context.Background() + suffix := randomSuffix(t) + + users := NewUserStore(pool) + owner, err := users.Create(ctx, domain.User{AccessHash: 51, Phone: "+1668" + suffix + "01", FirstName: "Owner"}) + if err != nil { + t.Fatalf("create owner: %v", err) + } + peerUser, err := users.Create(ctx, domain.User{AccessHash: 52, Phone: "+1668" + suffix + "02", FirstName: "Peer"}) + if err != nil { + t.Fatalf("create peer: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, "DELETE FROM users WHERE id = ANY($1::bigint[])", []int64{owner.ID, peerUser.ID}) + }) + + events := NewUpdateEventStore(pool) + muteUntil := 2000000000 + showPreviews := false + appended, err := events.AppendAllocatedWithDispatch(ctx, owner.ID, domain.UpdateEvent{ + Type: domain.UpdateEventNotifySettings, + Peer: domain.Peer{Type: domain.PeerTypeUser, ID: peerUser.ID}, + NotifyPeerSettings: &domain.PeerNotifySettings{ + MuteUntil: &muteUntil, + ShowPreviews: &showPreviews, + }, + PtsCount: 1, + }, [8]byte{}, 0) + if err != nil { + t.Fatalf("AppendAllocatedWithDispatch: %v", err) + } + t.Logf("appended event = %+v", appended) + if appended.NotifyPeerSettings == nil { + t.Fatalf("appended.NotifyPeerSettings = nil immediately after append, want the settings we sent") + } + + // Read it back exactly the way updates.getDifference does. + replayed, err := events.ListAfter(ctx, owner.ID, appended.Pts-1, 10) + if err != nil { + t.Fatalf("ListAfter: %v", err) + } + t.Logf("replayed events = %+v", replayed) + var found *domain.UpdateEvent + for i := range replayed { + if replayed[i].Type == domain.UpdateEventNotifySettings && replayed[i].Pts == appended.Pts { + found = &replayed[i] + } + } + if found == nil { + t.Fatalf("notify_settings event at pts=%d not found in ListAfter result", appended.Pts) + } + if found.NotifyPeerSettings == nil { + t.Fatalf("replayed event.NotifyPeerSettings = nil -- THE BUG: settings were lost across the Postgres round trip, so getDifference/outbox dispatch cannot build a TL update for this event") + } + if found.NotifyPeerSettings.MuteUntil == nil || *found.NotifyPeerSettings.MuteUntil != muteUntil { + t.Fatalf("replayed MuteUntil = %+v, want %d", found.NotifyPeerSettings.MuteUntil, muteUntil) + } + if found.NotifyPeerSettings.ShowPreviews == nil || *found.NotifyPeerSettings.ShowPreviews != showPreviews { + t.Fatalf("replayed ShowPreviews = %+v, want %v", found.NotifyPeerSettings.ShowPreviews, showPreviews) + } + if found.Peer != (domain.Peer{Type: domain.PeerTypeUser, ID: peerUser.ID}) { + t.Fatalf("replayed Peer = %+v, want peer %d", found.Peer, peerUser.ID) + } +} diff --git a/internal/store/postgres/update_event.go b/internal/store/postgres/update_event.go index b0842054..4a9f537e 100644 --- a/internal/store/postgres/update_event.go +++ b/internal/store/postgres/update_event.go @@ -183,7 +183,12 @@ func appendUserUpdateEvent(ctx context.Context, db sqlcgen.DBTX, q *sqlcgen.Quer if err != nil { return err } - settings, err := encodePeerSettings(event.Settings) + var settings []byte + if event.Type == domain.UpdateEventNotifySettings { + settings, err = encodeNotifyPeerSettings(event.NotifyPeerSettings) + } else { + settings, err = encodePeerSettings(event.Settings) + } if err != nil { return err } @@ -366,6 +371,13 @@ func (s *UpdateEventStore) ListAfter(ctx context.Context, userID int64, pts, lim if err != nil { return nil, fmt.Errorf("decode peer settings: %w", err) } + var notifyPeerSettings *domain.PeerNotifySettings + if domain.UpdateEventType(row.EventType) == domain.UpdateEventNotifySettings { + notifyPeerSettings, err = decodeNotifyPeerSettings(row.PeerSettingsJson) + if err != nil { + return nil, fmt.Errorf("decode notify peer settings: %w", err) + } + } messageIDs, err := decodeEventMessageIDs(row.MessageIdsJson) if err != nil { return nil, fmt.Errorf("decode message ids: %w", err) @@ -407,29 +419,30 @@ func (s *UpdateEventStore) ListAfter(ctx context.Context, userID int64, pts, lim return nil, fmt.Errorf("decode message rich message: %w", err) } event := domain.UpdateEvent{ - UserID: row.UserID, - Type: domain.UpdateEventType(row.EventType), - Pts: int(row.Pts), - PtsCount: int(row.PtsCount), - Date: int(row.Date), - Peer: domain.Peer{Type: domain.PeerType(row.EventPeerType), ID: row.EventPeerID}, - Story: story, - Peers: peers, - Bool: row.EventBool, - Phone: row.EventPhone, - Settings: settings, - MessageIDs: messageIDs, - MaxID: int(row.MaxID), - StillUnreadCount: int(row.StillUnreadCount), - ChannelPts: int(row.ChannelPts), - FilterID: int(row.FilterID), - DialogFilter: dialogFilter, - FilterOrder: filterOrder, - FolderPeers: folderPeers, - TagsEnabled: row.TagsEnabled, - FolderID: int(row.FolderID), - Reaction: reaction, - EmojiStatus: emojiStatus, + UserID: row.UserID, + Type: domain.UpdateEventType(row.EventType), + Pts: int(row.Pts), + PtsCount: int(row.PtsCount), + Date: int(row.Date), + Peer: domain.Peer{Type: domain.PeerType(row.EventPeerType), ID: row.EventPeerID}, + Story: story, + Peers: peers, + Bool: row.EventBool, + Phone: row.EventPhone, + Settings: settings, + NotifyPeerSettings: notifyPeerSettings, + MessageIDs: messageIDs, + MaxID: int(row.MaxID), + StillUnreadCount: int(row.StillUnreadCount), + ChannelPts: int(row.ChannelPts), + FilterID: int(row.FilterID), + DialogFilter: dialogFilter, + FilterOrder: filterOrder, + FolderPeers: folderPeers, + TagsEnabled: row.TagsEnabled, + FolderID: int(row.FolderID), + Reaction: reaction, + EmojiStatus: emojiStatus, Message: domain.Message{ ID: int(row.MessageID), UID: row.PrivateMessageID, @@ -566,6 +579,13 @@ func (s *UpdateEventStore) BatchByCursor(ctx context.Context, cursors []store.Ev if err != nil { return nil, fmt.Errorf("decode peer settings: %w", err) } + var notifyPeerSettings *domain.PeerNotifySettings + if domain.UpdateEventType(row.EventType) == domain.UpdateEventNotifySettings { + notifyPeerSettings, err = decodeNotifyPeerSettings(row.PeerSettingsJson) + if err != nil { + return nil, fmt.Errorf("decode notify peer settings: %w", err) + } + } messageIDs, err := decodeEventMessageIDs(row.MessageIdsJson) if err != nil { return nil, fmt.Errorf("decode message ids: %w", err) @@ -607,29 +627,30 @@ func (s *UpdateEventStore) BatchByCursor(ctx context.Context, cursors []store.Ev return nil, fmt.Errorf("decode message rich message: %w", err) } event := domain.UpdateEvent{ - UserID: row.UserID, - Type: domain.UpdateEventType(row.EventType), - Pts: int(row.Pts), - PtsCount: int(row.PtsCount), - Date: int(row.Date), - Peer: domain.Peer{Type: domain.PeerType(row.EventPeerType), ID: row.EventPeerID}, - Story: story, - Peers: peers, - Bool: row.EventBool, - Phone: row.EventPhone, - Settings: settings, - MessageIDs: messageIDs, - MaxID: int(row.MaxID), - StillUnreadCount: int(row.StillUnreadCount), - ChannelPts: int(row.ChannelPts), - FilterID: int(row.FilterID), - DialogFilter: dialogFilter, - FilterOrder: filterOrder, - FolderPeers: folderPeers, - TagsEnabled: row.TagsEnabled, - FolderID: int(row.FolderID), - Reaction: reaction, - EmojiStatus: emojiStatus, + UserID: row.UserID, + Type: domain.UpdateEventType(row.EventType), + Pts: int(row.Pts), + PtsCount: int(row.PtsCount), + Date: int(row.Date), + Peer: domain.Peer{Type: domain.PeerType(row.EventPeerType), ID: row.EventPeerID}, + Story: story, + Peers: peers, + Bool: row.EventBool, + Phone: row.EventPhone, + Settings: settings, + NotifyPeerSettings: notifyPeerSettings, + MessageIDs: messageIDs, + MaxID: int(row.MaxID), + StillUnreadCount: int(row.StillUnreadCount), + ChannelPts: int(row.ChannelPts), + FilterID: int(row.FilterID), + DialogFilter: dialogFilter, + FilterOrder: filterOrder, + FolderPeers: folderPeers, + TagsEnabled: row.TagsEnabled, + FolderID: int(row.FolderID), + Reaction: reaction, + EmojiStatus: emojiStatus, Message: domain.Message{ ID: int(row.MessageID), UID: row.PrivateMessageID, @@ -1072,6 +1093,53 @@ func decodePeerSettings(raw string) (domain.PeerSettings, error) { }, nil } +type notifyPeerSettingsJSON struct { + ShowPreviews *bool `json:"show_previews,omitempty"` + Silent *bool `json:"silent,omitempty"` + MuteUntil *int `json:"mute_until,omitempty"` + StoriesMuted *bool `json:"stories_muted,omitempty"` + StoriesHideSender *bool `json:"stories_hide_sender,omitempty"` +} + +// encodeNotifyPeerSettings/decodeNotifyPeerSettings reuse the peer_settings_json column: +// UpdateEventNotifySettings rows never populate domain.UpdateEvent.Settings (that field is +// PeerSettings, an unrelated type for a different event type), so there is no collision. +// Adding a dedicated column would need a migration + sqlc regen; this avoids both for a +// small, cleanly-scoped payload. +func encodeNotifyPeerSettings(settings *domain.PeerNotifySettings) ([]byte, error) { + if settings == nil { + return []byte("{}"), nil + } + raw, err := json.Marshal(notifyPeerSettingsJSON{ + ShowPreviews: settings.ShowPreviews, + Silent: settings.Silent, + MuteUntil: settings.MuteUntil, + StoriesMuted: settings.StoriesMuted, + StoriesHideSender: settings.StoriesHideSender, + }) + if err != nil { + return nil, fmt.Errorf("marshal notify peer settings: %w", err) + } + return raw, nil +} + +func decodeNotifyPeerSettings(raw string) (*domain.PeerNotifySettings, error) { + if raw == "" { + return nil, nil + } + var wire notifyPeerSettingsJSON + if err := json.Unmarshal([]byte(raw), &wire); err != nil { + return nil, err + } + return &domain.PeerNotifySettings{ + ShowPreviews: wire.ShowPreviews, + Silent: wire.Silent, + MuteUntil: wire.MuteUntil, + StoriesMuted: wire.StoriesMuted, + StoriesHideSender: wire.StoriesHideSender, + }, nil +} + func maxInt(a, b int) int { if a > b { return a