From d394a38f2b7d928b7dfb58e75f886b1cd0845bc0 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Apr 2026 21:20:31 +0000 Subject: [PATCH] feat(quartz): emit StoreEvent.Delete from observable for filter / expired sweeps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces ObservableEventStore.events: SharedFlow with SharedFlow, where StoreEvent is: Insert(event) Delete(rule: DeleteRule) DeleteRule.Filtered(filters) ← from delete(filter) / delete(filters) DeleteRule.Expired(asOf) ← from deleteExpiredEvents() ObservableEventStore now emits Insert for accepted events, and Delete for every out-of-band removal that went through it. The Expired rule carries the cutoff timestamp (TimeUtils.now() at call time) so the projection's in-memory drop matches the store's on-disk drop without clock skew between collectors. EventStoreProjection no longer runs its own NIP-40 ticker. Expired events linger in [items] until the application calls deleteExpiredEvents() on the observable — at which point the projection drops everything whose expiration is past `asOf`. The ticker, expirationTickMs constructor arg, and sweepExpired helper are all removed. For DeleteRule.Filtered the projection iterates current slots and drops any whose event matches any of the rule's filters via Filter.match — same predicate the store uses. Tests: - nip40ExpirationDroppedByTicker → nip40ExpirationDroppedOnStoreSweep: drives a deleteExpiredEvents() call instead of waiting on a ticker. - New deleteByFilterRemovesMatchingSlots: confirms delete(filter) on the observable removes exactly the matching slots from the projection without touching others. - 15/15 projection tests + all other store tests pass. https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5 --- .../store/observable/ObservableEventStore.kt | 43 ++++-- .../nip01Core/store/observable/StoreEvent.kt | 71 +++++++++ .../store/projection/EventStoreProjection.kt | 136 +++++++++--------- .../projection/EventStoreProjectionTest.kt | 59 ++++++-- 4 files changed, 220 insertions(+), 89 deletions(-) create mode 100644 quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/StoreEvent.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt index 7757b9fba..3f66839d2 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/ObservableEventStore.kt @@ -27,6 +27,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore import com.vitorpamplona.quartz.nip01Core.store.projection.EventStoreProjection import com.vitorpamplona.quartz.nip40Expiration.isExpired +import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow @@ -63,18 +64,23 @@ class ObservableEventStore( val inner: IEventStore, ) : IEventStore { private val _events = - MutableSharedFlow( + MutableSharedFlow( replay = 0, extraBufferCapacity = 256, onBufferOverflow = BufferOverflow.SUSPEND, ) /** - * Stream of events accepted for observation, persisted or not. - * One emission per successful [insert] (or per successful entry - * inside [transaction]); rejected inserts emit nothing. + * Stream of mutations accepted by the observable layer. One + * emission per successful [insert] (or per accepted event in a + * [transaction] body), one emission per [delete] / [delete] / + * [deleteExpiredEvents] call. Rejected inserts and rolled-back + * transactions emit nothing. + * + * Projections consume this stream — see [EventStoreProjection] + * for how each [StoreEvent] is interpreted. */ - val events: SharedFlow = _events.asSharedFlow() + val events: SharedFlow = _events.asSharedFlow() override suspend fun insert(event: Event) { if (event.kind.isEphemeral()) { @@ -82,13 +88,13 @@ class ObservableEventStore( // that are already expired — they were never going to live // long enough for a UI to render them. if (event.isExpired()) return - _events.emit(event) + _events.emit(StoreEvent.Insert(event)) return } // Non-ephemeral: let the inner store enforce expiration, // tombstones, supersession, etc. If it throws, we never emit. inner.insert(event) - _events.emit(event) + _events.emit(StoreEvent.Insert(event)) } override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { @@ -115,7 +121,7 @@ class ObservableEventStore( } // Emit only after the inner transaction commits. If it throws // / rolls back, `accepted` is discarded. - for (e in accepted) _events.emit(e) + for (e in accepted) _events.emit(StoreEvent.Insert(e)) } override suspend fun query(filter: Filter): List = inner.query(filter) @@ -136,11 +142,26 @@ class ObservableEventStore( override suspend fun count(filters: List): Int = inner.count(filters) - override suspend fun delete(filter: Filter) = inner.delete(filter) + override suspend fun delete(filter: Filter) { + inner.delete(filter) + _events.emit(StoreEvent.Delete(DeleteRule.Filtered(listOf(filter)))) + } - override suspend fun delete(filters: List) = inner.delete(filters) + override suspend fun delete(filters: List) { + inner.delete(filters) + _events.emit(StoreEvent.Delete(DeleteRule.Filtered(filters))) + } - override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents() + override suspend fun deleteExpiredEvents() { + // Pin the cutoff before forwarding so the projection's drop + // matches the store's drop exactly. The store's SQL uses + // `unixepoch()` so there's still a small skew, but pinning at + // call time is closer than letting each projection use its + // own clock when the event is processed. + val asOf = TimeUtils.now() + inner.deleteExpiredEvents() + _events.emit(StoreEvent.Delete(DeleteRule.Expired(asOf))) + } /** * Open a reactive [EventStoreProjection] over this observable diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/StoreEvent.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/StoreEvent.kt new file mode 100644 index 000000000..83d46d46a --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/observable/StoreEvent.kt @@ -0,0 +1,71 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.store.observable + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter + +/** + * Mutations published by [ObservableEventStore.events]. Projections + * react to these to keep their in-memory view in sync with the + * underlying store. + * + * - [Insert] is emitted for every event accepted by the observable + * layer (persistable or ephemeral). Carries the event itself so + * the projection can run its NIP-01 / NIP-09 / NIP-62 + * interpretation. + * - [Delete] is emitted for every out-of-band removal — manual + * `delete(filter)` / `delete(filters)` calls and the periodic + * `deleteExpiredEvents()` sweep. Carries the rule the store used + * so the projection can apply the same selection in memory + * without re-querying. + */ +sealed interface StoreEvent { + data class Insert( + val event: Event, + ) : StoreEvent + + data class Delete( + val rule: DeleteRule, + ) : StoreEvent +} + +/** + * Selection rule for a [StoreEvent.Delete]. [Filtered] mirrors the + * `delete(filter)` / `delete(filters)` API and OR-combines the + * filters; [Expired] mirrors `deleteExpiredEvents()` and removes + * everything whose NIP-40 expiration has lapsed. + */ +sealed interface DeleteRule { + data class Filtered( + val filters: List, + ) : DeleteRule + + /** + * NIP-40 expiration sweep. Optional [asOf] timestamp (unix + * seconds) lets the store pin the cutoff it actually used, so + * the projection's in-memory drop matches the store's on-disk + * drop exactly. When `null` the projection uses its own clock. + */ + data class Expired( + val asOf: Long? = null, + ) : DeleteRule +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt index 4e7c00bb4..3843fd8f0 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjection.kt @@ -27,7 +27,9 @@ import com.vitorpamplona.quartz.nip01Core.core.isAddressable import com.vitorpamplona.quartz.nip01Core.core.isReplaceable import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.observable.DeleteRule import com.vitorpamplona.quartz.nip01Core.store.observable.ObservableEventStore +import com.vitorpamplona.quartz.nip01Core.store.observable.StoreEvent import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent import com.vitorpamplona.quartz.nip40Expiration.expiration import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent @@ -36,7 +38,6 @@ import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -64,30 +65,35 @@ import kotlinx.coroutines.yield * - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration) * drops the handle from the list. * - * The seed is materialised by querying the store once at start, after - * which the projection is driven entirely by [ObservableEventStore.events] - * and its own expiration ticker. The store is never re-queried on - * mutation, and the projection never asks the store to delete - * anything — it interprets incoming Nostr events itself: + * The seed is materialised by querying the store once at start, + * after which the projection is driven entirely by + * [ObservableEventStore.events]. Two kinds of mutation arrive on + * that stream: * - * - **NIP-01 supersession.** New replaceable / addressable events - * replace prior ones for the same `kind:pubkey[:dtag]`. The - * NIP-01 lexical-id tiebreaker (`new.id < old.id` when - * `created_at` ties) is honoured. - * - **NIP-09 deletions.** A [DeletionEvent] removes any matching - * handle owned by the same author (for GiftWrap, the recipient). - * Cross-author deletions are inert. - * - **NIP-62 right to vanish.** A [RequestToVanishEvent] whose - * `shouldVanishFrom([relay])` is true drops every handle from the - * same author with `created_at < vanish.created_at`. - * - **NIP-40 expiration.** Events with a past `expiration` tag are - * rejected at insert time. A periodic ticker drops slots whose - * expiration has just lapsed; collectors see the slot disappear. + * - [StoreEvent.Insert] — interpreted in-projection so a single + * arriving event can carry NIP-01 / NIP-09 / NIP-62 semantics: * - * That's the same set of rules the SQLite / FS stores enforce for - * durability. The duplication is by design — the store enforces them - * on disk so the file isn't corrupt; the projection enforces them in - * memory so the live view stays correct without a re-query per event. + * - **NIP-01 supersession.** New replaceable / addressable events + * replace prior ones for the same `kind:pubkey[:dtag]`. The + * NIP-01 lexical-id tiebreaker (`new.id < old.id` when + * `created_at` ties) is honoured. + * - **NIP-09 deletions.** A [DeletionEvent] removes any matching + * handle owned by the same author (for GiftWrap, the recipient). + * Cross-author deletions are inert. + * - **NIP-62 right to vanish.** A [RequestToVanishEvent] whose + * `shouldVanishFrom([relay])` is true drops every handle from + * the same author with `created_at < vanish.created_at`. + * - **NIP-40 expiration.** Events whose `expiration` tag has + * already lapsed at the moment they arrive are dropped before + * they ever enter [items]. + * + * - [StoreEvent.Delete] — emitted when a caller invokes + * `delete(filter)`, `delete(filters)`, or `deleteExpiredEvents()` + * on the [ObservableEventStore]. The projection drops every slot + * matching the rule using the same [Filter.match] / NIP-40 + * expiration semantics the store used. **There is no per-projection + * expiration ticker** — projections only drop expired events when + * the application calls `deleteExpiredEvents()` on the store. * * Limit handling: the initial query honours the filter `limit`, and * we trim to the same cap when an insert pushes the list over. We do @@ -96,29 +102,23 @@ import kotlinx.coroutines.yield * new arrives. That tradeoff matches what callers were already getting * from `LocalCache.observeEvents`. * - * Out-of-band store mutations — `delete(id)`, `delete(filter)`, - * `clearDB()`, the periodic `deleteExpiredEvents()` sweep — are not - * visible on [ObservableEventStore.events] and won't update an open - * projection. Re-open the projection (e.g. cancel and recreate the - * scope) to pick up an out-of-band change. - * * Ephemeral events (kinds `20000-29999`) reach the projection via * [ObservableEventStore.events] without ever being persisted; they * appear in [items] for as long as the projection is alive but never - * survive a re-seed. NIP-40 expiration applies to them too — if they - * carry an `expiration` tag, the per-projection ticker drops them - * when it lapses. + * survive a re-seed. They aren't covered by the store's + * `deleteExpiredEvents()` sweep (the DB never had them), so an + * ephemeral with an `expiration` tag will linger in the projection + * until it's superseded or until the projection is closed. * - * Lifecycle: the projection runs a collector + an expiration ticker - * inside [scope]. Cancel the scope (or call [close]) when the screen - * using the projection goes away. + * Lifecycle: the projection runs a single collector inside [scope]. + * Cancel the scope (or call [close]) when the screen using the + * projection goes away. */ class EventStoreProjection( private val store: ObservableEventStore, private val filters: List, private val relay: NormalizedRelayUrl?, scope: CoroutineScope, - private val expirationTickMs: Long = 30_000L, private val nowProvider: () -> Long = TimeUtils::now, ) : AutoCloseable { private val _items = MutableStateFlow>>(emptyList()) @@ -156,27 +156,17 @@ class EventStoreProjection( .onSubscription { seed() ready.complete(Unit) - }.collect { event -> apply(event) } - } - - private val expirationJob: Job = - scope.launch { - // Sleep first so the seed-time sweep covers the initial - // contents — see [seed]. - while (true) { - delay(expirationTickMs) - sweepExpired() - } + }.collect { storeEvent -> apply(storeEvent) } } private suspend fun seed() { val initial = store.query(filters) val now = nowProvider() for (event in initial) { - // The store should already exclude expired rows from the - // result, but it doesn't hurt to skip them here too — - // covers any FS / in-memory store that hasn't run a sweep - // recently. + // Stores don't filter expired events at query time, so we + // do it here — otherwise an expired event would briefly + // appear in [items] before the next deleteExpiredEvents() + // sweep clears it. if (isExpiredAt(event, now)) continue insertNew(event) yield() @@ -184,7 +174,14 @@ class EventStoreProjection( publish() } - private fun apply(event: Event) { + private fun apply(storeEvent: StoreEvent) { + when (storeEvent) { + is StoreEvent.Insert -> applyInsert(storeEvent.event) + is StoreEvent.Delete -> applyDelete(storeEvent.rule) + } + } + + private fun applyInsert(event: Event) { if (isExpiredAt(event, nowProvider())) return var changed = false @@ -207,6 +204,27 @@ class EventStoreProjection( if (changed) publish() } + private fun applyDelete(rule: DeleteRule) { + val targets = + when (rule) { + is DeleteRule.Filtered -> { + if (rule.filters.isEmpty()) return + byId.values.filter { slot -> rule.filters.any { it.match(slot.flow.value) } } + } + + is DeleteRule.Expired -> { + val cutoff = rule.asOf ?: nowProvider() + byId.values.filter { isExpiredAt(it.flow.value, cutoff) } + } + } + if (targets.isEmpty()) return + var changed = false + for (slot in targets) { + if (removeSlot(slot)) changed = true + } + if (changed) publish() + } + /** * Returns true if the event matches the filter and its arrival * caused membership to change (a fresh slot was added). Returns @@ -280,17 +298,6 @@ class EventStoreProjection( return changed } - private fun sweepExpired() { - val now = nowProvider() - val targets = byId.values.filter { isExpiredAt(it.flow.value, now) } - if (targets.isEmpty()) return - var changed = false - for (slot in targets) { - if (removeSlot(slot)) changed = true - } - if (changed) publish() - } - @Suppress("UNCHECKED_CAST") private fun insertNew(event: Event) { val slot = Slot(event as T) @@ -325,11 +332,10 @@ class EventStoreProjection( /** * Stop tracking changes and clear internal state. Idempotent. The * scope passed to the constructor keeps running; only this - * projection's collector + expiration jobs are cancelled. + * projection's collector job is cancelled. */ override fun close() { collectorJob.cancel() - expirationJob.cancel() ordered.clear() byId.clear() byStableKey.clear() diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt index 182847661..4e7382786 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/projection/EventStoreProjectionTest.kt @@ -349,12 +349,12 @@ class EventStoreProjectionTest { } /** - * NIP-40 per-projection ticker: a slot whose `expiration` lapses - * after the projection has loaded should be dropped on the next - * tick, even though the store hasn't run its sweep yet. + * NIP-40 expiration drops slots only when the application calls + * `deleteExpiredEvents()` on the observable store — projections + * no longer run their own ticker. */ @Test - fun nip40ExpirationDroppedByTicker() = + fun nip40ExpirationDroppedOnStoreSweep() = runBlocking { val time = TimeUtils.now() val safe = signer.sign(TextNoteEvent.build("safe", createdAt = time) { expiration(time + 100) }) @@ -362,26 +362,59 @@ class EventStoreProjectionTest { observable.insert(safe) observable.insert(short) - // Drive the ticker frequently so the test doesn't sit idle. val projection = - EventStoreProjection( - observable, - listOf(Filter(kinds = listOf(TextNoteEvent.KIND))), - relay = null, - scope = scope, - expirationTickMs = 100, + observable.observe( + Filter(kinds = listOf(TextNoteEvent.KIND)), + store.relay, + scope, ) projection.ready.await() assertEquals(2, projection.items.value.size) - // Wait past the short expiration. + // Let the short expiration lapse, then ask the store to + // sweep — the projection drops the expired slot in + // response to the resulting StoreEvent.Delete(Expired). delay(2000) + observable.deleteExpiredEvents() - val after = projection.awaitItems(timeoutMs = 5_000) { it.size == 1 } + val after = projection.awaitItems { it.size == 1 } assertEquals(safe.id, after[0].value.id) projection.close() } + /** + * `delete(filter)` on the observable propagates to open + * projections: they drop every slot matching the filter using + * the same Filter.match logic the store would. + */ + @Test + fun deleteByFilterRemovesMatchingSlots() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + val foreign = otherSigner.sign(TextNoteEvent.build("foreign", createdAt = 150)) + observable.insert(a) + observable.insert(b) + observable.insert(foreign) + + val projection = + observable.observe( + Filter(kinds = listOf(TextNoteEvent.KIND)), + store.relay, + scope, + ) + projection.ready.await() + assertEquals(3, projection.items.value.size) + + // Drop everything authored by `signer` — should leave + // only the foreign event. + observable.delete(Filter(authors = listOf(signer.pubKey))) + + val after = projection.awaitItems { it.size == 1 } + assertEquals(foreign.id, after[0].value.id) + projection.close() + } + @Test fun limitIsEnforcedOnInsertOverflow() = runBlocking {