feat: always notify on DM messages like Slack/Discord (#405)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

crates/sprout-db/src/feed.rs

@@ -28,6 +28,7 @@
 pub const FEED_MAX_LIMIT: i64 = 100;
 
 use chrono::{DateTime, Utc};
+use sqlx::postgres::PgRow;
 use sqlx::{PgPool, QueryBuilder};
 use uuid::Uuid;
 
@@ -41,6 +42,39 @@ use sprout_core::StoredEvent;
 use crate::error::Result;
 use crate::event::row_to_stored_event;
 
+/// Column list shared by every feed subquery that aliases the `events` table as `e`.
+const EVENT_COLS: &str =
+    "e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, e.received_at, e.channel_id";
+
+/// Column list for queries that select directly from `events` (no table alias).
+const EVENT_COLS_UNALIASED: &str =
+    "id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id";
+
+/// Append `AND <col> IN ($1, $2, …)` for the given channel IDs.
+///
+/// No-ops when the slice is empty so callers don't need a guard.
+fn push_channel_id_filter(qb: &mut QueryBuilder<sqlx::Postgres>, col: &str, ids: &[Uuid]) {
+    if !ids.is_empty() {
+        qb.push(format!(" AND {col} IN ("));
+        let mut sep = qb.separated(", ");
+        for id in ids {
+            sep.push_bind(*id);
+        }
+        qb.push(")");
+    }
+}
+
+/// Convert fetched rows into `Vec<StoredEvent>`, skipping any that fail conversion.
+fn collect_stored_events(rows: Vec<PgRow>) -> Result<Vec<StoredEvent>> {
+    let mut out = Vec::with_capacity(rows.len());
+    for row in rows {
+        if let Some(ev) = row_to_stored_event(row)? {
+            out.push(ev);
+        }
+    }
+    Ok(out)
+}
+
 /// Find events that @mention the given pubkey (have `["p", pubkey_hex]` in tags).
 ///
 /// Joins against the `event_mentions` table -- Phase 2 implementation.
@@ -58,44 +92,26 @@ pub async fn query_mentions(
     let limit = limit.min(FEED_MAX_LIMIT);
     let pubkey_hex = hex::encode(pubkey_bytes);
 
-    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
-        "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, \
-         e.received_at, e.channel_id \
-         FROM events e \
+    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(format!(
+        "SELECT {EVENT_COLS} FROM events e \
          INNER JOIN event_mentions m ON e.id = m.event_id \
-         WHERE m.pubkey_hex = ",
-    );
+         WHERE m.pubkey_hex = "
+    ));
     qb.push_bind(&pubkey_hex);
     qb.push(" AND e.deleted_at IS NULL");
-
     qb.push(format!(
-        " AND e.kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, {KIND_FORUM_POST}, {KIND_FORUM_COMMENT})"
+        " AND e.kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, \
+         {KIND_FORUM_POST}, {KIND_FORUM_COMMENT})"
     ));
-
-    if !accessible_channel_ids.is_empty() {
-        qb.push(" AND e.channel_id IN (");
-        let mut sep = qb.separated(", ");
-        for id in accessible_channel_ids {
-            sep.push_bind(*id);
-        }
-        qb.push(")");
-    }
-
+    push_channel_id_filter(&mut qb, "e.channel_id", accessible_channel_ids);
     if let Some(s) = since {
         qb.push(" AND m.event_created_at >= ").push_bind(s);
     }
-
     qb.push(" ORDER BY m.event_created_at DESC LIMIT ")
         .push_bind(limit);
 
     let rows = qb.build().fetch_all(pool).await?;
-    let mut out = Vec::with_capacity(rows.len());
-    for row in rows {
-        if let Some(ev) = row_to_stored_event(row)? {
-            out.push(ev);
-        }
-    }
-    Ok(out)
+    collect_stored_events(rows)
 }
 
 /// Find events that require action from the given pubkey:
@@ -117,44 +133,25 @@ pub async fn query_needs_action(
     let limit = limit.min(FEED_MAX_LIMIT);
     let pubkey_hex = hex::encode(pubkey_bytes);
 
-    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
-        "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, \
-         e.received_at, e.channel_id \
-         FROM events e \
+    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(format!(
+        "SELECT {EVENT_COLS} FROM events e \
          INNER JOIN event_mentions m ON e.id = m.event_id \
-         WHERE m.pubkey_hex = ",
-    );
+         WHERE m.pubkey_hex = "
+    ));
     qb.push_bind(&pubkey_hex);
     qb.push(" AND e.deleted_at IS NULL");
