Merge pull request #2534 from vitorpamplona/claude/add-notification-filtering-RF0Td

Refactor notification pipeline to use cache observer pattern
2026-04-23 18:23:09 -04:00
parent bbdcd8adbe dd30f03bba
commit 776160f6ef
11 changed files with 398 additions and 212 deletions
@@ -76,7 +76,10 @@ class PushMessageReceiver : MessagingReceiver() {
    private suspend fun receiveIfNew(event: GiftWrapEvent) {
        if (eventCache.get(event.id) == null) {
            eventCache.put(event.id, event.id)
-            EventNotificationConsumer(appContext).consume(event)
+            // The push server re-wraps the real GiftWrap and strips the p tag;
+            // unwrap the outer layer, feed the inner into LocalCache, and let
+            // the usual Account → EventProcessor chain take over.
+            PushWrapDecryptor.unwrapAndFeed(event)
        }
    }

@@ -50,6 +50,7 @@ import com.vitorpamplona.amethyst.service.images.ImageLoaderSetup
 import com.vitorpamplona.amethyst.service.images.ThumbnailDiskCache
 import com.vitorpamplona.amethyst.service.location.LocationState
 import com.vitorpamplona.amethyst.service.notifications.AlwaysOnNotificationServiceManager
+import com.vitorpamplona.amethyst.service.notifications.NotificationDispatcher
 import com.vitorpamplona.amethyst.service.notifications.PokeyReceiver
 import com.vitorpamplona.amethyst.service.okhttp.DualHttpClientManager
 import com.vitorpamplona.amethyst.service.okhttp.DualHttpClientManagerForRelays
@@ -397,6 +398,11 @@ class AppModules(
    // Manages always-on notification service lifecycle
    val alwaysOnNotificationServiceManager = AlwaysOnNotificationServiceManager(appContext, applicationIOScope)

+    // Observes LocalCache for notification-relevant events and routes them to
+    // EventNotificationConsumer. Sources: FCM, UnifiedPush, Pokey, active relay
+    // subscriptions, and NotificationRelayService.
+    val notificationDispatcher = NotificationDispatcher(appContext, applicationIOScope)
+
    // Organizes cache clearing
    val trimmingService by
        lazy {
@@ -494,6 +500,9 @@ class AppModules(
        // registers to receive events
        pokeyReceiver.register(appContext)

+        // starts observing LocalCache for notification-worthy events
+        notificationDispatcher.start()
+
        // Watch for account login and start/stop always-on notification service
        applicationIOScope.launch {
            sessionManager.accountContent.collectLatest { state ->
@@ -515,6 +524,7 @@ class AppModules(

    fun terminate(appContext: Context) {
        pokeyReceiver.unregister(appContext)
+        notificationDispatcher.stop()
        BackgroundMedia.removeBackgroundControllerAndReleaseIt()
        PlaybackServiceClient.shutdown()
        alwaysOnNotificationServiceManager.stop()
@@ -33,6 +33,7 @@ import com.vitorpamplona.amethyst.commons.model.nip28PublicChats.PublicChatChann
 import com.vitorpamplona.amethyst.commons.model.nip53LiveActivities.LiveActivitiesChannel
 import com.vitorpamplona.amethyst.commons.model.observables.CreatedAtIdHexComparator
 import com.vitorpamplona.amethyst.commons.model.observables.EventListMatchingFilter
+import com.vitorpamplona.amethyst.commons.model.observables.NewEventMatchingFilter
 import com.vitorpamplona.amethyst.commons.model.observables.NoteListMatchingFilter
 import com.vitorpamplona.amethyst.commons.model.observables.Observable
 import com.vitorpamplona.amethyst.commons.model.privateChats.ChatroomList
@@ -66,6 +67,7 @@ import com.vitorpamplona.quartz.experimental.profileGallery.ProfileGalleryEntryE
 import com.vitorpamplona.quartz.experimental.zapPolls.ZapPollEvent
 import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent
 import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent
+import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent
 import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent
 import com.vitorpamplona.quartz.nip01Core.core.Address
 import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent
@@ -380,6 +382,25 @@ object LocalCache : ILocalCache, ICacheProvider {
            }
        }.buffer(kotlinx.coroutines.channels.Channel.CONFLATED)

+    /**
+     * Emits each new event that matches the filter, one at a time, as it is
+     * inserted into the cache. Unlike [observeEvents], this does not accumulate
+     * a list — useful for per-event reactive pipelines like notifications.
+     */
+    fun <T : Event> observeNewEvents(filter: Filter): Flow<T> =
+        callbackFlow {
+            val newFilter =
+                NewEventMatchingFilter<T>(filter) {
+                    trySend(it)
+                }
+
+            observables.put(newFilter, newFilter)
+
+            awaitClose {
+                observables.remove(newFilter)
+            }
+        }
+
    @Suppress("UNCHECKED_CAST")
    fun <T : Event> observeLatestEvent(filter: Filter) = observeEvents<T>(filter).map { it.firstOrNull() }

@@ -2788,6 +2809,7 @@ object LocalCache : ILocalCache, ICacheProvider {
                is VoiceReplyEvent -> consumeRegularEvent(event, relay, wasVerified)
                is WakeUpEvent -> consumeRegularEvent(event, relay, wasVerified)
                is WebBookmarkEvent -> consumeBaseReplaceable(event, relay, wasVerified)
+                is WelcomeEvent -> consumeRegularEvent(event, relay, wasVerified)
                is WikiNoteEvent -> consume(event, relay, wasVerified)
                is PaymentTargetsEvent -> consume(event, relay, wasVerified)
                else -> Log.w("Event Not Supported") { "From ${relay?.url}: ${event.toJson()}" }.let { false }
@@ -45,10 +45,10 @@ import com.vitorpamplona.amethyst.service.notifications.NotificationUtils.sendZa
 import com.vitorpamplona.amethyst.service.relayClient.authCommand.model.ScreenAuthAccount
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.EventFinderQueryState
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderQueryState
+import com.vitorpamplona.amethyst.ui.MainActivity
 import com.vitorpamplona.amethyst.ui.note.showAmount
 import com.vitorpamplona.amethyst.ui.stringRes
 import com.vitorpamplona.quartz.experimental.notifications.wake.WakeUpEvent
-import com.vitorpamplona.quartz.marmot.WelcomeResult
 import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
@@ -63,8 +63,6 @@ import com.vitorpamplona.quartz.nip21UriScheme.toNostrUri
 import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
 import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent
 import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
-import com.vitorpamplona.quartz.nip59Giftwrap.seals.SealedRumorEvent
-import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
 import com.vitorpamplona.quartz.nip64Chess.baseEvent.BaseChessEvent
 import com.vitorpamplona.quartz.nip64Chess.challenge.accept.LiveChessGameAcceptEvent
 import com.vitorpamplona.quartz.nip64Chess.move.LiveChessMoveEvent
@@ -115,96 +113,73 @@ class EventNotificationConsumer(
        }
    }

-    suspend fun consume(event: GiftWrapEvent) =
+    /**
+     * Entry point for notification-relevant events arriving into [LocalCache]
+     * from any source (FCM push, UnifiedPush, Pokey, active relay subscriptions,
+     * NotificationRelayService). The [NotificationDispatcher] only invokes this
+     * after [Account.newNotesPreProcessor] has fully unwrapped wraps and seals,
+     * so this method receives the final inner payload directly.
+     *
+     * Matches the event to a logged-in account by its `p` tags and dispatches
+     * to [dispatchForAccount].
+     */
+    suspend fun consumeFromCache(event: Event) =
        withWakeLock {
-            Log.d(TAG, "New Notification Arrived")
+            Log.d(TAG) { "New Notification from cache: kind=${event.kind} id=${event.id}" }

-            // PushNotification Wraps don't include a receiver.
-            // Test with all logged in accounts
-            var matchAccount = false
-            LocalPreferences.allSavedAccounts().forEach {
-                if (!matchAccount && (it.hasPrivKey || it.loggedInWithExternalSigner)) {
-                    LocalPreferences.loadAccountConfigFromEncryptedStorage(it.npub)?.let { acc ->
-                        Log.d(TAG) { "New Notification Testing if for ${it.npub}" }
-                        try {
-                            val account = Amethyst.instance.accountsCache.loadAccount(acc)
-                            consumeIfMatchesAccount(event, account)
-                            matchAccount = true
-                        } catch (e: Exception) {
-                            if (e is CancellationException) throw e
-                            Log.d(TAG) { "Message was not for user ${it.npub}: ${e.message}" }
-                        }
-                    }
+            if (!notificationManager().areNotificationsEnabled()) return@withWakeLock
+
+            val taggedNpubs =
+                event
+                    .taggedUserIds()
+                    .mapTo(mutableSetOf()) { LocalCache.getOrCreateUser(it).pubkeyNpub() }
+            if (taggedNpubs.isEmpty()) return@withWakeLock
+
+            LocalPreferences.allSavedAccounts().forEach { savedAccount ->
+                if (!savedAccount.hasPrivKey && !savedAccount.loggedInWithExternalSigner) return@forEach
+                if (savedAccount.npub !in taggedNpubs) return@forEach
+
+                val accountSettings = LocalPreferences.loadAccountConfigFromEncryptedStorage(savedAccount.npub) ?: return@forEach
+                try {
+                    val account = Amethyst.instance.accountsCache.loadAccount(accountSettings)
+                    dispatchForAccount(event, account)
+                } catch (e: Exception) {
+                    if (e is CancellationException) throw e
+                    Log.d(TAG) { "Failed to dispatch ${event.kind} ${event.id} for ${savedAccount.npub}: ${e.message}" }
                }
            }
        }

-    private suspend fun consumeIfMatchesAccount(
-        pushWrappedEvent: GiftWrapEvent,
+    private suspend fun dispatchForAccount(
+        event: Event,
        account: Account,
    ) {
-        val notificationEvent = pushWrappedEvent.unwrapThrowing(account.signer)
-        consumeNotificationEvent(notificationEvent, account)
-    }
-
-    suspend fun consumeNotificationEvent(
-        notificationEvent: Event,
-        account: Account,
-    ) {
-        Log.d(TAG) { "New Notification ${notificationEvent.kind} ${notificationEvent.id} Arrived for ${account.signer.pubKey}" }
-        val consumed = LocalCache.hasConsumed(notificationEvent)
-        Log.d(TAG) { "New Notification ${notificationEvent.kind} ${notificationEvent.id} Arrived for ${account.signer.pubKey} consumed= $consumed" }
-        if (!consumed) {
-            Log.d(TAG, "New Notification was verified")
-            if (!notificationManager().areNotificationsEnabled()) return
-            Log.d(TAG, "Notifications are enabled")
-
-            unwrapAndConsume(notificationEvent, account.signer)?.let { innerNote ->
-                val innerEvent = innerNote.event
-                Log.d(TAG) { "Unwrapped consume ${innerEvent?.javaClass?.simpleName}" }
-
-                when (innerEvent) {
-                    is PrivateDmEvent -> {
-                        notify(innerEvent, account)
-                    }
-
-                    is LnZapEvent -> {
-                        notify(innerEvent, account)
-                    }
-
-                    is ChatMessageEvent -> {
-                        notify(innerEvent, account)
-                    }
-
-                    is ChatMessageEncryptedFileHeaderEvent -> {
-                        notify(innerEvent, account)
-                    }
-
-                    is ReactionEvent -> {
-                        notify(innerEvent, account)
-                    }
-
-                    is LiveChessGameAcceptEvent -> {
-                        notifyChessEvent(innerEvent, account, R.string.app_notification_chess_challenge_accepted)
-                    }
-
-                    is LiveChessMoveEvent -> {
-                        notifyChessEvent(innerEvent, account, R.string.app_notification_chess_your_turn)
-                    }
-
-                    is CallOfferEvent -> {
-                        notifyIncomingCall(innerEvent, account)
-                    }
-
-                    is WakeUpEvent -> {
-                        wakeUpFor(innerEvent, innerNote, account)
-                    }
-
-                    is WelcomeEvent -> {
-                        notify(innerEvent, account)
-                    }
-                }
+        // Calls and wake-ups are high-priority and always notify, even when MainActivity is visible.
+        when (event) {
+            is CallOfferEvent -> {
+                notifyIncomingCall(event, account)
+                return
            }
+
+            is WakeUpEvent -> {
+                wakeUpFor(event, LocalCache.getOrCreateNote(event.id), account)
+                return
+            }
+        }
+
+        // Everything else is suppressed while the user is actively on the home screen.
+        if (MainActivity.isResumed) return
+
+        when (event) {
+            is PrivateDmEvent -> notify(event, account)
+            is LnZapEvent -> notify(event, account)
+            is ChatMessageEvent -> notify(event, account)
+            is ChatMessageEncryptedFileHeaderEvent -> notify(event, account)
+            is ReactionEvent -> notify(event, account)
+            is LiveChessGameAcceptEvent -> notifyChessEvent(event, account, R.string.app_notification_chess_challenge_accepted)
+            is LiveChessMoveEvent -> notifyChessEvent(event, account, R.string.app_notification_chess_your_turn)
+            // WelcomeEvent is dispatched directly from processMarmotWelcomeFlow
+            // (no `p` tag, so tag-based matching doesn't work).
        }
    }

@@ -249,79 +224,6 @@ class EventNotificationConsumer(
        }
    }

-    suspend fun findAccountAndConsume(event: Event) =
-        withWakeLock {
-            Log.d(TAG, "New Notification Arrived")
-            val users = event.taggedUserIds().map { LocalCache.getOrCreateUser(it) }
-            val npubs = users.map { it.pubkeyNpub() }.toSet()
-
-            // PushNotification Wraps don't include a receiver.
-            // Test with all logged in accounts
-            var matchAccount = false
-            LocalPreferences.allSavedAccounts().forEach {
-                if (!matchAccount && (it.hasPrivKey || it.loggedInWithExternalSigner) && it.npub in npubs) {
-                    LocalPreferences.loadAccountConfigFromEncryptedStorage(it.npub)?.let { accountSettings ->
-                        Log.d(TAG) { "New Notification Testing if for ${it.npub}" }
-                        try {
-                            val account = Amethyst.instance.accountsCache.loadAccount(accountSettings)
-                            consumeNotificationEvent(event, account)
-                            matchAccount = true
-                        } catch (e: Exception) {
-                            if (e is CancellationException) throw e
-                            Log.d(TAG) { "Message was not for user ${it.npub}: ${e.message}" }
-                        }
-                    }
-                }
-            }
-        }
-
-    private suspend fun unwrapAndConsume(
-        event: Event,
-        signer: NostrSigner,
-    ): Note? {
-        if (LocalCache.hasConsumed(event)) return null
-
-        return when (event) {
-            is GiftWrapEvent -> {
-                if (LocalCache.justConsume(event, null, false)) {
-                    // new event
-                    val inner = event.unwrapThrowing(signer)
-                    // clear the encrypted payload to save memory
-                    LocalCache.getOrCreateNote(event.id).event = event.copyNoContent()
-
-                    unwrapAndConsume(inner, signer)
-                } else {
-                    null
-                }
-            }
-
-            is SealedRumorEvent -> {
-                if (LocalCache.justConsume(event, null, false)) {
-                    // new event
-                    val inner = event.unsealThrowing(signer)
-                    // clear the encrypted payload to save memory
-                    LocalCache.getOrCreateNote(event.id).event = event.copyNoContent()
-
-                    val note = LocalCache.getOrCreateNote(inner.id)
-                    // this is not verifiable
-                    if (LocalCache.justConsume(inner, null, true)) {
-                        note
-                    } else {
-                        null
-                    }
-                } else {
-                    null
-                }
-            }
-
-            else -> {
-                val note = LocalCache.getOrCreateNote(event.id)
-                LocalCache.justConsume(event, null, false)
-                note
-            }
-        }
-    }
-
    private suspend fun notify(
        event: ChatMessageEncryptedFileHeaderEvent,
        account: Account,
@@ -469,38 +371,28 @@ class EventNotificationConsumer(
        }
    }

-    private suspend fun notify(
+    /**
+     * Welcomes have no `p` tag, so [consumeFromCache]'s tag-based account match
+     * can't route them. They are instead dispatched here directly by
+     * [com.vitorpamplona.amethyst.ui.screen.loggedIn.processMarmotWelcomeFlow]
+     * after [MarmotManager.processWelcome] joins the group — which is also the
+     * only place we reliably know which account the invite was for.
+     */
+    suspend fun notifyWelcome(
        event: WelcomeEvent,
        account: Account,
-    ) {
+    ) = withWakeLock {
        Log.d(TAG, "New Marmot Welcome to Notify")

+        if (!notificationManager().areNotificationsEnabled()) return@withWakeLock
+        if (MainActivity.isResumed) return@withWakeLock
+
        // old event being re-broadcast
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
+        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return@withWakeLock
        // a welcome we ourselves emitted
-        if (event.pubKey == account.signer.pubKey) return
+        if (event.pubKey == account.signer.pubKey) return@withWakeLock

-        val nostrGroupId = event.nostrGroupId() ?: return
-        val manager = account.marmotManager ?: return
-
-        // Best-effort: process the welcome here so the chatroom is hydrated
-        // before composing the notification body. The push-notification
-        // background path does NOT go through Account.eventProcessor, so
-        // without this the invitee would only join the group later, when
-        // they next open the app and the relay subscription redelivers.
-        if (!manager.isMember(nostrGroupId)) {
-            try {
-                val result = manager.processWelcome(event, nostrGroupId)
-                if (result is WelcomeResult.Joined) {
-                    val chatroom = account.marmotGroupList.getOrCreateGroup(result.nostrGroupId)
-                    manager.syncMetadataTo(result.nostrGroupId, chatroom)
-                    account.marmotGroupList.notifyGroupChanged(result.nostrGroupId)
-                }
-            } catch (e: Exception) {
-                if (e is CancellationException) throw e
-                Log.w(TAG) { "Failed to process Marmot Welcome from notification path: ${e.message}" }
-            }
-        }
+        val nostrGroupId = event.nostrGroupId() ?: return@withWakeLock

        val chatroom = account.marmotGroupList.getOrCreateGroup(nostrGroupId)
        val groupName = chatroom.displayName.value?.takeIf { it.isNotBlank() } ?: "a private group"
@@ -0,0 +1,132 @@
+/*
+ * 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.service.notifications
+
+import android.content.Context
+import com.vitorpamplona.amethyst.model.Account
+import com.vitorpamplona.amethyst.model.LocalCache
+import com.vitorpamplona.quartz.experimental.notifications.wake.WakeUpEvent
+import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent
+import com.vitorpamplona.quartz.nip17Dm.files.ChatMessageEncryptedFileHeaderEvent
+import com.vitorpamplona.quartz.nip17Dm.messages.ChatMessageEvent
+import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
+import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent
+import com.vitorpamplona.quartz.nip64Chess.challenge.accept.LiveChessGameAcceptEvent
+import com.vitorpamplona.quartz.nip64Chess.move.LiveChessMoveEvent
+import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallOfferEvent
+import com.vitorpamplona.quartz.utils.Log
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+
+/**
+ * Observes [LocalCache] for notification-relevant events and hands them to
+ * [EventNotificationConsumer]. Events can reach [LocalCache] from any source —
+ * FCM push, UnifiedPush, Pokey, active relay subscriptions, or the
+ * [NotificationRelayService] — and the observer fires once per new insertion,
+ * giving us a single dedup'd source of truth for notification triggers.
+ *
+ * The dispatcher itself is always on (any of the delivery paths can fire it).
+ * Foreground suppression and per-event filtering are handled downstream in
+ * [EventNotificationConsumer].
+ */
+class NotificationDispatcher(
+    private val context: Context,
+    private val scope: CoroutineScope,
+) {
+    companion object {
+        private const val TAG = "NotificationDispatcher"
+
+        // The dispatcher observes the *final* notification payload kinds.
+        // GiftWrap/EphemeralGiftWrap (1059/21059) and SealedRumor (13) are NOT
+        // listed here — by the time we care, Account.newNotesPreProcessor has
+        // already unwrapped them and inserted the inner payload into LocalCache,
+        // which fires the observer a second time on the inner event.
+        //
+        // WelcomeEvent (kind:444) is also excluded: it has no `p` tag, so
+        // consumeFromCache can't route it. It's delivered directly via
+        // [notifyWelcome] from processMarmotWelcomeFlow, which does know the
+        // recipient account.
+        private val NOTIFICATION_KINDS =
+            listOf(
+                // Direct-arrival
+                PrivateDmEvent.KIND,
+                LnZapEvent.KIND,
+                ReactionEvent.KIND,
+                LiveChessGameAcceptEvent.KIND,
+                LiveChessMoveEvent.KIND,
+                WakeUpEvent.KIND,
+                // Unwrapped from GiftWrap → Seal
+                ChatMessageEvent.KIND,
+                ChatMessageEncryptedFileHeaderEvent.KIND,
+                // Unwrapped from EphemeralGiftWrap
+                CallOfferEvent.KIND,
+            )
+    }
+
+    private val consumer = EventNotificationConsumer(context)
+    private var job: Job? = null
+
+    fun start() {
+        if (job?.isActive == true) return
+        Log.d(TAG, "Starting notification dispatcher")
+        job =
+            scope.launch {
+                LocalCache
+                    .observeNewEvents<Event>(Filter(kinds = NOTIFICATION_KINDS))
+                    .collect { event ->
+                        try {
+                            consumer.consumeFromCache(event)
+                        } catch (e: Exception) {
+                            if (e is CancellationException) throw e
+                            Log.e(TAG, "Failed to dispatch notification for ${event.kind} ${event.id}", e)
+                        }
+                    }
+            }
+    }
+
+    fun stop() {
+        job?.cancel()
+        job = null
+    }
+
+    /**
+     * Direct-invocation entry point for [WelcomeEvent]. Bypasses the
+     * cache-observer path because Welcomes have no `p` tag for account
+     * routing. Called from processMarmotWelcomeFlow once MLS group join
+     * succeeds — at which point we know which account the invite was for.
+     */
+    suspend fun notifyWelcome(
+        event: WelcomeEvent,
+        account: Account,
+    ) {
+        try {
+            consumer.notifyWelcome(event, account)
+        } catch (e: Exception) {
+            if (e is CancellationException) throw e
+            Log.e(TAG, "Failed to dispatch Welcome notification ${event.id}", e)
+        }
+    }
+}
@@ -26,6 +26,7 @@ import android.content.Context.RECEIVER_EXPORTED
 import android.content.Intent
 import android.content.IntentFilter
 import android.os.Build
+import com.vitorpamplona.amethyst.model.LocalCache
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.CoroutineExceptionHandler
@@ -75,7 +76,8 @@ class PokeyReceiver : BroadcastReceiver() {

            scope.launch {
                try {
-                    EventNotificationConsumer(context.applicationContext).findAccountAndConsume(Event.fromJson(eventStr))
+                    // Feeds the shared cache; NotificationDispatcher observes and dispatches.
+                    LocalCache.justConsume(Event.fromJson(eventStr), null, false)
                } catch (e: Exception) {
                    Log.e(TAG, "Failed to parse Pokey Event", e)
                }
@@ -0,0 +1,62 @@
+/*
+ * 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.service.notifications
+
+import com.vitorpamplona.amethyst.Amethyst
+import com.vitorpamplona.amethyst.LocalPreferences
+import com.vitorpamplona.amethyst.model.LocalCache
+import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
+import com.vitorpamplona.quartz.utils.Log
+import kotlinx.coroutines.CancellationException
+
+/**
+ * Handles GiftWraps that arrive via FCM / UnifiedPush. The push server
+ * re-wraps the real NIP-59 GiftWrap inside its own outer GiftWrap and strips
+ * the recipient `p` tag, so neither the relay's [CacheClientConnector] nor
+ * the account's [GiftWrapEventHandler] can identify which account owns it.
+ * We have to probe every saved account's signer until one decrypts.
+ *
+ * Once decrypted, the inner event is fed into [LocalCache]. From there the
+ * normal `Account.newNotesPreProcessor` → `GiftWrapEventHandler` →
+ * `SealedRumorEventHandler` chain unwraps any remaining layers, and
+ * [NotificationDispatcher] picks up the final payload and notifies.
+ */
+object PushWrapDecryptor {
+    private const val TAG = "PushWrapDecryptor"
+
+    suspend fun unwrapAndFeed(outerWrap: GiftWrapEvent) {
+        LocalPreferences.allSavedAccounts().forEach { savedAccount ->
+            if (!savedAccount.hasPrivKey && !savedAccount.loggedInWithExternalSigner) return@forEach
+
+            val accountSettings = LocalPreferences.loadAccountConfigFromEncryptedStorage(savedAccount.npub) ?: return@forEach
+            try {
+                val account = Amethyst.instance.accountsCache.loadAccount(accountSettings)
+                val inner = outerWrap.unwrapThrowing(account.signer)
+                LocalCache.justConsume(inner, null, false)
+                return
+            } catch (e: Exception) {
+                if (e is CancellationException) throw e
+                Log.d(TAG) { "Push wrap not for ${savedAccount.npub}: ${e.message}" }
+            }
+        }
+        Log.w(TAG) { "Push wrap ${outerWrap.id} did not decrypt for any saved account" }
+    }
+}
@@ -56,6 +56,15 @@ import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch

 class MainActivity : AppCompatActivity() {
+    companion object {
+        // True only while MainActivity is resumed. Used by the notification
+        // pipeline to suppress in-app notifications — PiP/Call activities
+        // have their own lifecycle, so MainActivity is paused while they're up.
+        @Volatile
+        var isResumed: Boolean = false
+            private set
+    }
+
    @RequiresApi(Build.VERSION_CODES.R)
    override fun onCreate(savedInstanceState: Bundle?) {
        enableEdgeToEdge()
@@ -74,6 +83,7 @@ class MainActivity : AppCompatActivity() {
    @OptIn(DelicateCoroutinesApi::class)
    override fun onResume() {
        super.onResume()
+        isResumed = true

        Log.d("ActivityLifecycle") { "MainActivity.onResume $this" }

@@ -82,6 +92,7 @@ class MainActivity : AppCompatActivity() {
    }

    override fun onPause() {
+        isResumed = false
        Log.d("ActivityLifecycle") { "MainActivity.onPause $this" }

        @OptIn(DelicateCoroutinesApi::class)
@@ -20,6 +20,7 @@
 */
 package com.vitorpamplona.amethyst.ui.screen.loggedIn

+import com.vitorpamplona.amethyst.Amethyst
 import com.vitorpamplona.amethyst.commons.call.CallManager
 import com.vitorpamplona.amethyst.commons.model.privateChats.ChatroomList
 import com.vitorpamplona.amethyst.model.Account
@@ -311,28 +312,21 @@ class GiftWrapEventHandler(

        eventNote.event = event.copyNoContent()

-        // Check if the unwrapped event is a Marmot WelcomeEvent (kind:444)
-        if (MarmotInboundProcessor.isWelcomeEvent(innerGift)) {
-            Log.d("MarmotDbg") {
-                "GiftWrapEventHandler: detected Marmot WelcomeEvent — routing to processMarmotWelcome"
-            }
-            processMarmotWelcome(innerGift, eventNote, publicNote)
-            return
-        }
-
        if (cache.justConsume(innerGift, null, false)) {
            cache.copyRelaysFromTo(publicNote, innerGift)
            val innerGiftNote = cache.getOrCreateNote(innerGift.id)
-            eventProcessor.consumeEvent(innerGift, innerGiftNote, publicNote)
-        }
-    }

-    private suspend fun processMarmotWelcome(
-        innerEvent: Event,
-        eventNote: Note,
-        publicNote: Note,
-    ) {
-        processMarmotWelcomeFlow(innerEvent, account)
+            // Marmot Welcomes need MLS processing in addition to being cached,
+            // but they do not route through the normal eventProcessor consumer.
+            if (MarmotInboundProcessor.isWelcomeEvent(innerGift)) {
+                Log.d("MarmotDbg") {
+                    "GiftWrapEventHandler: detected Marmot WelcomeEvent — routing to processMarmotWelcomeFlow"
+                }
+                processMarmotWelcomeFlow(innerGift, account)
+            } else {
+                eventProcessor.consumeEvent(innerGift, innerGiftNote, publicNote)
+            }
+        }
    }

    private suspend fun processExistingGiftWrap(
@@ -404,6 +398,11 @@ private suspend fun processMarmotWelcomeFlow(
            if (result.needsKeyPackageRotation) {
                account.publishMarmotKeyPackages()
            }
+
+            // Fire the "You've been added to <group>" notification. Welcomes
+            // have no `p` tag, so the cache-observer path can't route them;
+            // this is the single point where we know the recipient account.
+            Amethyst.instance.notificationDispatcher.notifyWelcome(innerEvent, account)
        }

        is WelcomeResult.AlreadyJoined -> {
@@ -472,23 +471,23 @@ class SealedRumorEventHandler(

        eventNote.event = event.copyNoContent()

+        cache.justConsume(innerRumor, null, true)
+        cache.copyRelaysFromTo(publicNote, innerRumor)
+
+        val innerRumorNote = cache.getOrCreateNote(innerRumor.id)
+
        // Marmot Welcome: GiftWrap → Seal → WelcomeEvent. The Seal handler
        // is the actual point at which we see the kind:444 inner. Route it
-        // straight to the shared flow — there's no normal LocalCache event
-        // handler for kind:444, so otherwise it would be silently dropped.
+        // to the MLS flow for group joining in addition to caching — there's
+        // no normal eventProcessor consumer for kind:444.
        if (MarmotInboundProcessor.isWelcomeEvent(innerRumor)) {
            Log.d("MarmotDbg") {
                "SealedRumorEventHandler: detected Marmot WelcomeEvent inside seal — routing to processMarmotWelcomeFlow"
            }
            processMarmotWelcomeFlow(innerRumor, account)
-            return
+        } else {
+            eventProcessor.consumeEvent(innerRumor, innerRumorNote, publicNote)
        }
-
-        cache.justConsume(innerRumor, null, true)
-        cache.copyRelaysFromTo(publicNote, innerRumor)
-
-        val innerRumorNote = cache.getOrCreateNote(innerRumor.id)
-        eventProcessor.consumeEvent(innerRumor, innerRumorNote, publicNote)
    }

    private suspend fun processExistingSealedRumor(
@@ -67,7 +67,10 @@ class PushNotificationReceiverService : FirebaseMessagingService() {
    private suspend fun receiveIfNew(event: GiftWrapEvent) {
        if (eventCache.get(event.id) == null) {
            eventCache.put(event.id, event.id)
-            EventNotificationConsumer(applicationContext).consume(event)
+            // The push server re-wraps the real GiftWrap and strips the p tag;
+            // unwrap the outer layer, feed the inner into LocalCache, and let
+            // the usual Account → EventProcessor chain take over.
+            PushWrapDecryptor.unwrapAndFeed(event)
        }
    }

@@ -0,0 +1,50 @@
+/*
+ * 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.observables
+
+import com.vitorpamplona.amethyst.commons.model.Note
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+
+/**
+ * Emits each new event that matches the given Nostr filter, one at a time.
+ *
+ * Unlike [EventListMatchingFilter] / [NoteListMatchingFilter], this does not
+ * accumulate a list — it simply calls [onNew] per matching event as it is
+ * inserted into the cache. Useful for reactive event-triggered pipelines
+ * (e.g. notifications) that need per-event delivery without list overhead.
+ */
+class NewEventMatchingFilter<T : Event>(
+    private val filter: Filter,
+    private val onNew: (T) -> Unit,
+) : Observable {
+    @Suppress("UNCHECKED_CAST")
+    override fun new(
+        event: Event,
+        note: Note,
+    ) {
+        if (filter.match(event)) {
+            onNew(event as T)
+        }
+    }
+
+    override fun remove(note: Note) {}
+}