refactor(notifications): centralize age + self gates, predicate observer

Audit of every notify() path found self-exclusion + 15-min age checks
duplicated across nearly all of them, with two inconsistencies (DM kind 4
and zap kind 9735 were missing explicit self-checks). Reactions, zaps,
and chess also re-ran isTaggedUser even though consumeFromCache already
routes by the same `p` tag.

Collapse the duplication into two layers:

- Observer layer (NotificationDispatcher): 15-min rolling age cutoff is
  now enforced by a predicate on LocalCache.observeNewEvents, before any
  account routing happens. The Nostr Filter grammar's `since` is a fixed
  value from dispatcher start; the predicate re-evaluates TimeUtils.
  fifteenMinutesAgo() per event so the window rolls forward as wall-clock
  time advances.

- Per-account layer (dispatchForAccount): right after the call/wake-up
  branch and the MainActivity.isResumed gate, drop events authored by
  the current account. Kept per-account (not observer-wide) because in a
  multi-account session account A's outgoing event legitimately becomes
  account B's incoming notification on the same device, so a device-wide
  author-exclusion set would swallow A→B.

Mechanically, NewEventMatchingFilter now takes an optional predicate so
observers can reject on fields the Filter grammar can't express (like a
moving `createdAt` cutoff). LocalCache.observeNewEvents adds a matching
overload that forwards it.

With the gates centralized, every notify() method drops its own age and
self-author checks (and for reaction/zap, the redundant isTaggedUser).
notifyWelcome keeps its own guards because it's dispatched directly
from processMarmotWelcomeFlow, bypassing the observer and dispatchForAccount.
Calls and wake-ups also bypass the shared gates by their short-circuit
return before the shared block.

https://claude.ai/code/session_01GQDJxiHPogdzCNhUBN7Pjc

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

@@ -386,11 +386,18 @@ object LocalCache : ILocalCache, ICacheProvider {
      * 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.
+     *
+     * The optional [predicate] runs after the Nostr [Filter] matches and lets
+     * callers reject events on fields the Filter grammar can't express (e.g.
+     * a rolling `createdAt` window that drifts as wall-clock time advances).
      */
