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..284846799 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 @@ -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) } } 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..2254da21f 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 @@ -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 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() } @@ -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 } 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..e40adf983 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,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" 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..f4ff5f82d --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt @@ -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(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) + } + } +} 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/service/notifications/PushWrapDecryptor.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PushWrapDecryptor.kt new file mode 100644 index 000000000..bd6b3735c --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/PushWrapDecryptor.kt @@ -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" } + } +} 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/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt index b1ab71975..6cbd30046 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt @@ -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 " 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( 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..599860025 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 @@ -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) } } 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) {} +}