From fa041078653f848a416958ec233ed5a1abc0c6a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Apr 2026 21:05:56 +0000 Subject: [PATCH] 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(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. --- .../notifications/PushMessageReceiver.kt | 6 +- .../com/vitorpamplona/amethyst/AppModules.kt | 10 + .../amethyst/model/LocalCache.kt | 20 ++ .../EventNotificationConsumer.kt | 234 +++++++----------- .../notifications/NotificationDispatcher.kt | 101 ++++++++ .../service/notifications/PokeyReceiver.kt | 4 +- .../vitorpamplona/amethyst/ui/MainActivity.kt | 11 + .../PushNotificationReceiverService.kt | 6 +- .../observables/NewEventMatchingFilter.kt | 50 ++++ 9 files changed, 299 insertions(+), 143 deletions(-) create mode 100644 amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NewEventMatchingFilter.kt diff --git a/amethyst/src/fdroid/java/com/vitorpamplona/amethyst/service/notifications/PushMessageReceiver.kt b/amethyst/src/fdroid/java/com/vitorpamplona/amethyst/service/notifications/PushMessageReceiver.kt index 7f7e4dfd5..cebacc25d 100644 --- a/amethyst/src/fdroid/java/com/vitorpamplona/amethyst/service/notifications/PushMessageReceiver.kt +++ b/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) } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt index db0e187a8..06f0066bc 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt +++ b/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() diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt index a6af8229c..1426f3f68 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt +++ b/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 observeNewEvents(filter: Filter): Flow = + callbackFlow { + val newFilter = + NewEventMatchingFilter(filter) { + trySend(it) + } + + observables.put(newFilter, newFilter) + + awaitClose { + observables.remove(newFilter) + } + } + @Suppress("UNCHECKED_CAST") fun observeLatestEvent(filter: Filter) = observeEvents(filter).map { it.firstOrNull() } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/EventNotificationConsumer.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/EventNotificationConsumer.kt index 0268dedd0..3a45e4616 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/EventNotificationConsumer.kt +++ b/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. - // 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 + + LocalPreferences.allSavedAccounts().forEach { savedAccount -> + if (!savedAccount.hasPrivKey && !savedAccount.loggedInWithExternalSigner) return@forEach + + // For unwrapped events with `p` tags, only try accounts that are tagged. + 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(accountSettings) + if (consumeForAccount(event, account)) return@withWakeLock + } catch (e: Exception) { + if (e is CancellationException) throw e + 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, - ) { - val notificationEvent = pushWrappedEvent.unwrapThrowing(account.signer) - consumeNotificationEvent(notificationEvent, account) + ): Boolean { + val inner = unwrapFully(event, account.signer) ?: return false + dispatchForAccount(inner, account) + return true } - suspend fun consumeNotificationEvent( - notificationEvent: Event, + private suspend fun dispatchForAccount( + event: 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) + is WelcomeEvent -> notify(event, account) } } @@ -249,78 +240,45 @@ 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( + /** + * Fully unwraps gift wraps and seals so the inner notification-relevant event + * can be dispatched. Updates [LocalCache] so the inner events are available to + * the UI and clears the outer encrypted payloads to save memory. Returns null + * if the signer cannot decrypt the wrap (i.e. this event isn't for this account). + * + * Idempotent: safe to call whether or not the outer event is already in cache, + * since [NotificationDispatcher] invokes this after [LocalCache] has already + * stored the wrapper. + */ + private suspend fun unwrapFully( event: Event, signer: NostrSigner, - ): Note? { - if (LocalCache.hasConsumed(event)) return null - - return when (event) { - is GiftWrapEvent -> { - if (LocalCache.justConsume(event, null, false)) { - // new event + ): Event? = + try { + when (event) { + // EphemeralGiftWrapEvent inherits GiftWrapEvent; both handled here. + is GiftWrapEvent -> { val inner = event.unwrapThrowing(signer) - // clear the encrypted payload to save memory LocalCache.getOrCreateNote(event.id).event = event.copyNoContent() - - unwrapAndConsume(inner, signer) - } else { - null + unwrapFully(inner, signer) } - } - is SealedRumorEvent -> { - if (LocalCache.justConsume(event, null, false)) { - // new event + is SealedRumorEvent -> { 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 + LocalCache.justConsume(inner, null, true) + inner + } - val note = LocalCache.getOrCreateNote(inner.id) - // this is not verifiable - if (LocalCache.justConsume(inner, null, true)) { - note - } else { - null - } - } else { - null + else -> { + event } } - - else -> { - val note = LocalCache.getOrCreateNote(event.id) - LocalCache.justConsume(event, null, false) - note - } + } catch (e: Exception) { + if (e is CancellationException) throw e + null } - } private suspend fun notify( event: ChatMessageEncryptedFileHeaderEvent, diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt new file mode 100644 index 000000000..2664fc360 --- /dev/null +++ b/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(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 + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PokeyReceiver.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PokeyReceiver.kt index d2f23c8e1..5f0980204 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PokeyReceiver.kt +++ b/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) } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/MainActivity.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/MainActivity.kt index 6871e1aca..4b1cd5789 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/MainActivity.kt +++ b/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) diff --git a/amethyst/src/play/java/com/vitorpamplona/amethyst/service/notifications/PushNotificationReceiverService.kt b/amethyst/src/play/java/com/vitorpamplona/amethyst/service/notifications/PushNotificationReceiverService.kt index 197dbd4d2..59cdf05cf 100644 --- a/amethyst/src/play/java/com/vitorpamplona/amethyst/service/notifications/PushNotificationReceiverService.kt +++ b/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) } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NewEventMatchingFilter.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NewEventMatchingFilter.kt new file mode 100644 index 000000000..48c17196b --- /dev/null +++ b/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( + 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) {} +}