-
     qb.push(format!(
         " AND e.kind IN ({KIND_WORKFLOW_APPROVAL_REQUESTED}, {KIND_STREAM_REMINDER})"
     ));
-
-    if !accessible_channel_ids.is_empty() {
-        qb.push(" AND e.channel_id IN (");
-        let mut sep = qb.separated(", ");
-        for id in accessible_channel_ids {
-            sep.push_bind(*id);
-        }
-        qb.push(")");
-    }
-
+    push_channel_id_filter(&mut qb, "e.channel_id", accessible_channel_ids);
     if let Some(s) = since {
         qb.push(" AND m.event_created_at >= ").push_bind(s);
     }
-
     qb.push(" ORDER BY m.event_created_at DESC LIMIT ")
         .push_bind(limit);
 
     let rows = qb.build().fetch_all(pool).await?;
-    let mut out = Vec::with_capacity(rows.len());
-    for row in rows {
-        if let Some(ev) = row_to_stored_event(row)? {
-            out.push(ev);
-        }
-    }
-    Ok(out)
+    collect_stored_events(rows)
 }
 
 /// Find recent activity across accessible channels (for watched topics / agent activity).
@@ -170,40 +167,21 @@ pub async fn query_activity(
     limit: i64,
 ) -> Result<Vec<StoredEvent>> {
     let limit = limit.min(FEED_MAX_LIMIT);
-    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
-        "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
-         FROM events WHERE 1=1",
-    );
-
-    qb.push(" AND deleted_at IS NULL");
-
-    qb.push(format!(
-        " AND kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, {KIND_FORUM_POST}, {KIND_JOB_REQUEST}, {KIND_JOB_PROGRESS}, {KIND_JOB_RESULT})"
+    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(format!(
+        "SELECT {EVENT_COLS_UNALIASED} FROM events WHERE deleted_at IS NULL"
     ));
-
-    if !accessible_channel_ids.is_empty() {
-        qb.push(" AND channel_id IN (");
-        let mut sep = qb.separated(", ");
-        for id in accessible_channel_ids {
-            sep.push_bind(*id);
-        }
-        qb.push(")");
-    }
-
+    qb.push(format!(
+        " AND kind IN ({KIND_STREAM_MESSAGE}, {KIND_STREAM_MESSAGE_V2}, {KIND_FORUM_POST}, \
+         {KIND_JOB_REQUEST}, {KIND_JOB_PROGRESS}, {KIND_JOB_RESULT})"
+    ));
+    push_channel_id_filter(&mut qb, "channel_id", accessible_channel_ids);
     if let Some(s) = since {
         qb.push(" AND created_at >= ").push_bind(s);
     }
-
     qb.push(" ORDER BY created_at DESC LIMIT ").push_bind(limit);
 
     let rows = qb.build().fetch_all(pool).await?;
-    let mut out = Vec::with_capacity(rows.len());
-    for row in rows {
-        if let Some(ev) = row_to_stored_event(row)? {
-            out.push(ev);
-        }
-    }
-    Ok(out)
+    collect_stored_events(rows)
 }
 
 // -- Tests --------------------------------------------------------------------

crates/sprout-relay/src/api/feed.rs

@@ -121,8 +121,14 @@ pub async fn feed_handler(
         tracing::warn!("feed: failed to load channel names for enrichment: {e}");
         vec![]
     });
