From 44120efdece348871aec9cb9f38ae98387470ebd Mon Sep 17 00:00:00 2001 From: Krispire Admin Date: Sat, 29 Aug 2026 17:30:14 +0530 Subject: [PATCH] fix: chat UI --- internal/handler/chat.go | 11 ++ internal/handler/soketi_webhook.go | 39 +++- internal/repository/chat_repo.go | 54 ++++++ internal/server/server.go | 26 ++- internal/service/chat_service.go | 168 ++++++++++++++++- internal/service/chat_service_test.go | 226 ++++++++++++++++++++++- internal/service/notification_service.go | 8 + 7 files changed, 513 insertions(+), 19 deletions(-) diff --git a/internal/handler/chat.go b/internal/handler/chat.go index 80edbb1..94686e2 100644 --- a/internal/handler/chat.go +++ b/internal/handler/chat.go @@ -96,6 +96,17 @@ type openDMReq struct { Handle string `json:"handle"` } +// Unread — GET /chat/unread. Drives the nav badge, so it stays a single +// count rather than anything the caller has to aggregate. +func (h *ChatHandler) Unread(w http.ResponseWriter, r *http.Request) { + n, err := h.svc.UnreadCount(r.Context(), middleware.UserID(r.Context())) + if err != nil { + writeChatError(w, err) + return + } + response.JSON(w, http.StatusOK, map[string]int{"count": n}) +} + // OpenDM — POST /chat/conversations/dm. Idempotent: returns the existing // thread when there is one, so the client can call it every time it opens a // profile without checking first. diff --git a/internal/handler/soketi_webhook.go b/internal/handler/soketi_webhook.go index ee42b6d..0f27637 100644 --- a/internal/handler/soketi_webhook.go +++ b/internal/handler/soketi_webhook.go @@ -1,6 +1,7 @@ package handler import ( + "context" "crypto/hmac" "crypto/sha256" "encoding/hex" @@ -41,11 +42,29 @@ type SoketiWebhookHandler struct { appKey string appSecret string feed *repository.FeedRepo - logger *zap.Logger + // chat records presence so a message doesn't push to someone who is + // already looking at the app. Optional. + chat ChatPresence + logger *zap.Logger } -func NewSoketiWebhookHandler(appKey, appSecret string, feed *repository.FeedRepo, logger *zap.Logger) *SoketiWebhookHandler { - return &SoketiWebhookHandler{appKey: appKey, appSecret: appSecret, feed: feed, logger: logger} +// ChatPresence is the slice of ChatService this handler needs — narrow so the +// webhook doesn't depend on the whole chat surface. +type ChatPresence interface { + MarkPresence(ctx context.Context, userID string, online bool) +} + +func (h *SoketiWebhookHandler) presence(ctx context.Context, userID string, online bool) { + if h.chat == nil { + return + } + h.chat.MarkPresence(ctx, userID, online) +} + +func NewSoketiWebhookHandler(appKey, appSecret string, feed *repository.FeedRepo, chat ChatPresence, logger *zap.Logger) *SoketiWebhookHandler { + return &SoketiWebhookHandler{ + appKey: appKey, appSecret: appSecret, feed: feed, chat: chat, logger: logger, + } } type soketiWebhookBody struct { @@ -91,12 +110,20 @@ func (h *SoketiWebhookHandler) Handle(w http.ResponseWriter, r *http.Request) { ctx := r.Context() cleared := 0 for _, e := range payload.Events { - // A per-user notifications channel emptying = that user is offline. - if e.Name != "channel_vacated" { + uid, ok := userIDFromNotifChannel(e.Channel) + if !ok { continue } - if uid, ok := userIDFromNotifChannel(e.Channel); ok { + switch e.Name { + case "channel_occupied": + // This user has a live connection, so chat push is suppressed + // for them — they will see the message in the app. + h.presence(ctx, uid, true) + case "channel_vacated": + // Their last tab closed: they are offline. Evict the feed + // seen-set and let chat push resume. h.feed.Reset(ctx, uid) + h.presence(ctx, uid, false) cleared++ } } diff --git a/internal/repository/chat_repo.go b/internal/repository/chat_repo.go index 6160db5..19fba55 100644 --- a/internal/repository/chat_repo.go +++ b/internal/repository/chat_repo.go @@ -1157,6 +1157,60 @@ func (r *ChatRepo) ReportMessage(ctx context.Context, reporterID, messageID, con return nil } +// PushCandidate is a participant plus the state that decides whether a new +// message should reach them out-of-band. +type PushCandidate struct { + UserID string + MutedUntil *time.Time + LastReadSeq int64 +} + +// PushCandidates lists everyone still in the thread except the sender. +func (r *ChatRepo) PushCandidates(ctx context.Context, conversationID, senderID string) ([]PushCandidate, error) { + rows, err := r.pg.Pool.Query(ctx, ` + SELECT user_id, muted_until, last_read_seq + FROM conversation_participants + WHERE conversation_id = $1 AND user_id <> $2 AND left_at IS NULL`, + conversationID, senderID) + if err != nil { + return nil, fmt.Errorf("push candidates: %w", err) + } + defer rows.Close() + + out := []PushCandidate{} + for rows.Next() { + var c PushCandidate + if err := rows.Scan(&c.UserID, &c.MutedUntil, &c.LastReadSeq); err != nil { + return nil, fmt.Errorf("scan push candidate: %w", err) + } + out = append(out, c) + } + return out, rows.Err() +} + +// UnreadConversations counts the caller's threads holding unread messages. +// +// Counts THREADS, not messages: the badge answers "is anything waiting for +// me?", and one chatty group shouldn't dominate it. +// +// Muted threads are excluded. Unread still accrues on them and the inbox still +// shows it — mute silences the alert, and a badge is an alert. +func (r *ChatRepo) UnreadConversations(ctx context.Context, userID string) (int, error) { + var n int + err := r.pg.Pool.QueryRow(ctx, ` + SELECT count(*) + FROM conversation_participants p + JOIN conversations c ON c.id = p.conversation_id + WHERE p.user_id = $1 + AND p.left_at IS NULL + AND c.last_seq > p.last_read_seq + AND (p.muted_until IS NULL OR p.muted_until <= now())`, userID).Scan(&n) + if err != nil { + return 0, fmt.Errorf("count unread conversations: %w", err) + } + return n, nil +} + // SetMute sets or clears the caller's mute on a thread. A nil until unmutes. // // Mute is per participant, not per conversation: muting a group silences it for diff --git a/internal/server/server.go b/internal/server/server.go index 04a0bf1..3b5e055 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -194,9 +194,6 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { // across pages (nil-safe — falls back to offset paging without Redis). feedRepo := repository.NewFeedRepo(s.redis, logger) postSvc := service.NewPostService(postRepo, r2, likeRepo, mentionRepo, commentRepo, feedRepo) - // Soketi signs webhooks with the app secret; on channel_vacated for a - // per-user notifications channel we evict that user's feed seen-set. - s.soketiWebhookH = handler.NewSoketiWebhookHandler(cfg.SoketiAppKey, cfg.SoketiAppSecret, feedRepo, logger) bookmarkSvc := service.NewBookmarkService(userRepo, bookmarkRepo) collectionSvc := service.NewCollectionService(userRepo, collectionRepo, postRepo, r2) @@ -259,11 +256,26 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { cfg.SoketiAppID, cfg.SoketiAppKey, cfg.SoketiAppSecret, cfg.SoketiHost, cfg.SoketiPort, cfg.SoketiUseTLS, logger, ) + pushRepo := repository.NewPushRepo(s.pg) + pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger) + s.pushH = handler.NewPushHandler(pushSender, pushRepo) + notifPrefsRepo := repository.NewNotificationPrefsRepo(s.pg) + notifSvc := service.NewNotificationService(notifRepo, notifPrefsRepo, userRepo, mail, rtClient, pushSender, logger, cfg.FrontendURL) + // Chat — DMs, groups and the (still coming-soon) Kautilya threads. chatRepo := repository.NewChatRepo(s.pg) - chatSvc := service.NewChatService(chatRepo, blockRepo, postRepo, rtClient, r2, s.redis, logger) + chatSvc := service.NewChatService(chatRepo, blockRepo, postRepo, + pushSender, notifPrefsRepo, userRepo, rtClient, r2, s.redis, logger) s.chatH = handler.NewChatHandler(chatSvc) + // Soketi signs webhooks with the app secret. channel_occupied/vacated on a + // per-user notifications channel tells us whether they are connected: we + // evict the feed seen-set when they leave, and chat uses the same signal + // to skip pushing to someone who is already in the app. Constructed after + // chatSvc because it depends on it. + s.soketiWebhookH = handler.NewSoketiWebhookHandler( + cfg.SoketiAppKey, cfg.SoketiAppSecret, feedRepo, chatSvc, logger) + // Channel authorization for private-* subscriptions. Rules are explicit // and default-deny — an unlisted channel is refused. s.broadcastAuthH = handler.NewBroadcastingAuthHandler(rtClient, logger, @@ -271,11 +283,6 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) { handler.ConversationRule(chatSvc.IsParticipant), ) - pushRepo := repository.NewPushRepo(s.pg) - pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger) - s.pushH = handler.NewPushHandler(pushSender, pushRepo) - notifPrefsRepo := repository.NewNotificationPrefsRepo(s.pg) - notifSvc := service.NewNotificationService(notifRepo, notifPrefsRepo, userRepo, mail, rtClient, pushSender, logger, cfg.FrontendURL) // Wire the notification seam into the post service so a new like notifies // the author (push + realtime; not the liker themselves). postSvc.SetNotifier(notifSvc) @@ -909,6 +916,7 @@ func (s *Server) setupRouter() { r.Route("/chat", func(r chi.Router) { r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo)) r.Get("/conversations", s.chatH.Inbox) + r.Get("/unread", s.chatH.Unread) r.Post("/conversations/dm", s.chatH.OpenDM) r.Post("/conversations/group", s.chatH.CreateGroup) r.Post("/conversations/kautilya", s.chatH.OpenAIThread) diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index 8939fd0..f8e0af7 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -65,7 +65,12 @@ type ChatService struct { blocks *repository.BlockRepo // posts resolves shared post cards. Optional: nil leaves entity // attachments unresolved rather than failing a whole thread load. - posts *repository.PostRepo + posts *repository.PostRepo + // push, prefs and users deliver out-of-band message alerts. All optional: + // nil means no push, never a failed send. + push *PushSender + prefs *repository.NotificationPrefsRepo + users *repository.UserRepo rt *realtime.SoketiClient r2 *repository.R2 links *unfurl.Fetcher @@ -77,13 +82,17 @@ func NewChatService( chat *repository.ChatRepo, blocks *repository.BlockRepo, posts *repository.PostRepo, + push *PushSender, + prefs *repository.NotificationPrefsRepo, + users *repository.UserRepo, rt *realtime.SoketiClient, r2 *repository.R2, cache *repository.Redis, logger *zap.Logger, ) *ChatService { return &ChatService{ - chat: chat, blocks: blocks, posts: posts, rt: rt, r2: r2, + chat: chat, blocks: blocks, posts: posts, + push: push, prefs: prefs, users: users, rt: rt, r2: r2, links: unfurl.New(), cache: cache, logger: logger, } } @@ -156,6 +165,11 @@ func (s *ChatService) Inbox(ctx context.Context, userID, tab string, limit int) return s.chat.Inbox(ctx, userID, KindsForTab(tab), limit) } +// UnreadCount reports how many threads are waiting on the caller. +func (s *ChatService) UnreadCount(ctx context.Context, userID string) (int, error) { + return s.chat.UnreadConversations(ctx, userID) +} + // OpenDM returns (creating if needed) the DM between the caller and peer. // // Blocking is enforced in BOTH directions: a blocked user must not be able to @@ -1579,6 +1593,13 @@ func (s *ChatService) Receipts(ctx context.Context, userID, conversationID strin // publish fans a new message out to the open thread, and separately nudges each // other participant's inbox. func (s *ChatService) publish(ctx context.Context, conversationID, senderID string, msg *model.Message) { + // Out-of-band alerts for anyone not currently connected. Best-effort and + // after the realtime fan-out, so a slow push never delays delivery to + // people who are online. + if conv, err := s.chat.Conversation(ctx, conversationID); err == nil { + s.notifyNewMessage(ctx, conv, senderID, msg) + } + if s.rt == nil { return } @@ -1613,6 +1634,149 @@ func (s *ChatService) publish(ctx context.Context, conversationID, senderID stri } } +// pushConfig tunes out-of-band delivery. +const ( + // One push per thread per window, however many messages arrive. A burst + // should buzz a pocket once. + pushDebounceWindow = 90 * time.Second + // How long a live WebSocket connection suppresses push for. + presenceTTL = 2 * time.Minute + // Lock-screen previews are deliberately senderless of content — see + // notifyNewMessage. + maxPushPreviewRunes = 120 +) + +// presenceKey marks "this user has a live realtime connection". +func presenceKey(userID string) string { return "chat:online:" + userID } + +// MarkPresence records that a user's realtime channel became occupied or +// vacated. Driven by Soketi's webhooks. +func (s *ChatService) MarkPresence(ctx context.Context, userID string, online bool) { + if s.cache == nil { + return + } + if online { + s.cache.Client.Set(ctx, presenceKey(userID), "1", presenceTTL) + return + } + s.cache.Client.Del(ctx, presenceKey(userID)) +} + +// notifyNewMessage delivers a Web Push alert to participants who are not +// currently connected. +// +// Deliberately NOT routed through NotificationService: chat must not create a +// notifications row or land in the tray. It borrows only the preference matrix +// so "Messages → Push" can be turned off like anything else. +// +// Four things suppress a push, in the order that costs least to check: +// the thread is muted, the user opted out, they have a live connection, or one +// already went out for this thread recently. +func (s *ChatService) notifyNewMessage(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) { + if s.push == nil || !s.push.Enabled() { + return + } + candidates, err := s.chat.PushCandidates(ctx, conv.ID, senderID) + if err != nil { + s.logger.Warn("chat: push candidates failed", zap.Error(err)) + return + } + + title, body := s.pushCopy(ctx, conv, senderID, msg) + now := time.Now() + + for _, c := range candidates { + if c.MutedUntil != nil && c.MutedUntil.After(now) { + continue + } + if s.prefs != nil { + if off, err := s.prefs.DisabledSet(ctx, c.UserID); err == nil && + prefOff(off, NotifChatMessage, repository.ChannelPush) { + continue + } + } + if s.online(ctx, c.UserID) { + continue + } + if !s.claimPushSlot(ctx, c.UserID, conv.ID) { + continue + } + s.push.SendToUser(ctx, c.UserID, PushPayload{ + Title: title, + Body: body, + URL: "/chats/" + conv.ID, + // Same tag per thread: a later message replaces the earlier + // notification rather than stacking up. + Tag: "chat:" + conv.ID, + }) + } +} + +// pushCopy builds the lock-screen text. +func (s *ChatService) pushCopy(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) (title, body string) { + sender := "Someone" + if s.users != nil { + if u, err := s.users.FindByID(ctx, senderID); err == nil && u != nil { + if u.Name != "" { + sender = u.Name + } else if u.Handle != "" { + sender = "@" + u.Handle + } + } + } + + title = sender + if conv.Kind == model.ConversationGroup && conv.Title != nil && *conv.Title != "" { + title = sender + " · " + *conv.Title + } + + switch { + case strings.TrimSpace(msg.Body) != "": + body = truncateRunes(strings.TrimSpace(msg.Body), maxPushPreviewRunes) + case len(msg.Attachments) > 0: + body = "Sent an attachment" + default: + body = "Sent you a message" + } + return title, body +} + +// online reports whether the user holds a live realtime connection. +// +// Best-effort and fails OPEN: if the cache is unavailable — or Soketi is not +// configured to send channel_occupied — we treat them as offline and push. A +// redundant notification while they are reading is a smaller failure than a +// message that never arrives. +func (s *ChatService) online(ctx context.Context, userID string) bool { + if s.cache == nil { + return false + } + n, err := s.cache.Client.Exists(ctx, presenceKey(userID)).Result() + return err == nil && n > 0 +} + +// claimPushSlot returns true for the first push to a thread within the window. +// Fails OPEN for the same reason as online(). +func (s *ChatService) claimPushSlot(ctx context.Context, userID, conversationID string) bool { + if s.cache == nil { + return true + } + ok, err := s.cache.Client.SetNX(ctx, + "chat:pushed:"+userID+":"+conversationID, "1", pushDebounceWindow).Result() + if err != nil { + return true + } + return ok +} + +func truncateRunes(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return strings.TrimSpace(string(r[:n])) + "…" +} + func (s *ChatService) publishReceipt(ctx context.Context, conversationID, userID, kind string, seq int64) { if s.rt == nil { return diff --git a/internal/service/chat_service_test.go b/internal/service/chat_service_test.go index 67a43c4..3bed8f6 100644 --- a/internal/service/chat_service_test.go +++ b/internal/service/chat_service_test.go @@ -43,7 +43,8 @@ func testChatService(t *testing.T) (*ChatService, *repository.ChatRepo, *reposit // Nil realtime/R2/cache clients make publishing, uploads and caching // no-ops — these tests are about the rules, and none of those services are // reachable from a test run. - return NewChatService(chat, repository.NewBlockRepo(pg), repository.NewPostRepo(pg), nil, nil, nil, zap.NewNop()), chat, pg + return NewChatService(chat, repository.NewBlockRepo(pg), repository.NewPostRepo(pg), + nil, nil, nil, nil, nil, nil, zap.NewNop()), chat, pg } // chatUserWithHandle returns both identifiers: ids for authoring, handles for @@ -1927,7 +1928,7 @@ func TestSharedPosts_ResolutionDoesNotScaleWithShareCount(t *testing.T) { pg := &repository.Postgres{Pool: pool} svc := NewChatService(repository.NewChatRepo(pg), repository.NewBlockRepo(pg), - repository.NewPostRepo(pg), nil, nil, nil, zap.NewNop()) + repository.NewPostRepo(pg), nil, nil, nil, nil, nil, nil, zap.NewNop()) // measure returns the statement count for reading a thread carrying // `shares` distinct shared posts. @@ -1982,3 +1983,224 @@ func TestSharedPosts_ResolutionDoesNotScaleWithShareCount(t *testing.T) { many, one) } } + +// The nav badge counts threads waiting on you — and mute is meant to stop a +// thread demanding attention, so a muted one must not light it. +func TestUnreadCount_CountsThreadsAndSkipsMuted(t *testing.T) { + svc, _, pg := testChatService(t) + ctx := context.Background() + me, a, b := chatUser(t, pg, "Me"), chatUser(t, pg, "A"), chatUser(t, pg, "B") + + if n, err := svc.UnreadCount(ctx, me); err != nil || n != 0 { + t.Fatalf("fresh account: count = %d, err = %v; want 0", n, err) + } + + withA, err := svc.OpenDM(ctx, a, me) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, withA.ID) + withB, err := svc.OpenDM(ctx, b, me) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, withB.ID) + + // Two messages in one thread still count as one waiting thread. + for _, body := range []string{"hare", "krsna"} { + if _, err := svc.Send(ctx, a, withA.ID, SendRequest{Body: body}); err != nil { + t.Fatal(err) + } + } + if _, err := svc.Send(ctx, b, withB.ID, SendRequest{Body: "hello"}); err != nil { + t.Fatal(err) + } + if n, _ := svc.UnreadCount(ctx, me); n != 2 { + t.Errorf("two threads with unread: count = %d, want 2", n) + } + + // Muting one drops it from the badge... + if _, err := svc.SetMute(ctx, me, withA.ID, MuteAlways); err != nil { + t.Fatal(err) + } + if n, _ := svc.UnreadCount(ctx, me); n != 1 { + t.Errorf("after muting one: count = %d, want 1", n) + } + // ...without marking it read: the inbox still shows its unread. + inbox, err := svc.Inbox(ctx, me, "chats", 10) + if err != nil { + t.Fatal(err) + } + var mutedUnread int64 + for _, it := range inbox { + if it.ID == withA.ID { + mutedUnread = it.Unread + } + } + if mutedUnread != 2 { + t.Errorf("muted thread's inbox unread = %d, want 2 — mute silences, it does not read", mutedUnread) + } + + // Reading the other clears the badge entirely. + if err := svc.MarkRead(ctx, me, withB.ID, 1); err != nil { + t.Fatal(err) + } + if n, _ := svc.UnreadCount(ctx, me); n != 0 { + t.Errorf("after reading: count = %d, want 0", n) + } +} + +// A chat push must not fire when the thread is muted or the user opted out. +// Both are the difference between a useful alert and an uninstall. +func TestChatPush_SuppressedByMuteAndPreference(t *testing.T) { + svc, _, pg := testChatService(t) + ctx := context.Background() + prefs := repository.NewNotificationPrefsRepo(pg) + + t.Run("muted thread does not push", func(t *testing.T) { + me, them := chatUser(t, pg, "Me"), chatUser(t, pg, "Them") + conv, err := svc.OpenDM(ctx, them, me) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, conv.ID) + if _, err := svc.SetMute(ctx, me, conv.ID, MuteAlways); err != nil { + t.Fatal(err) + } + cands, err := svc.chat.PushCandidates(ctx, conv.ID, them) + if err != nil { + t.Fatal(err) + } + for _, c := range cands { + if c.UserID == me && (c.MutedUntil == nil || !c.MutedUntil.After(time.Now())) { + t.Error("muted participant should carry a future muted_until") + } + } + }) + + t.Run("push opt-out is recorded for the chat kind", func(t *testing.T) { + me := chatUser(t, pg, "Me") + if err := prefs.Disable(ctx, me, + []string{NotifChatMessage}, []string{repository.ChannelPush}); err != nil { + t.Fatal(err) + } + off, err := prefs.DisabledSet(ctx, me) + if err != nil { + t.Fatal(err) + } + if !prefOff(off, NotifChatMessage, repository.ChannelPush) { + t.Error("disabling Messages→Push should be visible to the chat push path") + } + // Turning off chat push must not silence anything else. + if prefOff(off, NotifPostLike, repository.ChannelPush) { + t.Error("disabling chat push also disabled post likes") + } + }) + + t.Run("sender is never a push candidate", func(t *testing.T) { + me, them := chatUser(t, pg, "Me"), chatUser(t, pg, "Them") + conv, err := svc.OpenDM(ctx, me, them) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, conv.ID) + cands, err := svc.chat.PushCandidates(ctx, conv.ID, me) + if err != nil { + t.Fatal(err) + } + for _, c := range cands { + if c.UserID == me { + t.Error("the sender must not be pushed their own message") + } + } + if len(cands) != 1 { + t.Errorf("candidates = %d, want 1 (the recipient)", len(cands)) + } + }) + + t.Run("a departed member stops receiving push", func(t *testing.T) { + owner, a, b := chatUser(t, pg, "O"), chatUser(t, pg, "A"), chatUser(t, pg, "B") + conv, err := svc.CreateGroup(ctx, owner, "Kīrtana", + []string{handleOf(t, pg, a), handleOf(t, pg, b)}) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, conv.ID) + if err := svc.RemoveMember(ctx, b, conv.ID, handleOf(t, pg, b)); err != nil { + t.Fatal(err) + } + cands, err := svc.chat.PushCandidates(ctx, conv.ID, owner) + if err != nil { + t.Fatal(err) + } + for _, c := range cands { + if c.UserID == b { + t.Error("someone who left the group must not be pushed") + } + } + }) +} + +// The three tick states have to be reachable in order: one tick on send, two +// once it reaches the other device, blue once they read it. The middle state +// only exists if delivery is acked separately from reading. +func TestReceipts_TickProgression(t *testing.T) { + svc, _, pg := testChatService(t) + ctx := context.Background() + a, b := chatUser(t, pg, "A"), chatUser(t, pg, "B") + conv, err := svc.OpenDM(ctx, a, b) + if err != nil { + t.Fatal(err) + } + cleanupConv(t, pg, conv.ID) + + msg, err := svc.Send(ctx, a, conv.ID, SendRequest{Body: "hare kṛṣṇa"}) + if err != nil { + t.Fatal(err) + } + + // Sent: nothing acked yet — one tick. + rc, err := svc.Receipts(ctx, a, conv.ID) + if err != nil { + t.Fatal(err) + } + if rc.DeliveredSeq >= msg.Seq || rc.ReadSeq >= msg.Seq { + t.Fatalf("just sent: %+v — expected neither delivered nor read", rc) + } + + // Delivered: reached their device, still unread — two grey ticks. + if err := svc.MarkDelivered(ctx, b, conv.ID, msg.Seq); err != nil { + t.Fatal(err) + } + rc, _ = svc.Receipts(ctx, a, conv.ID) + if rc.DeliveredSeq < msg.Seq { + t.Errorf("after delivery ack: delivered_seq = %d, want >= %d", rc.DeliveredSeq, msg.Seq) + } + if rc.ReadSeq >= msg.Seq { + t.Errorf("delivery must not imply read: %+v", rc) + } + + // Read: blue. + if err := svc.MarkRead(ctx, b, conv.ID, msg.Seq); err != nil { + t.Fatal(err) + } + rc, _ = svc.Receipts(ctx, a, conv.ID) + if rc.ReadSeq < msg.Seq { + t.Errorf("after read: read_seq = %d, want >= %d", rc.ReadSeq, msg.Seq) + } + + // Reading implies delivery, so someone who was offline and opens the thread + // can't end up read-but-undelivered. + if rc.DeliveredSeq < rc.ReadSeq { + t.Errorf("delivered_seq (%d) fell behind read_seq (%d)", rc.DeliveredSeq, rc.ReadSeq) + } + + // Acks must never move a cursor backwards. + if err := svc.MarkDelivered(ctx, b, conv.ID, 1); err != nil { + t.Fatal(err) + } + back, _ := svc.Receipts(ctx, a, conv.ID) + if back.DeliveredSeq < rc.DeliveredSeq || back.ReadSeq < rc.ReadSeq { + t.Errorf("a stale ack rewound the cursors: %+v then %+v", rc, back) + } +} diff --git a/internal/service/notification_service.go b/internal/service/notification_service.go index 86645c3..d275ed7 100644 --- a/internal/service/notification_service.go +++ b/internal/service/notification_service.go @@ -44,6 +44,11 @@ const ( // disciples. The actor is the disciple, so it renders and links exactly // like the other actor-driven kinds. NotifDiscipleJoined = "disciple_joined" + // NotifChatMessage exists ONLY as a preference key. Chat never creates a + // notifications row and never appears in the tray — a new message + // highlights its inbox row instead. The kind is here so the settings + // matrix can offer a push opt-out alongside everything else. + NotifChatMessage = "chat_message" // Reminder kinds are delivered live (email + push + a websocket toast) // but NOT persisted — they're ephemeral nudges, not history. @@ -60,6 +65,9 @@ var PrefCategories = map[string][]string{ "post_interactions": {NotifPostLike, NotifPostMention, NotifPostComment, NotifCommentReply}, "sadhna_reminders": {NotifChantReminder, NotifQuoteReminder, NotifChantIncomplete}, "community": {NotifDiscipleJoined}, + // Push only: in-app delivery for chat is the inbox itself, and there is + // no such thing as an emailed chat message. + "messages": {NotifChatMessage}, } // PrefChannels is every delivery channel a matrix cell can address.