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 {