diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.kt index fe0d02005..d0f206645 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.kt @@ -24,31 +24,21 @@ import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey /** - * Process-wide interner that canonicalises [Event] instances by id so - * every consumer (relay client, store deserialization, projections, - * tests) sees the same object reference for the same event id. + * Canonicalises [Event] instances by id so every consumer sees the + * same object reference for the same event id. * * Backed by weak references — entries vanish when no live consumer - * holds the event, so the cache only grows as long as projections / - * UI state actually need the events. Sized for ~5000 hot events; the - * underlying map resizes if usage exceeds that. - * - * Use [intern] on every event arrival path. The first occurrence wins - * and becomes canonical; subsequent equivalent decodes return that - * canonical instance. + * holds the event. First-seen wins; equivalence is by event id, so + * callers must trust ids are content-derived (signed events satisfy + * this). Sized for ~5000 hot entries; resizes if usage exceeds that. * * Platforms without weak references fall back to a passthrough that - * returns [event] unchanged — no canonicalisation, but no leaks - * either. + * returns [event] unchanged — no canonicalisation, no leaks. */ expect class EventInterner() { /** - * Returns the canonical [Event] for [event]'s id. If a live - * canonical instance already exists for this id, returns it; - * otherwise stores [event] as the new canonical and returns it. - * - * Equivalence is by event id only — callers must trust the id - * was content-derived (signed events satisfy this). + * Returns the canonical [Event] for [event]'s id, storing + * [event] as canonical if no live entry exists. */ fun intern(event: Event): Event @@ -62,7 +52,7 @@ expect class EventInterner() { fun clear() companion object { - /** Process-wide default interner used by [Event.fromJson]. */ + /** Process-wide shared instance. */ val Default: EventInterner } } 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 bcd43f082..31092bd34 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 @@ -58,70 +58,35 @@ sealed interface ProjectionState { /** * State machine that maintains a reactive view over an - * [ObservableEventStore] for a fixed set of [filters]. Each visible - * event is wrapped in a [MutableStateFlow] so the UI can collect - * three different kinds of change with the right granularity: + * [ObservableEventStore] for a fixed set of [filters]. Pure logic, no + * coroutine ownership: construct, [seed] once, then [apply] each + * [StoreChange] and read [snapshot]. For the standard "wrap into a + * cold Flow and collect from a ViewModel" path, use + * [ObservableEventStore.project] which composes this class. * - * - **Membership** (events arriving or leaving) produces a brand - * new [List] in [snapshot]. The list reference is stable while - * membership is unchanged. - * - **In-place replaceable / addressable update** (a new version of - * the same `kind:pubkey:dtag` arrives) updates the existing - * handle's [MutableStateFlow.value] without changing membership. - * Only collectors of that one handle re-render. The list ordering - * is *not* reshuffled when the new version has a later - * `created_at` — each slot remembers the sort key it was inserted - * with, so updates feel like pure value mutations. - * - **Removal** (NIP-09 deletion, NIP-62 vanish, NIP-40 expiration, - * `delete(filter)`) drops the handle from the list. + * Each visible event is wrapped in a [MutableStateFlow], giving the + * UI three change granularities: * - * Three kinds of [StoreChange] are interpreted in-projection: + * - **Membership** (insert / removal): [snapshot] returns a fresh + * 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.* + * - **Removal**: NIP-09, NIP-62, NIP-40 expiration, `delete(filter)`. * - * - [StoreChange.Insert] — a single arriving event can carry NIP-01 - * / NIP-09 / NIP-62 semantics: - * - **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(store.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 the snapshot. - * - [StoreChange.DeleteByFilter] — emitted on `delete(filter)` / - * `delete(filters)`. Drops every slot matching any of the rule's - * filters via [Filter.match]. - * - [StoreChange.DeleteExpired] — emitted on `deleteExpiredEvents()`. - * Drops every slot whose `expiration` has lapsed at the cutoff the - * store pinned. **There is no per-projection expiration ticker** — - * expiration only triggers when the application calls - * `deleteExpiredEvents()` on the store. + * Limit is **per-filter**: each filter retains at most its own + * `limit` matches; the snapshot is the deduped union, so disjoint + * matches across filters can exceed any single cap. Removed slots + * are not refilled from the store. * - * Limit handling is **per-filter**: each filter retains at most its - * own `limit` matches in a private capped set, sorted by created_at - * DESC + id ASC. The snapshot is the deduped union of those sets, so - * when filter A and filter B match disjoint events the union can be - * larger than any single filter's `limit`. We do not refill from the - * store after a deletion — if a removal leaves a filter under cap, it - * stays under cap until another match arrives. - * - * Ephemeral events (kinds `20000-29999`) reach the projection without - * ever being persisted; they appear in the snapshot for as long as - * the projection is alive but never 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 - * until it's superseded or until the projection is closed. - * - * **Lifecycle**: this class is a pure state machine with no - * coroutine ownership. Construct it, call [seed] once, then call - * [apply] for each [StoreChange] from [ObservableEventStore.changes]. - * For the standard "wrap into a Flow and collect from a ViewModel" - * use case, use the [ObservableEventStore.project] extension which - * does this composition for you. + * Ephemeral events appear in the snapshot for as long as the + * projection is alive; they never survive a re-seed (the inner store + * never had them) and aren't touched by `deleteExpiredEvents()`. */ class EventStoreProjection( private val store: ObservableEventStore, @@ -174,9 +139,8 @@ class EventStoreProjection( dropWhere { ev -> storeEvent.filters.any { it.match(ev) } } } - // Store's sweep uses strict `<`; isExpirationBefore is - // `<=`, so subtract 1 to match. is StoreChange.DeleteExpired -> { + // Store's sweep uses strict `<`; isExpirationBefore is `<=`, so subtract 1. val cutoff = (storeEvent.asOf ?: nowProvider()) - 1 dropWhere { it.isExpirationBefore(cutoff) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt index a65c1e5e1..1def6bf4f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -32,31 +32,18 @@ import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.asSharedFlow /** - * A reactive façade over any [IEventStore] that publishes every event - * accepted for *observation* — a superset of the events the inner - * store persists. + * Reactive façade over any [IEventStore]. Publishes a [StoreChange] + * on [changes] for every accepted mutation so projections can stay in + * sync without re-querying. * - * The split between persistence and observation is the whole point of - * this class: - * - * - **Non-ephemeral events** are forwarded to the inner store. If the - * inner store rejects (expired, NIP-09 / NIP-62 tombstone, NIP-01 - * supersession loser), the rejection propagates and nothing is - * emitted on [changes]. - * - **Ephemeral events** (kinds `20000-29999`) skip the inner store - * entirely — they're never persisted — but they still emit on - * [changes] so projections can render them while they live. Already - * expired ephemerals are silently dropped. - * - * Wrap any store you want to observe — `SQLiteEventStore`, FS-backed, - * an in-memory test fake — and feed `EventStoreProjection` from the - * resulting [changes] flow. - * - * Reads (`query`, `count`) and out-of-band writes (`delete`, - * `deleteExpiredEvents`) forward to the inner store; the latter are - * also surfaced on [changes] as [StoreChange.DeleteByFilter] / - * [StoreChange.DeleteExpired] so projections can drop the matching - * slots in memory without re-querying. + * - Non-ephemeral [insert] / [transaction] entries forward to the + * inner store. On rejection (expired, NIP-09 / NIP-62 tombstone, + * NIP-01 supersession loser) the throw propagates and nothing is + * emitted. + * - Ephemeral events (kinds `20000-29999`) skip the inner store but + * still emit. Already-expired ephemerals are silently dropped. + * - [delete] and [deleteExpiredEvents] also emit so projections drop + * the matching slots in memory. */ class ObservableEventStore( val inner: IEventStore, @@ -70,35 +57,15 @@ class ObservableEventStore( onBufferOverflow = BufferOverflow.SUSPEND, ) - /** - * 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] / - * [deleteExpiredEvents] call. Rejected inserts and rolled-back - * transactions emit nothing. - * - * Projections consume this stream — see `EventStoreProjection` - * for how each [StoreChange] is interpreted. - */ + /** Stream of mutations accepted by this layer. See [StoreChange] for the cases. */ val changes: SharedFlow = _changes.asSharedFlow() /** - * Mutations published by [changes]. 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. - * - [DeleteByFilter] is emitted for every `delete(filter)` / - * `delete(filters)` call on the observable. Carries the same - * filters the store used so projections can apply - * [Filter.match] in memory and drop the matching slots without - * re-querying. - * - [DeleteExpired] is emitted for every `deleteExpiredEvents()` - * sweep. The optional [DeleteExpired.asOf] cutoff lets the - * store pin the timestamp it actually used, so the projection - * drops exactly the events the store dropped. + * One [Insert] per accepted event; one [DeleteByFilter] per + * `delete(filter[s])`; one [DeleteExpired] per + * `deleteExpiredEvents()` sweep. The cutoff carried by + * [DeleteExpired] is pinned at the moment the sweep ran so + * projections drop exactly the events the store dropped. */ sealed interface StoreChange { data class Insert(