fix: track event counts per filter in downloadFromRelay

Change the event counter from a single global counter to per-filter
tracking using Filter.match(). Each filter's limit is now respected
individually - fulfilled filters are excluded from subsequent pages,
and pagination stops when all filters with limits are satisfied.

https://claude.ai/code/session_01Baaira4xoEHEsaSQCX52Sy
This commit is contained in:
Claude
2026-03-17 17:41:40 +00:00
parent 034cc0ab6d
commit 76c00dea3c
@@ -40,6 +40,11 @@ import kotlin.coroutines.coroutineContext
* After EOSE the oldest [Event.createdAt] seen in that page minus one becomes the
* next `until`, and the query repeats until the relay returns no new events.
*
* Event counting is tracked per filter using [Filter.match]. A filter is considered
* fulfilled when the number of matching events reaches its [Filter.limit]. Pagination
* stops when all filters with limits are fulfilled or when a page returns no events.
* Filters without a limit are considered unbounded and only stop on empty pages.
*
* @param relay The relay to query.
* @param baseFilters Filters to apply on every page (the `until` field is overwritten per page).
* @param timeoutMs Maximum time to wait for a single page's EOSE before giving up.
@@ -55,18 +60,32 @@ suspend fun INostrClient.downloadFromRelay(
var until: Long? = null
var totalEvents = 0
// Track how many matching events each filter has received so far.
val matchCountPerFilter = IntArray(baseFilters.size)
while (true) {
coroutineContext.ensureActive()
// Only include filters that still need more events.
val activeFilterIndices =
baseFilters.indices.filter { i ->
val limit = baseFilters[i].limit
limit == null || matchCountPerFilter[i] < limit
}
if (activeFilterIndices.isEmpty()) break
val activeBaseFilters = activeFilterIndices.map { baseFilters[it] }
val eventChannel = Channel<Event>(UNLIMITED)
val doneChannel = Channel<Unit>(Channel.CONFLATED)
val subId = newSubId()
val filters =
if (until == null) {
baseFilters
activeBaseFilters
} else {
baseFilters.map { it.copy(until = until) }
activeBaseFilters.map { it.copy(until = until) }
}
val listener =
@@ -116,6 +135,13 @@ suspend fun INostrClient.downloadFromRelay(
onEvent(event)
pageCount++
if (event.createdAt < pageMinTs) pageMinTs = event.createdAt
// Count this event against every base filter it matches.
for (i in baseFilters.indices) {
if (baseFilters[i].match(event)) {
matchCountPerFilter[i]++
}
}
}
if (pageCount == 0) break