feat(notifications): dispatch from LocalCache for all delivery sources

The NotificationRelayService consumes events directly into LocalCache, which
made EventNotificationConsumer's hasConsumed dedup check short-circuit every
subsequent push-notification path. As a result, no notifications fired while
the service was active.

Rework the notification pipeline so every delivery source (FCM, UnifiedPush,
Pokey, active relay subscriptions, NotificationRelayService) feeds LocalCache
and a single dispatcher observes it for notification-relevant kinds. Foreground
suppression is scoped to MainActivity (PiP/Call activities keep notifying),
with carve-outs for CallOffer and WakeUp so time-sensitive events always fire.

- Add NewEventMatchingFilter observable primitive alongside
  EventListMatchingFilter for per-event emission (no list accumulation).
- Add LocalCache.observeNewEvents<T>(filter) backed by the new observable.
- Add MainActivity.isResumed flag flipped in onResume/onPause.
- Add NotificationDispatcher that observes LocalCache for GiftWrap,
  EphemeralGiftWrap, PrivateDm, LnZap, Reaction, Chess, and WakeUp.
- Replace EventNotificationConsumer.consume / findAccountAndConsume with
  consumeFromCache, which skips the hasConsumed check (the observer itself
  provides first-delivery semantics) and gates non-priority kinds on
  MainActivity.isResumed.
- Rewire FCM/UnifiedPush/Pokey receivers to feed LocalCache directly.
- Wire the dispatcher into AppModules lifecycle.

amethyst/src/fdroid/java/com/vitorpamplona/amethyst/service/notifications/PushMessageReceiver.kt

@@ -26,6 +26,7 @@ import android.util.LruCache
 import androidx.core.content.ContextCompat
 import com.vitorpamplona.amethyst.Amethyst
 import com.vitorpamplona.amethyst.LocalPreferences
+import com.vitorpamplona.amethyst.model.LocalCache
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
 import com.vitorpamplona.quartz.utils.Log
@@ -73,10 +74,11 @@ class PushMessageReceiver : MessagingReceiver() {
         return null
     }
 
-    private suspend fun receiveIfNew(event: GiftWrapEvent) {
+    private fun receiveIfNew(event: GiftWrapEvent) {
         if (eventCache.get(event.id) == null) {
             eventCache.put(event.id, event.id)
-            EventNotificationConsumer(appContext).consume(event)
+            // Feeds the shared cache; NotificationDispatcher observes and dispatches.
+            LocalCache.justConsume(event, null, false)
         }
     }
 

amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt

@@ -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()

amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt

@@ -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
@@ -380,6 +381,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() }
 

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/EventNotificationConsumer.kt

@@ -45,6 +45,7 @@ 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
@@ -115,96 +116,86 @@ class EventNotificationConsumer(
         }
     }
 
