fix: chat UI

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

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)
+	}
+}

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.