-    let channel_name_map: HashMap<uuid::Uuid, String> =
-        all_channels.into_iter().map(|c| (c.id, c.name)).collect();
+    let channel_name_map: HashMap<uuid::Uuid, String> = all_channels
+        .iter()
+        .map(|c| (c.id, c.name.clone()))
+        .collect();
+    let channel_type_map: HashMap<uuid::Uuid, String> = all_channels
+        .into_iter()
+        .map(|c| (c.id, c.channel_type))
+        .collect();
 
     let to_feed_item = |event: &sprout_core::StoredEvent, category: &str| -> serde_json::Value {
         let channel_name = event
@@ -131,6 +137,12 @@ pub async fn feed_handler(
             .cloned()
             .unwrap_or_default();
 
+        let channel_type = event
+            .channel_id
+            .and_then(|id| channel_type_map.get(&id))
+            .cloned()
+            .unwrap_or_default();
+
         let tags: Vec<serde_json::Value> = event
             .event
             .tags
@@ -149,6 +161,7 @@ pub async fn feed_handler(
             "created_at": event.event.created_at.as_u64(),
             "channel_id": event.channel_id.map(|id| id.to_string()),
             "channel_name": channel_name,
+            "channel_type": channel_type,
             "tags": tags,
             "category": category,
         })

desktop/src-tauri/src/models.rs

@@ -224,6 +224,8 @@ pub struct FeedItemInfo {
     pub created_at: u64,
     pub channel_id: Option<String>,
     pub channel_name: String,
+    #[serde(default)]
+    pub channel_type: Option<String>,
     pub tags: Vec<Vec<String>>,
     pub category: String,
 }

desktop/src/app/AppShell.tsx

@@ -23,6 +23,7 @@ import { useHomeFeedNotifications } from "@/features/notifications/hooks";
 import {
   listenForDesktopNotificationActions,
   revealDesktopAppWindow,
+  sendDesktopNotification,
   setDesktopAppBadgeCount,
   type DesktopNotificationTarget,
 } from "@/features/notifications/lib/desktop";
@@ -38,7 +39,7 @@ import { relayClient } from "@/shared/api/relayClient";
 import { useIdentityQuery } from "@/shared/api/hooks";
 import { useDeferredStartup } from "@/shared/hooks/useDeferredStartup";
 import { joinChannel } from "@/shared/api/tauri";
-import type { SearchHit } from "@/shared/api/types";
+import type { Channel, RelayEvent, SearchHit } from "@/shared/api/types";
 import { ChannelNavigationProvider } from "@/shared/context/ChannelNavigationContext";
 import { Button } from "@/shared/ui/button";
 import {
@@ -152,6 +153,37 @@ export function AppShell() {
     void homeFeedQuery.refetch();
   });
 
+  const handleDmNotification = React.useEffectEvent(
+    (event: RelayEvent, channel: Channel) => {
+      if (!notificationSettings.settings.desktopEnabled) {
+        return;
+      }
+
+      const channelName = channel.name?.trim() || "Direct message";
+      const content = event.content.trim();
+      const body =
+        content.length > 0
+          ? content.length > 140
+            ? `${content.slice(0, 137).trimEnd()}...`
+            : content
+          : "New message";
+
+      void sendDesktopNotification({
+        title: "Direct message",
+        body,
+        target: {
+          channelId: channel.id,
+          channelName,
+          content: event.content,
+          createdAt: event.created_at,
+          eventId: event.id,
+          kind: event.kind,
+          pubkey: event.pubkey,
+        },
+      });
+    },
+  );
+
   const channelsQuery = useChannelsQuery();
   const { refetch: refetchChannels } = channelsQuery;
   const channels = channelsQuery.data ?? [];
@@ -179,6 +211,7 @@ export function AppShell() {
     null,
     {
       currentPubkey: identityQuery.data?.pubkey,
+      onDmMessage: handleDmNotification,
       onLiveMention: refetchHomeFeedOnLiveMention,
     },
   );

desktop/src/features/channels/useLiveChannelUpdates.ts

@@ -11,6 +11,7 @@ import type { Channel, RelayEvent } from "@/shared/api/types";
 
 export type UseLiveChannelUpdatesOptions = {
   currentPubkey?: string;
+  onDmMessage?: (event: RelayEvent, channel: Channel) => void;
   onLiveMention?: () => void;
 };
 
@@ -27,19 +28,16 @@ function isExternalMentionEvent(event: RelayEvent, currentPubkey: string) {
   );
 }
 