-    suspend fun consume(event: GiftWrapEvent) =
+    /**
+     * Entry point for events arriving into [LocalCache] from any source (FCM push,
+     * UnifiedPush, Pokey, active relay subscriptions, NotificationRelayService).
+     * The caller is the [NotificationDispatcher] which observes [LocalCache] for
+     * notification-relevant kinds and guarantees per-event first-delivery semantics
+     * (the [LocalCache] observer only fires on new insertions), so no hasConsumed
+     * check is needed here.
+     *
+     * Iterates the logged-in accounts, decrypting wraps with each signer until one
+     * matches (wraps don't carry a recipient hint).
+     */
+    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.
+            if (!notificationManager().areNotificationsEnabled()) return@withWakeLock
-            // Test with all logged in accounts
+
-            var matchAccount = false
+            LocalPreferences.allSavedAccounts().forEach { savedAccount ->
-            LocalPreferences.allSavedAccounts().forEach {
+                if (!savedAccount.hasPrivKey && !savedAccount.loggedInWithExternalSigner) return@forEach
-                if (!matchAccount && (it.hasPrivKey || it.loggedInWithExternalSigner)) {
+
-                    LocalPreferences.loadAccountConfigFromEncryptedStorage(it.npub)?.let { acc ->
+                // For unwrapped events with `p` tags, only try accounts that are tagged.
-                        Log.d(TAG) { "New Notification Testing if for ${it.npub}" }
+                if (event !is GiftWrapEvent) {
+                    val taggedNpubs = event.taggedUserIds().mapTo(mutableSetOf()) { LocalCache.getOrCreateUser(it).pubkeyNpub() }
+                    if (taggedNpubs.isNotEmpty() && savedAccount.npub !in taggedNpubs) return@forEach
+                }
+
+                val accountSettings = LocalPreferences.loadAccountConfigFromEncryptedStorage(savedAccount.npub) ?: return@forEach
                 try {
-                            val account = Amethyst.instance.accountsCache.loadAccount(acc)
+                    val account = Amethyst.instance.accountsCache.loadAccount(accountSettings)
-                            consumeIfMatchesAccount(event, account)
+                    if (consumeForAccount(event, account)) return@withWakeLock
-                            matchAccount = true
                 } catch (e: Exception) {
                     if (e is CancellationException) throw e
-                            Log.d(TAG) { "Message was not for user ${it.npub}: ${e.message}" }
+                    Log.d(TAG) { "Message was not for user ${savedAccount.npub}: ${e.message}" }
-                        }
-                    }
                 }
             }
         }
 
-    private suspend fun consumeIfMatchesAccount(
+    /**
-        pushWrappedEvent: GiftWrapEvent,
+     * Tries to process [event] for [account]. Unwraps gift wraps and seals as needed,
+     * updating [LocalCache] so the UI can surface the content later. Returns true if
+     * the event matched this account (i.e. decryption succeeded or tagging matched).
+     */
+    private suspend fun consumeForAccount(
+        event: Event,
+        account: Account,
+    ): Boolean {
+        val inner = unwrapFully(event, account.signer) ?: return false
+        dispatchForAccount(inner, account)
+        return true
+    }
+
+    private suspend fun dispatchForAccount(
+        event: Event,
         account: Account,
     ) {
-        val notificationEvent = pushWrappedEvent.unwrapThrowing(account.signer)
+        // Calls and wake-ups are high-priority and always notify, even when MainActivity is visible.
-        consumeNotificationEvent(notificationEvent, account)
+        when (event) {
-    }
-
-    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)
+                notifyIncomingCall(event, account)
+                return
             }
 
             is WakeUpEvent -> {
-                        wakeUpFor(innerEvent, innerNote, account)
+                wakeUpFor(event, LocalCache.getOrCreateNote(event.id), account)
+                return
+            }
         }
 
-                    is WelcomeEvent -> {
+        // Everything else is suppressed while the user is actively on the home screen.
-                        notify(innerEvent, account)
+        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)
+            is WelcomeEvent -> notify(event, account)
         }
     }
 
@@ -249,77 +240,44 @@ class EventNotificationConsumer(
         }
     }
 
