nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor(notifications): observer takes a single predicate

Browse Source 
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<Int> 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
This commit is contained in:
Claude
2026-04-24 14:39:42 +00:00
parent 12a5926f6a
commit 86a86e1426
3 changed files with 44 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt
+12 -11
View File
@@ -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 <T : Event> observeNewEvents(
filter: Filter,
predicate: (Event) -> Boolean = { true },
): Flow<T> =
fun <T : Event> observeNewEvents(predicate: (Event) -> Boolean): Flow<T> =
callbackFlow {
val newFilter =
NewEventMatchingFilter<T>(filter, predicate) {
NewEventMatchingFilter<T>(predicate) {
trySend(it)
}
@@ -408,6 +407,8 @@ object LocalCache : ILocalCache, ICacheProvider {
}
}
fun <T : Event> observeNewEvents(filter: Filter): Flow<T> = observeNewEvents(filter::match)
@Suppress("UNCHECKED_CAST")
fun <T : Event> observeLatestEvent(filter: Filter) = observeEvents<T>(filter).map { it.firstOrNull() }
amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationDispatcher.kt
+21 -24
View File
@@ -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<Int> =
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<Event>(filter, freshnessPredicate)
.observeNewEvents<Event>(predicate)
.collect { event ->
try {
consumer.consumeFromCache(event)
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NewEventMatchingFilter.kt
+11 -14
View File
@@ -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<T : Event>(
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<T : Event>(
event: Event,
note: Note,
) {
if (filter.match(event) && predicate(event)) {
if (predicate(event)) {
onNew(event as T)
}
}