-    fun <T : Event> observeNewEvents(filter: Filter): Flow<T> =
+    fun <T : Event> observeNewEvents(
+        filter: Filter,
+        predicate: (Event) -> Boolean = { true },
+    ): Flow<T> =
         callbackFlow {
             val newFilter =
-                NewEventMatchingFilter<T>(filter) {
+                NewEventMatchingFilter<T>(filter, predicate) {
                     trySend(it)
                 }
 

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

@@ -172,6 +172,9 @@ class EventNotificationConsumer(
         account: Account,
     ) {
         // Calls and wake-ups are high-priority and always notify, even when MainActivity is visible.
+        // They have their own freshness rules (CallManager.MAX_EVENT_AGE_SECONDS = 20s) and
+        // author-identity semantics (caller pubkey is the other party), so they bypass the
+        // shared gates below.
         when (event) {
             is CallOfferEvent -> {
                 notifyIncomingCall(event, account)
@@ -187,6 +190,14 @@ class EventNotificationConsumer(
         // Everything else is suppressed while the user is actively on the home screen.
         if (MainActivity.isResumed) return
 
+        // Shared per-account gate: don't push-notify events this account authored.
+        // Applied here (not at the observer) because in a multi-account session
+        // account A's outgoing event legitimately becomes account B's incoming
+        // notification on the same device. The observer already enforces the
+        // 15-min rolling age window, so individual notify() methods don't need
+        // to repeat either check.
+        if (event.pubKey == account.signer.pubKey) return
+
         when (event) {
             is PrivateDmEvent -> notify(event, account)
 
@@ -270,13 +281,7 @@ class EventNotificationConsumer(
         account: Account,
     ) {
         Log.d(TAG, "New ChatMessage File to Notify")
-        if (
+        // Age + self-author gates run centrally in dispatchForAccount.
-            // old event being re-broadcasted
-            event.createdAt > TimeUtils.fifteenMinutesAgo() &&
-            // don't display if it comes from me.
-            event.pubKey != account.signer.pubKey
-        ) { // from the user
-            Log.d(TAG, "Notifying")
         val chatroomList = LocalCache.getOrCreateChatroomList(account.signer.pubKey)
         val chatNote = LocalCache.getNoteIfExists(event.id) ?: return
         val chatRoom = event.chatroomKey(account.signer.pubKey)
@@ -284,11 +289,11 @@ class EventNotificationConsumer(
         val followingKeySet = account.followingKeySet()
 
         val isKnownRoom =
-                (
+            chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true ||
-                    chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true || chatroomList.hasSentMessagesTo(chatRoom)
+                chatroomList.hasSentMessagesTo(chatRoom)
-                )
+
+        if (!isKnownRoom) return
 
-            if (isKnownRoom) {
         val content = chatNote.event?.content ?: ""
         val user = chatNote.author?.toBestDisplayName() ?: ""
         val userPicture = chatNote.author?.profilePicture()
@@ -313,30 +318,25 @@ class EventNotificationConsumer(
                 chatroomMembers = chatroomMembers,
             )
     }
-        }
-    }
 
     private suspend fun notify(
         event: ChatMessageEvent,
         account: Account,
     ) {
         Log.d(TAG, "New ChatMessage to Notify")
-        if (
+        // Age + self-author gates run centrally in dispatchForAccount.
-            // old event being re-broadcasted
-            event.createdAt > TimeUtils.fifteenMinutesAgo() &&
-            // don't display if it comes from me.
-            event.pubKey != account.signer.pubKey
-        ) { // from the user
-            Log.d(TAG, "Notifying")
         val chatroomList = LocalCache.getOrCreateChatroomList(account.signer.pubKey)
         val chatNote = LocalCache.getNoteIfExists(event.id) ?: return
         val chatRoom = event.chatroomKey(account.signer.pubKey)
 
         val followingKeySet = account.followingKeySet()
 
-            val isKnownRoom = chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true || chatroomList.hasSentMessagesTo(chatRoom)
+        val isKnownRoom =
+            chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true ||
+                chatroomList.hasSentMessagesTo(chatRoom)
+
+        if (!isKnownRoom) return
 
-            if (isKnownRoom) {
         val content = chatNote.event?.content ?: ""
         val user = chatNote.author?.toBestDisplayName() ?: ""
         val userPicture = chatNote.author?.profilePicture()
@@ -361,18 +361,18 @@ class EventNotificationConsumer(
                 chatroomMembers = chatroomMembers,
             )
     }
-        }
-    }
 
     private suspend fun notify(
         event: PrivateDmEvent,
         account: Account,
     ) {
         Log.d(TAG, "New Nip-04 DM to Notify")
-        // old event being re-broadcast
+        // Age + self-author gates run centrally in dispatchForAccount. The
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
+        // dispatchForAccount self-check (event.pubKey != account.signer.pubKey)
+        // also covers the "don't notify myself about DMs I sent" case that
+        // was previously implicit via the recipient match below.
+        if (account.signer.pubKey != event.verifiedRecipientPubKey()) return
 
-        if (account.signer.pubKey == event.verifiedRecipientPubKey()) {
         val note = LocalCache.getNoteIfExists(event.id) ?: return
         val chatroomList = LocalCache.getOrCreateChatroomList(account.signer.pubKey)
 
@@ -380,13 +380,16 @@ class EventNotificationConsumer(
 
         val chatRoom = event.chatroomKey(account.signer.pubKey)
 
-            val isKnownRoom = chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true || chatroomList.hasSentMessagesTo(chatRoom)
+        val isKnownRoom =
+            chatroomList.rooms.get(chatRoom)?.senderIntersects(followingKeySet) == true ||
+                chatroomList.hasSentMessagesTo(chatRoom)
 
-            if (isKnownRoom) {
+        if (!isKnownRoom) return
-                note.author?.let {
+
-                    decryptContent(note, account.signer)?.let { content ->
+        val author = note.author ?: return
-                        val user = note.author?.toBestDisplayName() ?: ""
+        val content = decryptContent(note, account.signer) ?: return
-                        val userPicture = note.author?.profilePicture()
+        val user = author.toBestDisplayName()
+        val userPicture = author.profilePicture()
         val accountNpub =
             account.signer.pubKey
                 .hexToByteArray()
@@ -407,10 +410,6 @@ class EventNotificationConsumer(
                 chatroomMembers = null,
             )
     }
-                }
-            }
-        }
-    }
 
     /**
      * Welcomes have no `p` tag, so [consumeFromCache]'s tag-based account match
@@ -501,13 +500,10 @@ class EventNotificationConsumer(
         Log.d(TAG) { "Notify Start ${event.toNostrUri()}" }
         LocalCache.getNoteIfExists(event.id) ?: return
 
-        Log.d(TAG, "Notify Not Notified Yet")
+        // Age + self-author gates run centrally in dispatchForAccount. For zaps
-
+        // the self-check is effectively a no-op (receipts are signed by the LN
-        // old event being re-broadcast
+        // service, not the zapper) but the uniform rule is cheap and keeps the
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
+        // downstream invariants simple.
-
-        Log.d(TAG, "Notify Not an old event")
-
         val noteZapRequest = event.zapRequest?.id?.let { LocalCache.checkGetOrCreateNote(it) } ?: return
         val noteZapped = event.zappedPost().firstOrNull()?.let { LocalCache.checkGetOrCreateNote(it) } ?: return
 
@@ -622,14 +618,9 @@ class EventNotificationConsumer(
     ) {
         Log.d(TAG, "New Reaction to Notify")
 
-        // old event being re-broadcast
+        // Age + self-author gates run centrally in dispatchForAccount.
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
+        // p-tag match already enforced by consumeFromCache; no redundant
-
+        // isTaggedUser re-check needed.
-        // don't notify for own reactions
-        if (event.pubKey == account.signer.pubKey) return
-
-        // only notify if the reaction is for the current user
-        if (!event.isTaggedUser(account.signer.pubKey)) return
 
         val reactedPostId = event.originalPost().firstOrNull() ?: return
         val reactedNote = LocalCache.checkGetOrCreateNote(reactedPostId)
@@ -694,12 +685,7 @@ class EventNotificationConsumer(
         account: Account,
     ) {
         Log.d(TAG, "New TextNote to Notify")
-
+        // Age + self-author gates run centrally in dispatchForAccount.
-        // old event being re-broadcast
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
-
-        // don't notify for own notes
-        if (event.pubKey == account.signer.pubKey) return
 
         val replyTargetId = event.replyingTo()
 
@@ -721,12 +707,7 @@ class EventNotificationConsumer(
         account: Account,
     ) {
         Log.d(TAG, "New NIP-22 Comment to Notify")
-
+        // Age + self-author gates run centrally in dispatchForAccount.
-        // old event being re-broadcast
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
-
-        // don't notify for own comments
-        if (event.pubKey == account.signer.pubKey) return
 
         // NIP-22 marks direct-reply and root authors. Notify when the current
         // account is either (someone commenting on our post, or replying to our comment).
@@ -811,12 +792,7 @@ class EventNotificationConsumer(
         event: Event,
         account: Account,
     ) {
-        // old event being re-broadcast
+        // Age + self-author gates run centrally in dispatchForAccount.
-        if (event.createdAt < TimeUtils.fifteenMinutesAgo()) return
-
-        // don't notify for our own events
-        if (event.pubKey == account.signer.pubKey) return
-
         val note = LocalCache.getNoteIfExists(event.id) ?: return
 
         val author = LocalCache.getOrCreateUser(event.pubKey)
@@ -855,10 +831,7 @@ class EventNotificationConsumer(
         account: Account,
         contentStringRes: Int,
     ) {
-        if (
+        // Age + self-author gates run centrally in dispatchForAccount.
-            event.createdAt > TimeUtils.fifteenMinutesAgo() &&
-            event.pubKey != account.signer.pubKey
-        ) {
         val author = LocalCache.getOrCreateUser(event.pubKey)
         val user = author.toBestDisplayName()
         val userPicture = author.profilePicture()
@@ -882,7 +855,6 @@ class EventNotificationConsumer(
                 applicationContext,
             )
     }
-    }
 
     private suspend fun notifyIncomingCall(
         event: CallOfferEvent,

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

@@ -170,8 +170,20 @@ class NotificationDispatcher(
 
                         Log.d(TAG) { "Observing notifications for ${pubkeys.size} account(s)." }
 
+                        // Rolling-window age cutoff that matches each downstream
+                        // notify() method's existing 15-min freshness rule. The
+                        // Nostr Filter's `since` is a fixed value captured at
+                        // dispatcher start; this predicate re-evaluates per event
+                        // so we keep pruning re-broadcasts as time advances.
+                        // Applied account-agnostically — calls use 20s downstream
+                        // (see CallManager.MAX_EVENT_AGE_SECONDS), which is strictly
+                        // stricter than 15 min, so they still pass.
+                        val freshnessPredicate: (Event) -> Boolean = { event ->
+                            event.createdAt >= TimeUtils.fifteenMinutesAgo()
+                        }
+
                         LocalCache
-                            .observeNewEvents<Event>(filter)
+                            .observeNewEvents<Event>(filter, freshnessPredicate)
                             .collect { event ->
                                 try {
                                     consumer.consumeFromCache(event)

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

@@ -31,9 +31,15 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
  * 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.
+ *
+ * The optional [predicate] runs after the protocol [filter] matches; use it
+ * for checks the Nostr [Filter] grammar can't express (e.g. rolling age
+ * cutoffs, arbitrary tag shapes, derived content fields). The predicate has
+ * to be fast — it runs on every new cache insertion.
  */
 class NewEventMatchingFilter<T : Event>(
     private val filter: Filter,
+    private val predicate: (Event) -> Boolean = { true },
     private val onNew: (T) -> Unit,
 ) : Observable {
     @Suppress("UNCHECKED_CAST")
@@ -41,7 +47,7 @@ class NewEventMatchingFilter<T : Event>(
         event: Event,
         note: Note,
     ) {
-        if (filter.match(event)) {
+        if (filter.match(event) && predicate(event)) {
             onNew(event as T)
         }
     }