feat(buzz): surface Buzz DMs in the Notification feed

A Buzz DM is a relay-authoritative NIP-29 group whose messages carry no `p` tag, so nothing made them eligible for the Notifications tab and nothing fetched them app-wide (discovery was scoped to the open DM inbox, which only pulls 44100 + 39000 — never the message bodies). Two halves fix that: - NotificationFeedFilter now early-accepts a group chat message (kind-9 or kind-40002 — the deployed relay uses both) when it resolves to a `t=dm` channel whose 39000 participants include me, honoring the same "Messages in notifications" toggle and never notifying for my own message. LocalCache gains `getRelayGroupChannelForContent`, the read-only reverse-lookup this needs (same serving-relay-then-single-channel keying as the consume path). - An always-on discovery (BuzzDmDiscoveryPreload) subscribes 44100 #p=me across joined workspaces into the new BuzzDmChannels registry and fetches each DM's 39000 directory; BuzzDmJoinedChatTailFilterAssembler then keeps those channels' recent messages warm app-wide (reusing the joined-group #h tail), excluding hidden DMs. Both mount in LoggedInPage. This is what makes a Buzz DM show on Notifications / in push without opening the conversation. Tests: BuzzDmChannels registry; and a LocalCache resolution test proving a 40002 and a kind-9 message both resolve back to their DM channel (and a non-dm channel does not). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-08-10 16:33:27 +00:00 · 2026-07-23 12:03:56 -04:00
parent 20384f5481
commit eeeaab3c43
10 changed files with 603 additions and 0 deletions
@@ -730,6 +730,18 @@ object LocalCache : ILocalCache, ICacheProvider {
    /** Every relay group we know of that is hosted on [relay] (its channel directory). */
    fun getRelayGroupChannelsOnRelay(relay: NormalizedRelayUrl): List<RelayGroupChannel> = relayGroupChannels.filter { key, _ -> key.relayUrl == relay }

+    /**
+     * The [RelayGroupChannel] a group-scoped content [note] belongs to, resolved the same way
+     * [attachToRelayGroupIfScoped] keyed it: the serving-relay key first (fast O(1)), then the single
+     * channel bearing this group id when the note has no usable provenance relay. Read-only — used by
+     * feed filters that need a note's channel without scanning every channel's timeline.
+     */
+    fun getRelayGroupChannelForContent(note: Note): RelayGroupChannel? {
+        val groupId = note.event?.groupId() ?: return null
+        note.relays.firstNotNullOfOrNull { getRelayGroupChannelIfExists(GroupId(groupId, it)) }?.let { return it }
+        return relayGroupChannels.filter { key, _ -> key.id == groupId }.singleOrNull()
+    }
+
    fun getLiveActivityChannelIfExists(key: Address): LiveActivitiesChannel? = liveChatChannels.get(key)

    fun getNoteIfExists(event: Event): Note? =
@@ -39,6 +39,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.privateDM.datasource.
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.concord.datasource.ConcordChannelFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.concord.datasource.ConcordChannelHistoryFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.datasource.ChannelFilterAssembler
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.BuzzDmJoinedChatTailFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RelayGroupCardWarmupFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RelayGroupJoinedChatTailFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RelayGroupJoinedStateFilterAssembler
@@ -139,6 +140,7 @@ class RelaySubscriptionsCoordinator(
    // amethyst/plans/2026-07-18-nip29-group-chat-subscriptions.md).
    val relayGroupJoinedState = RelayGroupJoinedStateFilterAssembler(client) // always-on: joined groups' metadata/roster/roles/pins
    val relayGroupJoinedChatTail = RelayGroupJoinedChatTailFilterAssembler(client) // always-on: batched #h recent-tail for Messages previews
+    val buzzDmJoinedChatTail = BuzzDmJoinedChatTailFilterAssembler(client) // always-on: batched #h recent-tail for the viewer's Buzz DM channels
    val relayGroupOpenChatTail = RelayGroupOpenChatTailFilterAssembler(client) // the open group's recent chat (covers non-joined)
    val relayGroupOpenChatHistory = RelayGroupOpenChatHistoryFilterAssembler(client) // the open group's on-demand backward history pager

@@ -218,6 +220,7 @@ class RelaySubscriptionsCoordinator(
            relayGroupsDiscovery,
            relayGroupJoinedState,
            relayGroupJoinedChatTail,
+            buzzDmJoinedChatTail,
            relayGroupOpenChatTail,
            relayGroupOpenChatHistory,
            concordChannels,
@@ -51,7 +51,9 @@ import com.vitorpamplona.amethyst.service.resourceusage.innermostSigner
 import com.vitorpamplona.amethyst.ui.navigation.AppNavigation
 import com.vitorpamplona.amethyst.ui.navigation.routes.Route
 import com.vitorpamplona.amethyst.ui.screen.AccountSessionManager
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.buzz.BuzzDmDiscoveryPreload
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.concord.datasource.ConcordChannelPreload
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.BuzzDmJoinedChatTailPreload
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RelayGroupJoinedChatTailPreload
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RelayGroupJoinedStatePreload
 import com.vitorpamplona.quartz.nip55AndroidSigner.client.IActivityLauncher
@@ -103,6 +105,11 @@ fun LoggedInPage(
    RelayGroupJoinedStatePreload(accountViewModel)
    RelayGroupJoinedChatTailPreload(accountViewModel)

+    // Discover the viewer's Buzz DM channels (44100 #p=me) across joined workspaces and keep their
+    // messages warm app-wide, so a Buzz DM shows on the Notifications tab / in push without opening it.
+    BuzzDmDiscoveryPreload(accountViewModel)
+    BuzzDmJoinedChatTailPreload(accountViewModel)
+
    // Foreground-only loaders: follows-outbox finder + random-relay notifications.
    // Pauses on ON_STOP, resumes on ON_START.
    AccountForegroundFilterAssemblerSubscription(accountViewModel)
@@ -0,0 +1,131 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.ui.screen.loggedIn.buzz
+
+import androidx.compose.runtime.Composable
+import androidx.compose.runtime.LaunchedEffect
+import androidx.compose.runtime.getValue
+import androidx.lifecycle.compose.collectAsStateWithLifecycle
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmChannels
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzWorkspaces
+import com.vitorpamplona.amethyst.commons.relayauth.RelayAuthDecision
+import com.vitorpamplona.amethyst.model.Account
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource.RELAY_GROUP_METADATA_KINDS
+import com.vitorpamplona.quartz.buzz.dvDmVisibility.DmVisibilityEvent
+import com.vitorpamplona.quartz.buzz.notifications.MemberAddedNotificationEvent
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllWithHooks
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.subscribeAsFlow
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+
+/**
+ * Always-on discovery of the viewer's Buzz **DM channels** across every joined workspace relay, mounted
+ * once high in the logged-in tree ([com.vitorpamplona.amethyst.ui.screen.loggedIn.LoggedInPage]).
+ *
+ * The deployed relay does not expose a queryable DM list; it addresses each member a kind-44100
+ * member-added notification (`#p` = me). This warm-auths a `#p=me` fetch of 44100 (+ the 30622 visibility
+ * snapshot) across the joined relays, records the channels into [BuzzDmChannels], fetches each channel's
+ * 39000-39003 directory (so its `t`=dm marker + participants land in `LocalCache`), and keeps a live
+ * `#p=me` 44100 subscription open for new DMs. The companion [BuzzDmJoinedChatTailPreload] then keeps the
+ * discovered channels' messages warm app-wide — which is what lets a Buzz DM show on the Notifications tab
+ * and in push without the viewer opening the conversation first.
+ *
+ * This mirrors [BuzzDmListViewModel]'s discovery, but account-scoped and always-on rather than bound to
+ * the open inbox screen; the inbox keeps its own scoped copy for its per-relay projection.
+ */
+@Composable
+fun BuzzDmDiscoveryPreload(accountViewModel: AccountViewModel) {
+    val account = accountViewModel.account
+    val joined by BuzzWorkspaces.flow.collectAsStateWithLifecycle()
+
+    // Restart the whole discovery (initial warm-auth fetch + live 44100 subs) whenever the joined
+    // workspace set changes; the LaunchedEffect scope owns the live subscriptions and cancels them on
+    // account switch or dispose.
+    LaunchedEffect(account, joined) {
+        if (joined.isEmpty()) return@LaunchedEffect
+        runBuzzDmDiscovery(account, joined)
+    }
+}
+
+/**
+ * Warm-auth the initial 44100/30622 `#p=me` read across [relays], record every discovered channel, fetch
+ * their directories, then keep a live 44100 subscription per relay open until the caller's scope is
+ * cancelled. Suspends for the lifetime of the live subscriptions.
+ */
+private suspend fun runBuzzDmDiscovery(
+    account: Account,
+    relays: Set<NormalizedRelayUrl>,
+) = coroutineScope {
+    val me = account.userProfile().pubkeyHex
+
+    // A joined workspace is first-party: pre-approve NIP-42 so the `#p=me` DM reads authenticate (the
+    // restore-from-disk path doesn't set this, unlike the inbox/import/console entry points).
+    relays.forEach { account.relayAuthLedger.setDecision(it.url, RelayAuthDecision.ALLOW) }
+
+    val discoveryFilters =
+        listOf(
+            Filter(kinds = listOf(MemberAddedNotificationEvent.KIND), tags = mapOf("p" to listOf(me))),
+            Filter(kinds = listOf(DmVisibilityEvent.KIND), tags = mapOf("p" to listOf(me))),
+        )
+    // `#p`-gated reads: use the warm-auth fetch so an `auth-required` CLOSED authenticates and retries
+    // rather than returning empty.
+    account.client.fetchAllWithHooks(
+        filters = relays.associateWith { discoveryFilters },
+        timeoutMs = 8_000,
+        pendingOnAuthRequired = true,
+    ) { relay, event ->
+        (event as? MemberAddedNotificationEvent)?.channel()?.let { BuzzDmChannels.record(me, it, relay) }
+        false
+    }
+    fetchDmMetadata(account, me)
+
+    relays.forEach { relay ->
+        launch {
+            val filter = Filter(kinds = listOf(MemberAddedNotificationEvent.KIND), tags = mapOf("p" to listOf(me)))
+            account.client.subscribeAsFlow(relay, filter).collect { events ->
+                var changed = false
+                events.filterIsInstance<MemberAddedNotificationEvent>().forEach { e ->
+                    e.channel()?.let { if (BuzzDmChannels.record(me, it, relay)) changed = true }
+                }
+                if (changed) fetchDmMetadata(account, me)
+            }
+        }
+    }
+}
+
+/** Fetch the NIP-29 directory (39000-39003) of every known DM channel so its `t`=dm marker + roster load. */
+private suspend fun fetchDmMetadata(
+    account: Account,
+    viewer: HexKey,
+) {
+    val byRelay =
+        BuzzDmChannels
+            .channelsFor(viewer)
+            .entries
+            .groupBy({ it.value }, { it.key })
+            .mapValues { (_, ids) -> listOf(Filter(kinds = RELAY_GROUP_METADATA_KINDS, tags = mapOf("d" to ids))) }
+    if (byRelay.isEmpty()) return
+    account.client.fetchAllWithHooks(filters = byRelay, timeoutMs = 8_000, pendingOnAuthRequired = true) { _, _ -> false }
+}
@@ -0,0 +1,106 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.relayGroup.datasource
+
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmChannels
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmRegistry
+import com.vitorpamplona.amethyst.commons.model.privateChats.DmHistoryTuning
+import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.ComposeSubscriptionManager
+import com.vitorpamplona.amethyst.commons.relayClient.paging.WindowLoadTracker
+import com.vitorpamplona.amethyst.commons.relayClient.paging.trackingListener
+import com.vitorpamplona.amethyst.model.Account
+import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUniqueIdEoseManager
+import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
+import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip51Lists.simpleGroupList.GroupTag
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.flow.StateFlow
+
+/** One screen's request to keep the viewer's Buzz DM channels' recent chat live. */
+class BuzzDmJoinedChatTailQueryState(
+    val account: Account,
+)
+
+/**
+ * Always-on **live tail** for the recent chat of every Buzz DM channel the viewer belongs to — the DM
+ * analog of [RelayGroupJoinedChatTailFilterAssembler]. A Buzz DM is a relay-authoritative NIP-29 group
+ * (UUID `h`), so its messages ride the exact same `#h`-scoped batched tail; only the channel *source*
+ * differs — [BuzzDmChannels] (populated by
+ * [com.vitorpamplona.amethyst.ui.screen.loggedIn.buzz.BuzzDmDiscoveryPreload]) rather than the published
+ * kind-10009 group list, because DM memberships are server-side and deliberately never published.
+ *
+ * Keeping these warm app-wide is what lets a Buzz DM surface on the Notifications tab and in push without
+ * the viewer opening the conversation. Hidden DMs (per the 30622 snapshot in [BuzzDmRegistry]) are
+ * excluded so a hidden conversation neither streams nor notifies.
+ */
+class BuzzDmJoinedChatTailFilterAssembler(
+    client: INostrClient,
+) : ComposeSubscriptionManager<BuzzDmJoinedChatTailQueryState>() {
+    val tail = BuzzDmJoinedChatTailSubAssembler(client, ::allKeys)
+
+    val group = listOf(tail)
+
+    override fun invalidateKeys() = invalidateFilters()
+
+    override fun invalidateFilters() = group.forEach { it.invalidateFilters() }
+
+    override fun destroy() = group.forEach { it.destroy() }
+}
+
+class BuzzDmJoinedChatTailSubAssembler(
+    client: INostrClient,
+    allKeys: () -> Set<BuzzDmJoinedChatTailQueryState>,
+) : PerUniqueIdEoseManager<BuzzDmJoinedChatTailQueryState, Account>(client, allKeys) {
+    private val windowLoad = WindowLoadTracker("buzzDm.preview.live")
+    val loadingMore: StateFlow<Boolean> = windowLoad.loading
+
+    override fun updateFilter(
+        key: BuzzDmJoinedChatTailQueryState,
+        since: SincePerRelayMap?,
+    ): List<RelayBasedFilter>? {
+        val me = key.account.userProfile().pubkeyHex
+        val hidden = BuzzDmRegistry.hiddenFor(me)
+        val channels = BuzzDmChannels.channelsFor(me).filterKeys { it !in hidden }
+        if (channels.isEmpty()) {
+            windowLoad.setExpectedRelays(emptySet())
+            return null
+        }
+
+        // Reuse the joined-group tail builder: one #h filter per host relay carrying every DM channel id
+        // on it, bounded by the shared recent floor (no per-channel limit, reconnect-safe).
+        val asTags = channels.map { (channelId, relay) -> GroupTag(channelId, relay.url) }
+        val filters = buildRelayGroupJoinedChatTailFilters(asTags, DmHistoryTuning.recentBoundary())
+        windowLoad.setExpectedRelays(filters.mapTo(mutableSetOf()) { it.relay })
+        return filters
+    }
+
+    override fun id(key: BuzzDmJoinedChatTailQueryState) = key.account
+
+    override fun newSub(key: BuzzDmJoinedChatTailQueryState): Subscription {
+        windowLoad.startLoading(key.account.scope)
+        return requestNewSubscription(
+            windowLoad.trackingListener { relay: NormalizedRelayUrl, filters -> newEose(key, relay, TimeUtils.now(), filters) },
+        )
+    }
+}
@@ -25,6 +25,8 @@ import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.remember
 import androidx.lifecycle.compose.collectAsStateWithLifecycle
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmChannels
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmRegistry
 import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.KeyDataSourceSubscription
 import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.LifecycleAwareKeyDataSourceSubscription
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
@@ -64,6 +66,26 @@ fun RelayGroupJoinedChatTailPreload(accountViewModel: AccountViewModel) {
    KeyDataSourceSubscription(state, dataSource)
 }

+/**
+ * Always-on preview **live tail** for the viewer's Buzz DM channels' recent chat, mounted alongside
+ * [RelayGroupJoinedChatTailPreload] — keeps discovered DM conversations warm app-wide so a Buzz DM can
+ * surface on the Notifications tab and in push without opening it. Re-derives whenever a DM is discovered
+ * ([com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmChannels]) or hidden/unhidden
+ * ([com.vitorpamplona.amethyst.commons.model.buzz.BuzzDmRegistry]).
+ */
+@Composable
+fun BuzzDmJoinedChatTailPreload(accountViewModel: AccountViewModel) {
+    val account = accountViewModel.account
+    val dataSource = accountViewModel.dataSources().buzzDmJoinedChatTail
+    val state = remember(account) { BuzzDmJoinedChatTailQueryState(account) }
+
+    val channels by BuzzDmChannels.flow.collectAsStateWithLifecycle()
+    val hidden by BuzzDmRegistry.hidden.collectAsStateWithLifecycle()
+    LaunchedEffect(channels, hidden) { dataSource.invalidateFilters() }
+
+    KeyDataSourceSubscription(state, dataSource)
+}
+
 /**
 * Mount on the open group chat screen to keep the *currently open* group's recent chat live — covers a
 * non-joined group opened by link (the batched preview tail is joined-only) and live updates. Lifecycle-
@@ -32,6 +32,9 @@ import com.vitorpamplona.amethyst.model.topNavFeeds.IFeedTopNavFilter
 import com.vitorpamplona.amethyst.ui.dal.AdditiveFeedFilter
 import com.vitorpamplona.amethyst.ui.dal.FilterByListParams
 import com.vitorpamplona.amethyst.ui.dal.sortedByDefaultFeedOrder
+import com.vitorpamplona.quartz.buzz.stream.StreamMessageV2Event
+import com.vitorpamplona.quartz.buzz.workspace.buzzParticipants
+import com.vitorpamplona.quartz.buzz.workspace.isBuzzDm
 import com.vitorpamplona.quartz.experimental.attestations.request.AttestationRequestEvent
 import com.vitorpamplona.quartz.experimental.audio.track.AudioTrackEvent
 import com.vitorpamplona.quartz.experimental.forks.IForkableEvent
@@ -373,6 +376,20 @@ class NotificationFeedFilter(
        return collection.filterTo(HashSet()) { acceptableEvent(it, filterParams) }
    }

+    /**
+     * The Buzz DM message [note] targets me: a kind-40002 in a `t=dm` channel whose 39000 participants
+     * include [me]. DM messages carry no `p` tag, so being a participant of the DM channel is the
+     * relevance signal — like a Marmot/Concord message. False for non-DM channels, for metadata we
+     * haven't loaded yet, or when I'm not a participant. (Own-message and toggle checks live at the call site.)
+     */
+    private fun isBuzzDmForMe(
+        note: Note,
+        me: HexKey,
+    ): Boolean {
+        val md = LocalCache.getRelayGroupChannelForContent(note)?.event ?: return false
+        return md.isBuzzDm() && md.buzzParticipants().contains(me)
+    }
+
    fun acceptableEvent(
        it: Note,
        filterParams: FilterByListParams,
@@ -403,6 +420,16 @@ class NotificationFeedFilter(

        val noteEvent = it.event

+        // Buzz DM: a group chat message in a `t=dm` channel whose 39000 participants include me. A Buzz
+        // relay carries DM messages as either kind-9 (NIP-29 chat) or kind-40002 (stream message v2), and
+        // neither `p`-tags the recipient, so being a participant of the DM channel is the relevance signal
+        // — like a Marmot/Concord message above. Honors the same "Messages in notifications" toggle, and
+        // never notifies for my own message.
+        if ((noteEvent is StreamMessageV2Event || noteEvent is ChatEvent) && isBuzzDmForMe(it, loggedInUserHex)) {
+            if (!showMessages) return false
+            return it.author?.pubkeyHex != loggedInUserHex
+        }
+
        if (!showMessages &&
            (
                noteEvent is ChatMessageEvent ||
@@ -0,0 +1,145 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.model
+
+import android.os.Looper
+import com.vitorpamplona.amethyst.commons.model.buzz.BuzzRelayDialect
+import com.vitorpamplona.quartz.buzz.stream.StreamMessageV2Event
+import com.vitorpamplona.quartz.buzz.workspace.buzzParticipants
+import com.vitorpamplona.quartz.buzz.workspace.isBuzzDm
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import com.vitorpamplona.quartz.nip29RelayGroups.GroupId
+import com.vitorpamplona.quartz.nip29RelayGroups.metadata.GroupMetadataEvent
+import com.vitorpamplona.quartz.nip29RelayGroups.tags.GroupIdTag
+import com.vitorpamplona.quartz.nipC7Chats.ChatEvent
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.mockkStatic
+import io.mockk.unmockkStatic
+import kotlinx.coroutines.runBlocking
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import java.util.UUID
+
+/**
+ * The resolution the Notification feed relies on to surface a Buzz DM: given a bare kind-40002 message
+ * [Note], [LocalCache.getRelayGroupChannelForContent] finds the group channel it belongs to, and that
+ * channel's kind-39000 metadata answers `isBuzzDm()` + `buzzParticipants()`. This is what
+ * `NotificationFeedFilter.isBuzzDmForMe` composes; the per-kind acceptance is exercised at the filter.
+ */
+class BuzzDmNotificationResolutionTest {
+    private val buzzRelay = RelayUrlNormalizer.normalizeOrNull("wss://buzz.example.team/")!!
+    private val me = NostrSignerInternal(KeyPair())
+    private val other = NostrSignerInternal(KeyPair())
+
+    @Before
+    fun setup() {
+        // LocalCache.consume refuses the main thread; plain JVM tests have no Looper (null == null reads
+        // as "main"). Distinct mocks make it a worker thread. See BuzzWorkspaceChannelTest.
+        mockkStatic(Looper::class)
+        every { Looper.myLooper() } returns mockk<Looper>()
+        every { Looper.getMainLooper() } returns mockk<Looper>()
+        BuzzRelayDialect.clearForTesting()
+    }
+
+    @After
+    fun tearDown() {
+        unmockkStatic(Looper::class)
+        BuzzRelayDialect.clearForTesting()
+    }
+
+    private suspend fun dmMetadata(
+        channelId: String,
+        participants: List<String>,
+    ) = me.sign(
+        GroupMetadataEvent.build(channelId, name = "DM") {
+            add(arrayOf("t", "dm"))
+            participants.forEach { add(arrayOf("p", it)) }
+        },
+    )
+
+    // Attach a 40002 by consuming it (which materializes the group channel) and then set its 39000
+    // metadata directly — the real 39000 consume gates on Amethyst.instance.nip11Cache, which isn't
+    // available in a plain JVM test (see BuzzWorkspaceChannelTest). We still drive the real resolver.
+    private fun consumeMessageInto(
+        channelId: String,
+        metadata: GroupMetadataEvent,
+        message: Event,
+    ): Note {
+        LocalCache.checkDeletionAndConsume(message, buzzRelay, false)
+        LocalCache.getRelayGroupChannelIfExists(GroupId(channelId, buzzRelay))!!.event = metadata
+        return LocalCache.getNoteIfExists(message.id)!!
+    }
+
+    @Test
+    fun `a 40002 in a dm channel resolves to its channel whose metadata says dm and lists me`() =
+        runBlocking {
+            val channelId = UUID.randomUUID().toString()
+            val meHex = me.pubKey
+            val otherHex = other.pubKey
+
+            val message = other.sign(StreamMessageV2Event.build(channelId, "hey"))
+            val note = consumeMessageInto(channelId, dmMetadata(channelId, listOf(meHex, otherHex)), message)
+
+            val metadata = LocalCache.getRelayGroupChannelForContent(note)?.event
+            assertNotNull("the message resolves back to its DM channel", metadata)
+            assertTrue("the channel metadata is marked t=dm", metadata!!.isBuzzDm())
+            assertTrue("I am one of the DM participants", metadata.buzzParticipants().contains(meHex))
+            assertEquals(setOf(meHex, otherHex), metadata.buzzParticipants().toSet())
+        }
+
+    @Test
+    fun `a kind-9 chat message in a dm channel also resolves to its dm channel`() =
+        runBlocking {
+            // The deployed Buzz relay carries DM messages as kind-9 too, not only 40002; the resolver is
+            // kind-agnostic (keys off the h tag) so both surface the same DM channel.
+            val channelId = UUID.randomUUID().toString()
+            val meHex = me.pubKey
+            val chat = other.sign(ChatEvent.build("hey") { add(GroupIdTag.assemble(channelId)) })
+
+            val note = consumeMessageInto(channelId, dmMetadata(channelId, listOf(meHex, other.pubKey)), chat)
+
+            val metadata = LocalCache.getRelayGroupChannelForContent(note)?.event
+            assertNotNull("the kind-9 message resolves back to its DM channel", metadata)
+            assertTrue(metadata!!.isBuzzDm())
+        }
+
+    @Test
+    fun `a 40002 in a non-dm channel resolves to a channel that is not a dm`() =
+        runBlocking {
+            val channelId = UUID.randomUUID().toString()
+            // A plain (non-dm) channel: 39000 without the t=dm marker.
+            val message = other.sign(StreamMessageV2Event.build(channelId, "gm"))
+            val note = consumeMessageInto(channelId, me.sign(GroupMetadataEvent.build(channelId, name = "general")), message)
+
+            val metadata = LocalCache.getRelayGroupChannelForContent(note)?.event
+            assertNotNull(metadata)
+            assertFalse("a general channel is not a DM", metadata!!.isBuzzDm())
+        }
+}
@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.model.buzz
+
+import com.vitorpamplona.amethyst.commons.util.KmpLock
+import com.vitorpamplona.amethyst.commons.util.withLock
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+
+/**
+ * App-wide, per-viewer set of the Buzz **DM channels** the viewer belongs to — each a relay-generated
+ * UUID (`channelId`) plus the relay that vouched for it.
+ *
+ * The deployed relay does not emit a queryable kind-41001; it enumerates a member's channels by
+ * addressing each one a kind-44100 member-added notification (`#p` = me). The always-on
+ * `BuzzDmDiscovery` subscribes that `#p=me` stream across the joined Buzz relays and records the
+ * result here, so the always-on DM chat tail can keep those channels' messages warm in `LocalCache`
+ * app-wide (which is what lets a Buzz DM surface on the Notifications tab and in push without the
+ * viewer opening the conversation first).
+ *
+ * Kept **per-viewer** because the 44100 stream is `#p`-gated to its owner and the process can switch
+ * accounts. Mutations are lock-guarded because discovery runs across several relay reader threads.
+ * Like [BuzzDmRegistry] / [BuzzWorkspaces], a process-wide singleton.
+ */
+object BuzzDmChannels {
+    private val lock = KmpLock()
+    private val byViewer = HashMap<HexKey, MutableMap<String, NormalizedRelayUrl>>()
+    private val mutableFlow = MutableStateFlow<Map<HexKey, Map<String, NormalizedRelayUrl>>>(emptyMap())
+
+    /** Per-viewer discovered DM channels (`channelId` -> the relay it was discovered on). */
+    val flow: StateFlow<Map<HexKey, Map<String, NormalizedRelayUrl>>> = mutableFlow
+
+    /**
+     * Records that [viewer] belongs to DM channel [channelId] on [relay]. Returns true when this is a
+     * newly seen (viewer, channel) pair so callers can trigger a re-subscribe; a repeat that only
+     * re-confirms the same relay returns false and does not churn the flow.
+     */
+    fun record(
+        viewer: HexKey,
+        channelId: String,
+        relay: NormalizedRelayUrl,
+    ): Boolean =
+        lock.withLock {
+            val channels = byViewer.getOrPut(viewer) { mutableMapOf() }
+            if (channels[channelId] == relay) return@withLock false
+            channels[channelId] = relay
+            mutableFlow.value = snapshot()
+            true
+        }
+
+    /** The DM channels [viewer] is in (`channelId` -> relay), possibly empty. */
+    fun channelsFor(viewer: HexKey): Map<String, NormalizedRelayUrl> = mutableFlow.value[viewer] ?: emptyMap()
+
+    private fun snapshot(): Map<HexKey, Map<String, NormalizedRelayUrl>> = byViewer.mapValues { it.value.toMap() }
+
+    /** Test-only: clears all registry state so unit tests don't leak into each other. */
+    fun clearForTesting() =
+        lock.withLock {
+            byViewer.clear()
+            mutableFlow.value = emptyMap()
+        }
+}
@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.model.buzz
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class BuzzDmChannelsTest {
+    private val alice = "a".repeat(64)
+    private val bob = "b".repeat(64)
+    private val relayA = RelayUrlNormalizer.normalizeOrNull("wss://a.example.team/")!!
+    private val relayB = RelayUrlNormalizer.normalizeOrNull("wss://b.example.team/")!!
+
+    @BeforeTest fun setup() = BuzzDmChannels.clearForTesting()
+
+    @AfterTest fun teardown() = BuzzDmChannels.clearForTesting()
+
+    @Test
+    fun recordsAndReadsAChannel() {
+        assertTrue(BuzzDmChannels.record(alice, "chan-1", relayA), "a first sighting is new")
+        assertEquals(mapOf("chan-1" to relayA), BuzzDmChannels.channelsFor(alice))
+    }
+
+    @Test
+    fun reRecordingTheSameRelayIsNotNewAndDoesNotChurn() {
+        BuzzDmChannels.record(alice, "chan-1", relayA)
+        val before = BuzzDmChannels.flow.value
+        assertFalse(BuzzDmChannels.record(alice, "chan-1", relayA), "re-confirming the same (channel, relay) is not new")
+        assertTrue(before === BuzzDmChannels.flow.value, "the flow instance is unchanged on a no-op")
+    }
+
+    @Test
+    fun aChannelMovingRelaysIsRecordedAsNew() {
+        BuzzDmChannels.record(alice, "chan-1", relayA)
+        assertTrue(BuzzDmChannels.record(alice, "chan-1", relayB), "a new relay for a known channel is a change")
+        assertEquals(mapOf("chan-1" to relayB), BuzzDmChannels.channelsFor(alice))
+    }
+
+    @Test
+    fun channelsArePerViewer() {
+        BuzzDmChannels.record(alice, "chan-1", relayA)
+        assertEquals(mapOf("chan-1" to relayA), BuzzDmChannels.channelsFor(alice))
+        assertEquals(emptyMap(), BuzzDmChannels.channelsFor(bob))
+    }
+}