From 86a86e142681e51a211b4e820b296e4664d6cccc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 24 Apr 2026 14:39:42 +0000 Subject: [PATCH] refactor(notifications): observer takes a single predicate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Filter.match is itself just a boolean predicate, so carrying a Filter and a separate composition predicate on NewEventMatchingFilter duplicated the abstraction. Collapse to one predicate: - NewEventMatchingFilter(predicate, onNew): drops the Filter parameter. - LocalCache.observeNewEvents(predicate): primary API. - LocalCache.observeNewEvents(filter): kept as a one-liner that forwards filter::match, so existing feed/DAL callers are unchanged. In the dispatcher the freshness check, kind check, p-tag match, and since cutoff now live in a single lambda, short-circuiting on cheap kind comparison first. NOTIFICATION_KINDS is now a Set for O(1) membership instead of O(n) List.contains — cheap, but on a hot path that runs for every new cache insertion it pays for itself. https://claude.ai/code/session_01GQDJxiHPogdzCNhUBN7Pjc --- .../amethyst/model/LocalCache.kt | 23 +++++----- .../notifications/NotificationDispatcher.kt | 45 +++++++++---------- .../observables/NewEventMatchingFilter.kt | 25 +++++------ 3 files changed, 44 insertions(+), 49 deletions(-) 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 644025801..00f9cb467 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt @@ -383,21 +383,20 @@ 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. + * Emits each new event for which [predicate] returns true, 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). + * The predicate runs on every insertion, so keep it cheap. Callers with a + * Nostr [Filter] can pass `filter::match`; compose additional local checks + * (rolling windows, derived fields the Filter grammar can't express) with + * `&&`. */ - fun observeNewEvents( - filter: Filter, - predicate: (Event) -> Boolean = { true }, - ): Flow = + fun observeNewEvents(predicate: (Event) -> Boolean): Flow = callbackFlow { val newFilter = - NewEventMatchingFilter(filter, predicate) { + NewEventMatchingFilter(predicate) { trySend(it) } @@ -408,6 +407,8 @@ object LocalCache : ILocalCache, ICacheProvider { } } + fun observeNewEvents(filter: Filter): Flow = observeNewEvents(filter::match) + @Suppress("UNCHECKED_CAST") fun observeLatestEvent(filter: Filter) = observeEvents(filter).map { it.firstOrNull() } 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 index 147c8070f..c78d8325a 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt @@ -28,7 +28,6 @@ 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.core.toHexKey -import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip17Dm.files.ChatMessageEncryptedFileHeaderEvent @@ -91,8 +90,8 @@ class NotificationDispatcher( // consumeFromCache can't route it. It's delivered directly via // [notifyWelcome] from processMarmotWelcomeFlow, which does know the // recipient account. - private val NOTIFICATION_KINDS = - listOf( + private val NOTIFICATION_KINDS: Set = + setOf( // Direct-arrival PrivateDmEvent.KIND, LnZapEvent.KIND, @@ -158,32 +157,30 @@ class NotificationDispatcher( return@collectLatest } - val filter = - Filter( - kinds = NOTIFICATION_KINDS, - // Only route events that p-tag one of our accounts. - // consumeFromCache still routes by `p` tag, so the - // filter narrows exactly the same set it would accept. - tags = mapOf("p" to pubkeys.toList()), - since = dispatcherSince, - ) - 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() + // Single observer predicate. Each check is cheap and + // short-circuits so kind mismatch (by far the most + // common case) rejects before any allocation. + // + // - kind ∈ NOTIFICATION_KINDS — channel-relevant types + // - createdAt ≥ dispatcherSince — `limit: 0` semantics, + // drops re-broadcasts from before this session + // - createdAt ≥ fifteenMinutesAgo — rolling freshness, + // matches the downstream per-channel policy. Calls + // use a stricter 20s check in notifyIncomingCall so + // they still pass through. + // - any `p` tag matching one of our accounts — narrows + // to events consumeFromCache would route anyway. + val predicate = { event: Event -> + event.kind in NOTIFICATION_KINDS && + event.createdAt >= dispatcherSince && + event.createdAt >= TimeUtils.fifteenMinutesAgo() && + event.tags.any { it.size > 1 && it[0] == "p" && it[1] in pubkeys } } LocalCache - .observeNewEvents(filter, freshnessPredicate) + .observeNewEvents(predicate) .collect { event -> try { consumer.consumeFromCache(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 index 45e9b98f3..436f3933e 100644 --- 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 @@ -22,24 +22,21 @@ 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. + * Emits each new event for which [predicate] returns true, one at a time, + * as it is inserted into the cache. Unlike [EventListMatchingFilter] / + * [NoteListMatchingFilter], this does not accumulate a list — it simply + * calls [onNew] per matching event. Useful for reactive event-triggered + * pipelines like notifications. * - * 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. - * - * 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. + * The predicate has to be fast — it runs on every new cache insertion. + * Callers with a Nostr [com.vitorpamplona.quartz.nip01Core.relay.filters.Filter] + * can pass `filter::match` as the predicate and compose additional checks + * with `&&`. */ class NewEventMatchingFilter( - private val filter: Filter, - private val predicate: (Event) -> Boolean = { true }, + private val predicate: (Event) -> Boolean, private val onNew: (T) -> Unit, ) : Observable { @Suppress("UNCHECKED_CAST") @@ -47,7 +44,7 @@ class NewEventMatchingFilter( event: Event, note: Note, ) { - if (filter.match(event) && predicate(event)) { + if (predicate(event)) { onNew(event as T) } }