-    suspend fun findAccountAndConsume(event: Event) =
+    /**
-        withWakeLock {
+     * Fully unwraps gift wraps and seals so the inner notification-relevant event
-            Log.d(TAG, "New Notification Arrived")
+     * can be dispatched. Updates [LocalCache] so the inner events are available to
-            val users = event.taggedUserIds().map { LocalCache.getOrCreateUser(it) }
+     * the UI and clears the outer encrypted payloads to save memory. Returns null
-            val npubs = users.map { it.pubkeyNpub() }.toSet()
+     * if the signer cannot decrypt the wrap (i.e. this event isn't for this account).
-
+     *
-            // PushNotification Wraps don't include a receiver.
+     * Idempotent: safe to call whether or not the outer event is already in cache,
-            // Test with all logged in accounts
+     * since [NotificationDispatcher] invokes this after [LocalCache] has already
-            var matchAccount = false
+     * stored the wrapper.
-            LocalPreferences.allSavedAccounts().forEach {
+     */
-                if (!matchAccount && (it.hasPrivKey || it.loggedInWithExternalSigner) && it.npub in npubs) {
+    private suspend fun unwrapFully(
-                    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? {
+    ): Event? =
-        if (LocalCache.hasConsumed(event)) return null
+        try {
-
+            when (event) {
-        return when (event) {
+                // EphemeralGiftWrapEvent inherits GiftWrapEvent; both handled here.
                 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()
-
+                    unwrapFully(inner, signer)
-                    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()
-
+                    // seal payload is not verifiable via signature
-                    val note = LocalCache.getOrCreateNote(inner.id)
+                    LocalCache.justConsume(inner, null, true)
-                    // this is not verifiable
+                    inner
-                    if (LocalCache.justConsume(inner, null, true)) {
-                        note
-                    } else {
-                        null
-                    }
-                } else {
-                    null
-                }
                 }
 
                 else -> {
-                val note = LocalCache.getOrCreateNote(event.id)
+                    event
-                LocalCache.justConsume(event, null, false)
-                note
                 }
             }
+        } catch (e: Exception) {
+            if (e is CancellationException) throw e
+            null
         }
 
     private suspend fun notify(

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt

@@ -0,0 +1,101 @@
+/*
+ * 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.LocalCache
+import com.vitorpamplona.quartz.experimental.notifications.wake.WakeUpEvent
+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.nip25Reactions.ReactionEvent
+import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent
+import com.vitorpamplona.quartz.nip59Giftwrap.wraps.EphemeralGiftWrapEvent
+import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
+import com.vitorpamplona.quartz.nip64Chess.challenge.accept.LiveChessGameAcceptEvent
+import com.vitorpamplona.quartz.nip64Chess.move.LiveChessMoveEvent
+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"
+
+        // Kinds that can trigger notifications. GiftWrap and EphemeralGiftWrap carry
+        // most encrypted payloads (DMs, Chess, CallOffer, Welcome); the rest arrive
+        // unwrapped. SealedRumor (kind 13) is intentionally omitted — it only exists
+        // inside a GiftWrap, and the wrapper drives the unwrap.
+        private val NOTIFICATION_KINDS =
+            listOf(
+                GiftWrapEvent.KIND,
+                EphemeralGiftWrapEvent.KIND,
+                PrivateDmEvent.KIND,
+                LnZapEvent.KIND,
+                ReactionEvent.KIND,
+                LiveChessGameAcceptEvent.KIND,
+                LiveChessMoveEvent.KIND,
+                WakeUpEvent.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
+    }
+}

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PokeyReceiver.kt

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

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/MainActivity.kt

@@ -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)

amethyst/src/play/java/com/vitorpamplona/amethyst/service/notifications/PushNotificationReceiverService.kt

@@ -27,6 +27,7 @@ import com.google.firebase.messaging.FirebaseMessagingService
 import com.google.firebase.messaging.RemoteMessage
 import com.vitorpamplona.amethyst.Amethyst
 import com.vitorpamplona.amethyst.LocalPreferences
+import com.vitorpamplona.amethyst.model.LocalCache
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
 import com.vitorpamplona.quartz.utils.Log
@@ -64,10 +65,11 @@ class PushNotificationReceiverService : FirebaseMessagingService() {
         return null
     }
 
-    private suspend fun receiveIfNew(event: GiftWrapEvent) {
+    private fun receiveIfNew(event: GiftWrapEvent) {
         if (eventCache.get(event.id) == null) {
             eventCache.put(event.id, event.id)
-            EventNotificationConsumer(applicationContext).consume(event)
+            // Feeds the shared cache; NotificationDispatcher observes and dispatches.
+            LocalCache.justConsume(event, null, false)
         }
     }
 

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NewEventMatchingFilter.kt

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