diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt index 31092bd34..d734e7e95 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt @@ -71,12 +71,13 @@ sealed interface ProjectionState { * list. Stable list reference while membership is unchanged. * - **In-place replaceable / addressable update**: same slot, new * [MutableStateFlow.value]. Only collectors of that handle - * re-render. The slot's sort key is frozen at insertion, so the - * list does not reshuffle when the new version has a later - * `created_at`. *Caveat: the slot stays in whatever filters - * matched the original version; if the new version no longer - * matches a filter (e.g. its tags changed), the slot is not - * re-evaluated and remains live.* + * re-render. The slot's sort key is frozen at insertion so the + * list ordering doesn't reshuffle on a later `created_at`. Filter + * membership *is* re-evaluated against the new event, so a v2 + * that no longer matches a filter (e.g. tag changed) drops out; + * a v2 that newly matches another filter joins it. In the common + * case (filter on `kinds + authors`) v2 still matches and the + * list reference stays stable. * - **Removal**: NIP-09, NIP-62, NIP-40 expiration, `delete(filter)`. * * Limit is **per-filter**: each filter retains at most its own @@ -194,17 +195,37 @@ class EventStoreProjection( if (existing != null) { if (!supersedes(event, existing.flow.value)) return false - // Same address, new winner. Rekey byId from the - // previous event id to the new one and update the - // handle's value in place — list reference stays the - // same; only the handle's collectors re-render. + // Same address, new winner. Rekey byId, mutate + // flow.value in place, then re-evaluate filter + // membership against the new event — v2 may add + // matches or lose previously-matching filters. val previousId = existing.flow.value.id if (previousId != event.id) { byId.remove(previousId) byId[event.id] = existing } existing.flow.value = event as T - return false + + var changed = false + for ((f, set) in perFilter) { + val nowMatches = f.match(event) + val wasIn = set.contains(existing) + when { + nowMatches && !wasIn -> { + if (admit(existing, f, set)) changed = true + } + + !nowMatches && wasIn -> { + set.remove(existing) + changed = true + } + } + } + // If no filter retains the slot anymore, fully drop it. + if (perFilter.values.none { it.contains(existing) }) { + if (removeIndexes(existing)) changed = true + } + return changed } } else if (byId.containsKey(event.id)) { return false @@ -213,32 +234,42 @@ class EventStoreProjection( // Genuinely new slot. Offer it to every matching filter; if // any filter still holds it after cap-eviction, the slot // becomes live and gets indexed. - var membershipChanged = false + var changed = false val slot = Slot(event as T) for ((f, set) in perFilter) { if (!f.match(event)) continue - set.add(slot) - val cap = f.limit ?: continue - while (set.size > cap) { - val tail = set.last() - set.remove(tail) - if (tail !== slot && perFilter.values.none { it.contains(tail) }) { - // Tail no longer retained by any filter — fully drop. - if (removeIndexes(tail)) membershipChanged = true - } - } + if (admit(slot, f, set)) changed = true } - - // The slot survived cap-eviction in at least one filter, so it - // belongs in the indexes. Otherwise nothing was indexed and - // the only membership effect is whatever evictions happened - // along the way. if (perFilter.values.any { it.contains(slot) }) { byId[event.id] = slot if (address != null) byAddress[address] = slot - membershipChanged = true + changed = true } - return membershipChanged + return changed + } + + /** + * Add [slot] to filter [f]'s [set] and evict the tail if the cap + * is exceeded. Returns true if an eviction triggered a slot drop + * from the indexes (its only filter retention was the evicted + * one). + */ + private fun admit( + slot: Slot, + f: Filter, + set: java.util.SortedSet>, + ): Boolean { + set.add(slot) + val cap = f.limit ?: return false + var changed = false + while (set.size > cap) { + val tail = set.last() + set.remove(tail) + if (tail !== slot && perFilter.values.none { it.contains(tail) }) { + if (removeIndexes(tail)) changed = true + } + } + return changed } private fun handleDeletion(deletion: DeletionEvent): Boolean { diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt index 68a98e28b..992b43620 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync import com.vitorpamplona.quartz.nip01Core.store.ObservableEventStore import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore +import com.vitorpamplona.quartz.nip01Core.tags.hashtags.hashtag import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip23LongContent.LongTextNoteEvent @@ -208,6 +209,49 @@ class EventStoreProjectionTest { assertSame(seedList, projection.items, "addressable update must not change list reference") } + /** + * If v2 of an addressable no longer matches the filter (e.g. + * tag list changed), the slot is dropped. The projection + * re-evaluates filter membership on every supersession. + */ + @Test + fun addressableUpdateDropsSlotWhenFilterStopsMatching() = + runBlocking { + val time = TimeUtils.now() + // v1 carries tag "nostr"; v2 changes the tag to "bitcoin". + val v1 = + signer.sign( + LongTextNoteEvent.build("blog v1", "title", dTag = "blog", createdAt = time) { + hashtag("nostr") + }, + ) + observable.insert(v1) + + // Filter narrows to events that ALSO carry hashtag "nostr". + val projection = + projectionOf( + Filter( + kinds = listOf(LongTextNoteEvent.KIND), + authors = listOf(v1.pubKey), + tags = mapOf("t" to listOf("nostr")), + ), + ) + projection.awaitLoaded() + assertEquals(1, projection.items.size) + + val v2 = + signer.sign( + LongTextNoteEvent.build("blog v2", "title", dTag = "blog", createdAt = time + 1) { + hashtag("bitcoin") + }, + ) + observable.insert(v2) + + // v2 doesn't match — slot drops and the snapshot is empty. + val after = projection.awaitItems { it.isEmpty() } + assertTrue(after.isEmpty()) + } + /** * Out-of-order arrival: the projection sees the *newer* version * first (e.g. the relay sent v2 first), then v1. The projection