-function rememberMentionEvent(
-  seenMentionEventIds: Set<string>,
-  eventId: string,
-): boolean {
-  if (seenMentionEventIds.has(eventId)) {
+function trackSeenEvent(seenEventIds: Set<string>, eventId: string): boolean {
+  if (seenEventIds.has(eventId)) {
     return false;
   }
 
-  seenMentionEventIds.add(eventId);
-  if (seenMentionEventIds.size > 200) {
-    const oldestEventId = seenMentionEventIds.values().next().value;
+  seenEventIds.add(eventId);
+  if (seenEventIds.size > 200) {
+    const oldestEventId = seenEventIds.values().next().value;
     if (oldestEventId) {
-      seenMentionEventIds.delete(oldestEventId);
+      seenEventIds.delete(oldestEventId);
     }
   }
 
@@ -64,6 +62,24 @@ export function useLiveChannelUpdates(
       ),
     [channels],
   );
+  const dmChannelMap = React.useMemo(
+    () =>
+      new Map(
+        channels
+          .filter((channel) => channel.channelType === "dm")
+          .map((channel) => [channel.id, channel]),
+      ),
+    [channels],
+  );
+  const seenDmEventIdsRef = React.useRef(new Set<string>());
+  const dmSubscriptionStartedAtRef = React.useRef(0);
+
+  // Reset subscription timestamp when identity changes.
+  React.useEffect(() => {
+    void normalizedCurrentPubkey;
+    dmSubscriptionStartedAtRef.current = 0;
+  }, [normalizedCurrentPubkey]);
+
   // Effect deps use primitive keys so refetches that produce new refs with
   // identical contents don't churn subscriptions. The Set/array memos are
   // still handy for closure reads via useEffectEvent.
@@ -73,9 +89,51 @@ export function useLiveChannelUpdates(
     [channels],
   );
 
+  const handleDmEvent = React.useEffectEvent((event: RelayEvent) => {
+    // Suppress backlog events that predate our subscription — these are
+    // historical replays, not live messages.
+    if (event.created_at < dmSubscriptionStartedAtRef.current) {
+      return;
+    }
+
+    const channelId = getChannelIdFromTags(event.tags);
+    if (!channelId) {
+      return;
+    }
+
+    if (!isExternalMentionEvent(event, normalizedCurrentPubkey)) {
+      return;
+    }
+
+    const dmChannel = dmChannelMap.get(channelId);
+    if (!dmChannel) {
+      return;
+    }
+
+    if (!trackSeenEvent(seenDmEventIdsRef.current, event.id)) {
+      return;
+    }
+
+    // Don't fire a notification for the channel the user is already viewing.
+    if (channelId === activeChannelId) {
+      return;
+    }
+
+    options.onDmMessage?.(event, dmChannel);
+  });
+
   const handleIncomingMessage = React.useEffectEvent((event: RelayEvent) => {
     const channelId = getChannelIdFromTags(event.tags);
-    if (!channelId || channelId === activeChannelId) {
+    if (!channelId) {
+      return;
+    }
+
+    // Track DM events even for the active channel so the dedup set stays
+    // current. The handler itself skips firing the notification callback
+    // when the user is already viewing the DM.
+    handleDmEvent(event);
+
+    if (channelId === activeChannelId) {
       return;
     }
 
@@ -104,7 +162,7 @@ export function useLiveChannelUpdates(
       return;
     }
 
-    if (!rememberMentionEvent(seenMentionEventIdsRef.current, event.id)) {
+    if (!trackSeenEvent(seenMentionEventIdsRef.current, event.id)) {
       return;
     }
 
@@ -114,6 +172,10 @@ export function useLiveChannelUpdates(
   React.useEffect(() => {
     return relayClient.subscribeToReconnects(() => {
       void queryClient.invalidateQueries({ queryKey: channelsQueryKey });
+
+      // Update the subscription timestamp so replayed backlog events
+      // (which have created_at in the past) are naturally suppressed.
+      dmSubscriptionStartedAtRef.current = Math.floor(Date.now() / 1000);
     });
   }, [queryClient]);
 
@@ -125,6 +187,10 @@ export function useLiveChannelUpdates(
     let isDisposed = false;
     let cleanup: (() => Promise<void>) | undefined;
 
+    // Record the subscription start time so handleDmEvent can distinguish
+    // backlog replays (created_at < startedAt) from live messages.
+    dmSubscriptionStartedAtRef.current = Math.floor(Date.now() / 1000);
+
     relayClient
       .subscribeToAllStreamMessages((event) => {
         if (!isDisposed) {

desktop/src/features/notifications/hooks.ts

@@ -285,13 +285,12 @@ export function useFeedDesktopNotifications(
   settings: NotificationSettings,
 ) {
   const normalizedPubkey = pubkey?.trim().toLowerCase() ?? "";
-  const seenItemIdsRef = React.useRef<Set<string>>(new Set());
-  const hasInitializedFeedRef = React.useRef(false);
+  const seenItemIdsRef = React.useRef<Set<string>>(
+    new Set(readStoredSeenFeedIds(normalizedPubkey)),
+  );
 
   React.useEffect(() => {
-    void normalizedPubkey;
-    seenItemIdsRef.current = new Set();
-    hasInitializedFeedRef.current = false;
+    seenItemIdsRef.current = new Set(readStoredSeenFeedIds(normalizedPubkey));
   }, [normalizedPubkey]);
 
   const deliverFeedNotification = React.useEffectEvent(
@@ -318,9 +317,12 @@ export function useFeedDesktopNotifications(
     }
 
     const currentFeedItems = collectHomeAlertItems(feed);
-    if (!hasInitializedFeedRef.current) {
-      hasInitializedFeedRef.current = true;
+
+    // Guard: empty seen set + populated feed means first load or cleared
+    // storage. Seed the seen set without notifying to prevent a flood.
+    if (seenItemIdsRef.current.size === 0 && currentFeedItems.length > 0) {
       seenItemIdsRef.current = new Set(currentFeedItems.map((item) => item.id));
+      writeStoredSeenFeedIds(normalizedPubkey, [...seenItemIdsRef.current]);
       return;
     }
 
@@ -337,9 +339,8 @@ export function useFeedDesktopNotifications(
     }
 
     // Prevent unbounded growth — keep only the most recent entries.
-    const MAX_SEEN_FEED_ITEMS = 500;
-    if (nextSeenItemIds.size > MAX_SEEN_FEED_ITEMS) {
-      const excess = nextSeenItemIds.size - MAX_SEEN_FEED_ITEMS;
+    if (nextSeenItemIds.size > HOME_FEED_SEEN_MAX_ITEMS) {
+      const excess = nextSeenItemIds.size - HOME_FEED_SEEN_MAX_ITEMS;
       let removed = 0;
       for (const id of nextSeenItemIds) {
         if (removed >= excess) break;
@@ -349,11 +350,18 @@ export function useFeedDesktopNotifications(
     }
 
     seenItemIdsRef.current = nextSeenItemIds;
+    writeStoredSeenFeedIds(normalizedPubkey, [...nextSeenItemIds]);
 
     for (const item of newItems) {
       void deliverFeedNotification(item);
     }
-  }, [feed, settings.desktopEnabled, settings.mentions, settings.needsAction]);
+  }, [
+    feed,
+    normalizedPubkey,
+    settings.desktopEnabled,
+    settings.mentions,
+    settings.needsAction,
+  ]);
 }
 
 export function useHomeFeedNotificationState(

desktop/src/features/notifications/lib/feed.ts

@@ -7,6 +7,10 @@ export function notificationTitle(item: FeedItem) {
     ? ` in #${item.channelName.trim()}`
     : "";
 
+  if (item.channelType === "dm") {
+    return "Direct message";
+  }
+
   if (item.category === "mention") {
     return `@Mention${channelLabel}`;
   }
@@ -43,8 +47,12 @@ export function eligibleFeedNotificationItems(
 ) {
   const items: FeedItem[] = [];
 
+  // DM notifications are handled by the real-time WebSocket hook, so we
+  // exclude DM items here to avoid duplicate toasts.
   if (options.mentions) {
-    items.push(...feed.feed.mentions);
+    items.push(
+      ...feed.feed.mentions.filter((item) => item.channelType !== "dm"),
+    );
   }
 
   if (options.needsAction) {

desktop/src/shared/api/tauri.ts

@@ -148,6 +148,7 @@ type RawFeedItem = {
   created_at: number;
   channel_id: string | null;
   channel_name: string;
+  channel_type: string;
   tags: string[][];
   category: "mention" | "needs_action" | "activity" | "agent_activity";
 };
@@ -394,6 +395,7 @@ function fromRawFeedItem(item: RawFeedItem) {
     createdAt: item.created_at,
     channelId: item.channel_id,
     channelName: item.channel_name,
+    channelType: item.channel_type,
     tags: item.tags,
     category: item.category,
   };

desktop/src/shared/api/types.ts

@@ -178,6 +178,7 @@ export type FeedItem = {
   createdAt: number;
   channelId: string | null;
   channelName: string;
+  channelType?: string;
   tags: string[][];
   category: FeedItemCategory;
 };

desktop/src/testing/e2eBridge.ts

@@ -167,6 +167,7 @@ type RawFeedItem = {
   created_at: number;
   channel_id: string | null;
   channel_name: string;
+  channel_type?: string;
   tags: string[][];
   category: "mention" | "needs_action" | "activity" | "agent_activity";
 };