diff --git a/quartz/README.md b/quartz/README.md new file mode 100644 index 000000000..f9732c119 --- /dev/null +++ b/quartz/README.md @@ -0,0 +1,143 @@ +# Quartz + +A Kotlin Multiplatform Nostr library — protocol, signing, relay client, event store, and reactive projections. No UI, no Android dependencies in `commonMain`. Targets Android, JVM/Desktop, iOS, and Linux. + +## Layered architecture + +Build apps by composing layers from durable storage at the bottom up to UI projections at the top: + +``` +┌──────────────────────────────────────────┐ +│ UI / ViewModel │ +│ observable.project(filter) │ +│ .stateIn(scope, ...) │ ← reactive list of MutableStateFlow +└──────────────────────────────────────────┘ + ▲ +┌──────────────────────────────────────────┐ +│ ObservableEventStore │ +│ publishes StoreChange on `changes` │ ← bus for reactive consumers +└──────────────────────────────────────────┘ + ▲ +┌──────────────────────────────────────────┐ +│ InterningEventStore │ +│ one Event instance per id, weak refs │ ← shared identity across reads +└──────────────────────────────────────────┘ + ▲ +┌──────────────────────────────────────────┐ +│ EventStore (SQLite) / FsEventStore │ +│ durable, NIP-01/09/40/62 enforced │ ← persistence + Nostr semantics +└──────────────────────────────────────────┘ + ▲ +┌──────────────────────────────────────────┐ +│ NostrClient │ +│ relay subscriptions, NIP-01 messages │ ← network +└──────────────────────────────────────────┘ +``` + +Each layer is optional — you can use just `NostrClient` (see [CLIENT.md](CLIENT.md)), just `EventStore` (see [the SQLite store README](src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/README.md)), or compose the full pipeline below. + +## Wiring relays into the store + +`NostrClient` delivers events through a `SubscriptionListener.onEvent` callback. Pipe each event into the `ObservableEventStore`, which persists it (via the inner store) and publishes it to every open projection: + +```kotlin +// Application init — one set of these per process. +val sqlite = EventStore(dbName = "events.db", relay = "wss://relay.damus.io".normalizeRelayUrl()) +val observable = ObservableEventStore(InterningEventStore(sqlite)) + +val client = NostrClient(websocketBuilder = ktorBuilder) +client.connect() + +// Open a relay subscription that pumps every arriving event into the store. +client.subscribe( + subId = "home", + filters = mapOf( + "wss://relay.damus.io".normalizeRelayUrl() to listOf( + Filter(kinds = listOf(1), authors = followedAuthors, limit = 500), + ), + ), + listener = object : SubscriptionListener { + override fun onEvent(event: Event, isLive: Boolean, relay: NormalizedRelayUrl, forFilters: List?) { + applicationScope.launch { observable.insert(event) } + } + }, +) + +// Periodic NIP-40 sweep — projections drop expired events when this fires. +applicationScope.launch { + while (isActive) { delay(15.minutes); observable.deleteExpiredEvents() } +} +``` + +`observable.insert(event)` is idempotent under NIP-01 supersession (older replaceables / addressables are rejected by the inner store) and validates expiration / vanish tombstones, so it's safe to fire-and-forget for every relay arrival. + +`InterningEventStore` keeps one `Event` instance alive per id (weakly, via `EventInterner.Default`), so events that re-arrive from multiple relays — or get re-read by query — share the same object reference as long as some projection holds them. + +## Building a reactive feed UI + +A feed screen reads from the store via `ObservableEventStore.project()`, which returns a cold `Flow>`. Wrap it with `stateIn(...)` in a ViewModel and collect from Compose: + +```kotlin +class HomeFeedViewModel( + observable: ObservableEventStore, + followedAuthors: List, +) : ViewModel() { + val state: StateFlow> = + observable + .project(Filter(kinds = listOf(1), authors = followedAuthors, limit = 200)) + .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), ProjectionState.Loading) +} + +@Composable +fun HomeFeedScreen(vm: HomeFeedViewModel) { + val state by vm.state.collectAsState() + when (val s = state) { + is ProjectionState.Loading -> SpinnerScaffold() + is ProjectionState.Loaded -> { + // Layer 1: list reference is stable while membership is unchanged. + // LazyColumn only invalidates structure on inserts / removals. + LazyColumn { + items(s.items, key = { it.value.id }) { handle -> + NoteRow(handle) + } + } + } + } +} + +@Composable +fun NoteRow(handle: MutableStateFlow) { + // Layer 2: only collectors of THIS handle re-render when the + // event mutates in place (addressable supersession, etc.). + val event by handle.collectAsState() + Column { + Text(event.content) + // Layer 3: derived flows — e.g. counters reactive to OTHER projections. + ReactionRow(event.id) + } +} +``` + +The three layers map cleanly to Compose's recomposition model: + +- `state` re-renders the screen scaffolding (`Loading` vs `Loaded`). +- `s.items` re-emits only when membership changes (insert, NIP-09 deletion, NIP-62 vanish, NIP-40 expiration sweep, manual `delete(filter)`). +- `handle` re-emits only when its specific event mutates in place (e.g. a new version of a kind-30023 long-form post supersedes the previous one). + +## Publishing from the UI + +```kotlin +val signer = NostrSignerInternal(KeyPair()) +val signed = signer.sign(TextNoteEvent.build("hello nostr", createdAt = TimeUtils.now())) +observable.insert(signed) // hits the bus → all open projections see it +client.send(signed, relays = ...) // also publish to relays +``` + +The projection runs NIP-01 supersession, NIP-09 author checks, NIP-62 vanish scoping, and NIP-40 expiration filtering automatically. UI code never needs to know any of those rules. + +## Where to read more + +- [`CLIENT.md`](CLIENT.md) — building a relay client with `NostrClient` and Ktor. +- [`RELAY.md`](RELAY.md) — running a relay server with Quartz. +- [SQLite event store README](src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/README.md) — query planner, indexing strategies, NIP-09/40/62 enforcement details. +- KDoc on `EventStoreProjection`, `ObservableEventStore`, `EventInterner` — package-level reference for the projection layer. diff --git a/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.apple.kt b/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.apple.kt new file mode 100644 index 000000000..a28c22be6 --- /dev/null +++ b/quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.apple.kt @@ -0,0 +1,45 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * Apple actual: passthrough. Kotlin/Native has weak refs but no + * built-in concurrent map; rather than ship a half-baked impl, we + * skip canonicalisation entirely on Apple targets and let + * [Event.fromJson] return whatever the deserializer produced. Can + * be revisited if Amethyst's iOS port wants the memory savings. + */ +actual class EventInterner { + actual fun intern(event: Event): Event = event + + actual fun get(id: HexKey): Event? = null + + actual fun size(): Int = 0 + + actual fun clear() {} + + actual companion object { + actual val Default: EventInterner = EventInterner() + } +} 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 new file mode 100644 index 000000000..d0f206645 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.kt @@ -0,0 +1,58 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * 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. 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, no leaks. + */ +expect class EventInterner() { + /** + * Returns the canonical [Event] for [event]'s id, storing + * [event] as canonical if no live entry exists. + */ + fun intern(event: Event): Event + + /** Returns the canonical [Event] for [id] if one is live, else null. */ + fun get(id: HexKey): Event? + + /** Number of map entries (including dead weak refs not yet cleaned). */ + fun size(): Int + + /** Drop every entry. Mostly useful for tests. */ + fun clear() + + companion object { + /** Process-wide shared instance. */ + val Default: EventInterner + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt new file mode 100644 index 000000000..d224a17d9 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/InterningEventStore.kt @@ -0,0 +1,121 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.store.IEventStore + +/** + * Decorator that canonicalises every [Event] returned by the inner + * store through an [EventInterner], so all consumers of read paths + * share a single object reference per event id. + * + * Wrap any [IEventStore] (SQLite, FS, in-memory test fake) when you + * want event-identity sharing across the read paths: + * + * ``` + * val sqlite = EventStore(...) + * val cached = InterningEventStore(sqlite) + * val observable = ObservableEventStore(cached) + * ``` + * + * Writes ([insert] / [transaction]) forward to the inner store and, + * on success, register the accepted event with the interner. The + * caller's instance becomes canonical for that id (so a later read + * of the same id returns the same `===` reference, until the weak + * ref is collected). The decorator does *not* substitute the + * caller's event for a previously-cached one on the way in — sigs + * may differ across "same id, different decode" cases, and the + * caller's instance is the freshest. + * + * Reads ([query]) pipe results through `interner.intern`, returning + * canonical instances for ids that have a live cached entry. + * + * Out-of-band removals (`delete*`, `deleteExpiredEvents`) pass + * through unchanged. + * + * The default [interner] is [EventInterner.Default]; pass a fresh + * instance for tests or any context that needs isolation. + */ +class InterningEventStore( + private val inner: IEventStore, + private val interner: EventInterner = EventInterner.Default, +) : IEventStore { + override val relay: NormalizedRelayUrl? get() = inner.relay + + override suspend fun insert(event: Event) { + inner.insert(event) + // Inner accepted (didn't throw) — register the canonical + // instance so subsequent reads share the same reference. + interner.intern(event) + } + + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { + // Capture every accepted event so we can intern after the + // inner transaction commits. If body throws or the inner + // rolls back, `accepted` is discarded with the txn. + val accepted = ArrayList() + inner.transaction { + val innerTxn = this + val wrapped = + object : IEventStore.ITransaction { + override fun insert(event: Event) { + innerTxn.insert(event) + accepted.add(event) + } + } + wrapped.body() + } + for (e in accepted) interner.intern(e) + } + + @Suppress("UNCHECKED_CAST") + override suspend fun query(filter: Filter): List = inner.query(filter).map { interner.intern(it) as T } + + @Suppress("UNCHECKED_CAST") + override suspend fun query(filters: List): List = inner.query(filters).map { interner.intern(it) as T } + + @Suppress("UNCHECKED_CAST") + override suspend fun query( + filter: Filter, + onEach: (T) -> Unit, + ) = inner.query(filter) { onEach(interner.intern(it) as T) } + + @Suppress("UNCHECKED_CAST") + override suspend fun query( + filters: List, + onEach: (T) -> Unit, + ) = inner.query(filters) { onEach(interner.intern(it) as T) } + + override suspend fun count(filter: Filter): Int = inner.count(filter) + + override suspend fun count(filters: List): Int = inner.count(filters) + + override suspend fun delete(filter: Filter) = inner.delete(filter) + + override suspend fun delete(filters: List) = inner.delete(filters) + + override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents() + + override fun close() = inner.close() +} 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 new file mode 100644 index 000000000..ba8fd803b --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjection.kt @@ -0,0 +1,436 @@ +/* + * 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.cache.projection + +import com.vitorpamplona.quartz.nip01Core.core.Address +import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.isReplaceable +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.store.ObservableEventStore +import com.vitorpamplona.quartz.nip01Core.store.ObservableEventStore.StoreChange +import com.vitorpamplona.quartz.nip09Deletions.DeletionEvent +import com.vitorpamplona.quartz.nip40Expiration.isExpirationBefore +import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.onSubscription +import kotlinx.coroutines.yield + +/** + * Lifecycle state of an [ObservableEventStore.project] flow. + * + * - [Loading] is the initial state before the seed query completes. + * The UI should show a spinner / skeleton here. + * - [Loaded] holds the current deduped, sorted list of slots. Every + * membership change publishes a fresh [Loaded] instance; in-place + * addressable updates do *not* publish (the slots' own flows do). + */ +sealed interface ProjectionState { + data object Loading : ProjectionState + + data class Loaded( + val items: List>, + ) : ProjectionState +} + +/** + * State machine that maintains a reactive view over an + * [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. + * + * Each visible event is wrapped in a [MutableStateFlow], giving the + * UI three change granularities: + * + * - **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 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 + * `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. + * + * 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, + private val filters: List, + private val nowProvider: () -> Long = TimeUtils::now, +) { + /** Slots keyed by the *current* event id. Re-keyed when a replaceable / addressable handle takes a new version. */ + private val byId = HashMap>() + + /** Slots keyed by replaceable / addressable [Address] for in-place updates and supersession lookups. */ + private val byAddress = HashMap>() + + /** + * Per-filter capped sets. Each filter independently retains at most + * `filter.limit` matches; a slot is "live" iff it appears in at + * least one of these sets. Identity-keyed — [Filter] is `@Stable` + * but not a data class. + */ + private val perFilter: Map>> = + filters.associateWith { sortedSetOf(slotComparator()) } + + /** + * Run the seed query against the store and populate the indexes. + * Call once, before the first [apply]. Expired events are skipped + * (the store doesn't filter them at query time, so we do here). + */ + suspend fun seed() { + val now = nowProvider() + for (event in store.query(filters)) { + if (event.isExpirationBefore(now)) continue + applyInsert(event) + yield() + } + } + + /** + * Apply a [StoreChange] from [ObservableEventStore.changes]. + * Returns true if the change altered membership (caller should + * publish a fresh [snapshot]); false for in-place addressable + * updates, NIP-01 tiebreaker rejections, and arrivals that no + * filter matches. + */ + fun apply(storeEvent: StoreChange): Boolean = + when (storeEvent) { + is StoreChange.Insert -> { + applyInsert(storeEvent.event) + } + + is StoreChange.DeleteByFilter -> { + dropWhere { ev -> storeEvent.filters.any { it.match(ev) } } + } + + is StoreChange.DeleteExpired -> { + // Store's sweep uses strict `<`; isExpirationBefore is `<=`, so subtract 1. + val cutoff = (storeEvent.asOf ?: nowProvider()) - 1 + dropWhere { it.isExpirationBefore(cutoff) } + } + } + + /** + * Current snapshot as a [ProjectionState.Loaded]. The + * single-filter case skips the union pass — that filter's set + * is already sorted. Multi-filter projections dedup + merge + * via a transient TreeSet. + */ + fun snapshot(): ProjectionState.Loaded { + if (byId.isEmpty()) return ProjectionState.Loaded(emptyList()) + if (perFilter.size == 1) { + return ProjectionState.Loaded(perFilter.values.first().map { it.flow }) + } + val union = sortedSetOf(slotComparator()) + for (set in perFilter.values) union.addAll(set) + return ProjectionState.Loaded(union.map { it.flow }) + } + + private fun applyInsert(event: Event): Boolean { + if (event.isExpirationBefore(nowProvider())) return false + + var changed = false + + // NIP-09 / NIP-62 side effects come first — a deletion event + // that arrives at the same instant as a matching event still + // removes its targets. + if (event is DeletionEvent) { + if (handleDeletion(event)) changed = true + } + if (event is RequestToVanishEvent && event.shouldVanishFrom(store.relay)) { + if (dropWhere { ev -> ownerOf(ev) == event.pubKey && ev.createdAt < event.createdAt }) changed = true + } + + if (handleInsert(event)) changed = true + + return changed + } + + /** + * Returns true if processing the event caused the projection's + * membership to change. Returns false for in-place supersession + * updates, NIP-01 tiebreaker rejections, and arrivals that no + * filter matches. + */ + @Suppress("UNCHECKED_CAST") + private fun handleInsert(event: Event): Boolean { + val address = addressOf(event) + + if (address != null) { + val existing = byAddress[address] + if (existing != null) { + if (!supersedes(event, existing.flow.value)) return false + + // 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 + + 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 + } + + // 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 changed = false + val slot = Slot(event as T) + for ((f, set) in perFilter) { + if (!f.match(event)) continue + if (admit(slot, f, set)) changed = true + } + if (perFilter.values.any { it.contains(slot) }) { + byId[event.id] = slot + if (address != null) byAddress[address] = slot + changed = true + } + 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 { + var changed = false + + // NIP-09 by id, only if the deletion's author owns the target. + // For GiftWrap, the owner is the p-tag recipient. + for (id in deletion.deleteEventIds()) { + val slot = byId[id] ?: continue + if (ownerOf(slot.flow.value) == deletion.pubKey && removeSlot(slot)) changed = true + } + + // NIP-09 by address, only original author, only events with + // `created_at <= deletion.created_at`. `addr` is already an + // [Address] — same equals/hashCode as our index keys. + for (addr in deletion.deleteAddresses()) { + if (addr.pubKeyHex != deletion.pubKey) continue + val slot = byAddress[addr] ?: continue + if (slot.flow.value.createdAt <= deletion.createdAt && removeSlot(slot)) changed = true + } + + return changed + } + + /** Drop every live slot whose current event matches [predicate]. */ + private inline fun dropWhere(predicate: (Event) -> Boolean): Boolean { + // Snapshot before mutating — removeSlot mutates byId. + val targets = byId.values.filter { predicate(it.flow.value) } + var changed = false + for (slot in targets) { + if (removeSlot(slot)) changed = true + } + return changed + } + + /** Remove a live slot from every index AND from each per-filter set. */ + private fun removeSlot(slot: Slot): Boolean { + for (set in perFilter.values) set.remove(slot) + return removeIndexes(slot) + } + + /** + * Remove a slot from [byId] / [byAddress] without touching the + * per-filter sets. Used by the per-filter eviction loop, which + * already owns that bookkeeping. + */ + private fun removeIndexes(slot: Slot): Boolean { + val removed = byId.remove(slot.flow.value.id) != null + if (!removed) return false + slot.address?.let(byAddress::remove) + return true + } + + /** + * Wraps an event with the indexes the projection needs. Sort key + * and address are frozen at construction — in-place supersession + * updates rewrite `flow.value` but the slot's position inside + * each [perFilter] set is stable across updates. + */ + private class Slot( + initial: T, + ) { + val sortCreatedAt: Long = initial.createdAt + val sortId: HexKey = initial.id + val address: Address? = addressOf(initial) + val flow: MutableStateFlow = MutableStateFlow(initial) + } + + companion object { + /** [Address] for replaceable / addressable supersession; `null` for regular events. */ + private fun addressOf(event: Event): Address? = + when { + event is AddressableEvent -> event.address() + event.kind.isReplaceable() -> Address(event.kind, event.pubKey, "") + else -> null + } + + /** + * NIP-01 supersession tiebreaker. The new event wins iff its + * `created_at` is strictly greater, or the timestamps tie and + * its `id` is lexically smaller. + */ + private fun supersedes( + new: Event, + existing: Event, + ): Boolean = + when { + new.createdAt > existing.createdAt -> true + new.createdAt < existing.createdAt -> false + else -> new.id < existing.id + } + + /** + * Owner pubkey for ownership checks (NIP-09 author match, + * NIP-62 vanish target). For GiftWrap the owner is the p-tag + * recipient; for everything else it's `event.pubKey`. + */ + private fun ownerOf(event: Event): HexKey = (event as? GiftWrapEvent)?.recipientPubKey() ?: event.pubKey + + /** + * created_at DESC, id ASC. Sort keys are frozen at slot + * construction, so a slot's position never changes after it + * joins a set. Distinct events have distinct ids, so no third + * tiebreak is needed. + * + * Singleton — stored as `Comparator>` and cast at the + * use site, since the comparator only reads `sortCreatedAt` + * and `sortId` which don't depend on `T`. + */ + private val SLOT_COMPARATOR: Comparator> = + Comparator { a, b -> + if (a === b) return@Comparator 0 + val byTime = b.sortCreatedAt.compareTo(a.sortCreatedAt) + if (byTime != 0) return@Comparator byTime + a.sortId.compareTo(b.sortId) + } + + @Suppress("UNCHECKED_CAST") + private fun slotComparator(): Comparator> = SLOT_COMPARATOR as Comparator> + } +} + +/** + * Open a cold reactive projection over this observable store for the + * given [filters]. The returned flow: + * + * - emits [ProjectionState.Loading] on subscription; + * - runs the seed query against the store; + * - emits [ProjectionState.Loaded] with the seeded list; + * - then emits a fresh [ProjectionState.Loaded] every time membership + * changes (in-place addressable updates do not re-emit — collectors + * of the slot's own [MutableStateFlow] handle those). + * + * Each collection allocates its own state machine, runs its own seed, + * and unsubscribes when the collector cancels — there's no + * `close()`. For multiple collectors on the same view, share with + * `stateIn(scope, SharingStarted.WhileSubscribed(...), Loading)` (the + * standard ViewModel pattern). + * + * NIP-62 vanish handling is scoped by the inner store's `relay`. + */ +fun ObservableEventStore.project(filters: List): Flow> = + flow { + val projection = EventStoreProjection(this@project, filters) + // `onSubscription` runs after the SharedFlow subscription is + // active but before events are pulled, so emissions during + // seed land in the buffer and we drain them via `apply` when + // collect proceeds. `outer` bridges the StoreChange-typed + // collectors back to our ProjectionState output. + val outer = this + emit(ProjectionState.Loading) + changes + .onSubscription { + projection.seed() + outer.emit(projection.snapshot()) + }.collect { change -> + if (projection.apply(change)) outer.emit(projection.snapshot()) + } + } + +fun ObservableEventStore.project(filter: Filter): Flow> = project(listOf(filter)) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt index e70773c29..d376b8e7d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/IEventStore.kt @@ -22,8 +22,17 @@ package com.vitorpamplona.quartz.nip01Core.store import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl interface IEventStore : AutoCloseable { + /** + * Relay URL this store is acting on behalf of, or `null` for an + * unscoped store. Used by NIP-62 right-to-vanish handling: only + * vanish requests whose `relays` list contains this URL (or + * `ALL_RELAYS`) cascade. + */ + val relay: NormalizedRelayUrl? + suspend fun insert(event: Event) interface ITransaction { 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 new file mode 100644 index 000000000..1def6bf4f --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/ObservableEventStore.kt @@ -0,0 +1,166 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.isEphemeral +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip40Expiration.isExpired +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow + +/** + * Reactive façade over any [IEventStore]. Publishes a [StoreChange] + * on [changes] for every accepted mutation so projections can stay in + * sync 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, +) : IEventStore { + override val relay: NormalizedRelayUrl? get() = inner.relay + + private val _changes = + MutableSharedFlow( + replay = 0, + extraBufferCapacity = 256, + onBufferOverflow = BufferOverflow.SUSPEND, + ) + + /** Stream of mutations accepted by this layer. See [StoreChange] for the cases. */ + val changes: SharedFlow = _changes.asSharedFlow() + + /** + * 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( + val event: Event, + ) : StoreChange + + data class DeleteByFilter( + val filters: List, + ) : StoreChange + + data class DeleteExpired( + val asOf: Long? = null, + ) : StoreChange + } + + override suspend fun insert(event: Event) { + if (event.kind.isEphemeral()) { + // Ephemeral kinds bypass persistence. We still drop ones + // that are already expired — they were never going to live + // long enough for a UI to render them. + if (event.isExpired()) return + _changes.emit(StoreChange.Insert(event)) + return + } + // Non-ephemeral: let the inner store enforce expiration, + // tombstones, supersession, etc. If it throws, we never emit. + inner.insert(event) + _changes.emit(StoreChange.Insert(event)) + } + + override suspend fun transaction(body: IEventStore.ITransaction.() -> Unit) { + val accepted = ArrayList() + inner.transaction { + val innerTxn = this + val wrapped = + object : IEventStore.ITransaction { + override fun insert(event: Event) { + if (event.kind.isEphemeral()) { + // Same ephemeral handling as the single-event + // path. The inner txn never sees these so the + // store's batch contains only persistable + // events. + if (event.isExpired()) return + accepted.add(event) + return + } + innerTxn.insert(event) + accepted.add(event) + } + } + wrapped.body() + } + // Emit only after the inner transaction commits. If it throws + // / rolls back, `accepted` is discarded. + for (e in accepted) _changes.emit(StoreChange.Insert(e)) + } + + override suspend fun query(filter: Filter): List = inner.query(filter) + + override suspend fun query(filters: List): List = inner.query(filters) + + override suspend fun query( + filter: Filter, + onEach: (T) -> Unit, + ) = inner.query(filter, onEach) + + override suspend fun query( + filters: List, + onEach: (T) -> Unit, + ) = inner.query(filters, onEach) + + override suspend fun count(filter: Filter): Int = inner.count(filter) + + override suspend fun count(filters: List): Int = inner.count(filters) + + override suspend fun delete(filter: Filter) { + inner.delete(filter) + _changes.emit(StoreChange.DeleteByFilter(listOf(filter))) + } + + override suspend fun delete(filters: List) { + inner.delete(filters) + _changes.emit(StoreChange.DeleteByFilter(filters)) + } + + 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() + _changes.emit(StoreChange.DeleteExpired(asOf)) + } + + override fun close() = inner.close() +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt index 7fb287555..ef9510611 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/EventStore.kt @@ -27,9 +27,14 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.store.IEventStore +/** + * SQLite-backed [IEventStore] with default DB-file name and relay + * scoping. Wrap in `ObservableEventStore` if you want to feed + * `EventStoreProjection`. + */ class EventStore( dbName: String? = "events.db", - relay: NormalizedRelayUrl? = "wss://quartz.local/".normalizeRelayUrl(), + override val relay: NormalizedRelayUrl? = "wss://quartz.local/".normalizeRelayUrl(), val indexStrategy: IndexingStrategy = DefaultIndexingStrategy(), ) : IEventStore { val store = SQLiteEventStore(BundledSQLiteDriver(), dbName, relay, indexStrategy) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt index fa6ad1ffe..6a3412e52 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt @@ -232,7 +232,7 @@ class QueryBuilder( } } - private fun SQLiteStatement.toEvent() = + private fun SQLiteStatement.toEvent(): T = EventFactory.create( getText(0), getText(1), 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 new file mode 100644 index 000000000..992b43620 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/projection/EventStoreProjectionTest.kt @@ -0,0 +1,593 @@ +/* + * 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.cache.projection + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +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 +import com.vitorpamplona.quartz.nip40Expiration.expiration +import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent +import com.vitorpamplona.quartz.utils.Secp256k1Instance +import com.vitorpamplona.quartz.utils.TimeUtils +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharingStarted +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.stateIn +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotSame +import kotlin.test.assertSame +import kotlin.test.assertTrue + +class EventStoreProjectionTest { + private val signer = NostrSignerSync() + private val otherSigner = NostrSignerSync() + private lateinit var store: EventStore + private lateinit var observable: ObservableEventStore + private lateinit var scope: CoroutineScope + + @BeforeTest + fun setUp() { + Secp256k1Instance + store = EventStore(dbName = null) + observable = ObservableEventStore(store) + scope = CoroutineScope(SupervisorJob()) + } + + @AfterTest + fun tearDown() { + scope.cancel() + store.close() + } + + /** Open a hot StateFlow over the projection for the lifetime of the test scope. */ + private fun projectionOf(filter: Filter): StateFlow> = observable.project(filter).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading) + + private fun projectionOf(filters: List): StateFlow> = observable.project(filters).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading) + + /** Snapshot of the currently-loaded items, or empty if still seeding. */ + private val StateFlow>.items: List> + get() = (value as? ProjectionState.Loaded)?.items.orEmpty() + + /** Suspends until the seed completes; returns the loaded list. */ + private suspend fun StateFlow>.awaitLoaded(timeoutMs: Long = 5_000): List> = + withTimeout(timeoutMs) { + (first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items + } + + private suspend fun StateFlow>.awaitItems( + timeoutMs: Long = 5_000, + predicate: (List>) -> Boolean, + ): List> = + withTimeout(timeoutMs) { + (first { it is ProjectionState.Loaded && predicate(it.items) } as ProjectionState.Loaded).items + } + + private suspend fun awaitFlow( + flow: MutableStateFlow, + timeoutMs: Long = 5_000, + predicate: (T) -> Boolean, + ): T = + withTimeout(timeoutMs) { + flow.first { predicate(it) } + } + + @Test + fun seedReturnsExistingEvents() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + observable.insert(a) + observable.insert(b) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + + val items = projection.items + assertEquals(2, items.size) + assertEquals(b.id, items[0].value.id) + assertEquals(a.id, items[1].value.id) + } + + @Test + fun insertAddsNewSlot() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + observable.insert(a) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + val before = projection.items + assertEquals(1, before.size) + + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + observable.insert(b) + + val after = projection.awaitItems { it.size == 2 } + assertNotSame(before, after, "insert must produce a new list reference") + assertEquals(b.id, after[0].value.id) + } + + @Test + fun nonMatchingInsertDoesNotChangeList() = + runBlocking { + val text = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + observable.insert(text) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + val seed = projection.items + + val meta = signer.sign(MetadataEvent.createNew("Vitor", createdAt = 200)) + observable.insert(meta) + + delay(150) + assertSame(seed, projection.items) + } + + @Test + fun replaceableUpdateMutatesSlotInPlace() = + runBlocking { + val time = TimeUtils.now() + val v1 = signer.sign(MetadataEvent.createNew("v1", createdAt = time)) + observable.insert(v1) + + val projection = + projectionOf(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey))) + projection.awaitLoaded() + val seedList = projection.items + assertEquals(1, seedList.size) + val slot = seedList[0] + assertEquals(v1.id, slot.value.id) + + val v2 = signer.sign(MetadataEvent.createNew("v2", createdAt = time + 1)) + observable.insert(v2) + + awaitFlow(slot) { it.id == v2.id } + assertSame(seedList, projection.items, "replaceable update must not change list reference") + assertSame(slot, projection.items[0]) + } + + @Test + fun addressableUpdateMutatesSlotInPlace() = + runBlocking { + val time = TimeUtils.now() + val v1 = signer.sign(LongTextNoteEvent.build("blog v1", "title", dTag = "blog", createdAt = time)) + observable.insert(v1) + + val projection = + projectionOf( + Filter( + kinds = listOf(LongTextNoteEvent.KIND), + authors = listOf(v1.pubKey), + tags = mapOf("d" to listOf("blog")), + ), + ) + projection.awaitLoaded() + val seedList = projection.items + val slot = seedList[0] + + val v2 = signer.sign(LongTextNoteEvent.build("blog v2", "title", dTag = "blog", createdAt = time + 1)) + observable.insert(v2) + + awaitFlow(slot) { it.id == v2.id } + 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 + * must keep v2 in place and not regress to v1 — that's the NIP-01 + * supersession contract from the projection's side, since the + * store would have rejected v1 anyway. + */ + @Test + fun olderReplaceableArrivingAfterNewerIsRejected() = + runBlocking { + val time = TimeUtils.now() + val v1 = signer.sign(MetadataEvent.createNew("v1", createdAt = time)) + val v2 = signer.sign(MetadataEvent.createNew("v2", createdAt = time + 5)) + + // Seed the projection with v2 before v1 even hits the store + // — by inserting v2 first. + observable.insert(v2) + + val projection = + projectionOf(Filter(kinds = listOf(MetadataEvent.KIND), authors = listOf(v1.pubKey))) + projection.awaitLoaded() + val slot = projection.items[0] + assertEquals(v2.id, slot.value.id) + + // The store rejects v1 because v2 already won; the + // projection therefore never sees v1 on the inserts + // stream. The slot must still hold v2. + try { + observable.insert(v1) + } catch (_: Throwable) { + // expected — store enforces the same rule + } + + delay(150) + assertEquals(v2.id, slot.value.id) + } + + @Test + fun nip09DeletionRemovesSlot() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + observable.insert(a) + observable.insert(b) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + assertEquals(2, projection.items.size) + + val deletion = signer.sign(DeletionEvent.build(listOf(a))) + observable.insert(deletion) + + val after = projection.awaitItems { it.size == 1 } + assertEquals(b.id, after[0].value.id) + } + + /** + * NIP-09 cross-author deletions are inert. A different signer + * publishing a kind-5 targeting `a` must not drop the slot. + */ + @Test + fun nip09CrossAuthorDeletionIsInert() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + observable.insert(a) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + val seed = projection.items + assertEquals(1, seed.size) + + val foreignDeletion = otherSigner.sign(DeletionEvent.build(listOf(a))) + observable.insert(foreignDeletion) + + // Give the projection time to process the event. + delay(150) + assertSame(seed, projection.items) + assertEquals( + a.id, + projection.items[0] + .value.id, + ) + } + + @Test + fun nip62VanishRemovesAuthorEvents() = + runBlocking { + val time = TimeUtils.now() + val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = time + 1)) + observable.insert(a) + observable.insert(b) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + assertEquals(2, projection.items.size) + + val vanish = + signer.sign( + RequestToVanishEvent.build( + "wss://quartz.local".normalizeRelayUrl(), + createdAt = time + 2, + ), + ) + observable.insert(vanish) + + val after = projection.awaitItems { it.isEmpty() } + assertTrue(after.isEmpty()) + } + + /** + * NIP-62 only removes events from the same author. A vanish from + * a different author must not touch slots owned by [signer]. + */ + @Test + fun nip62OtherAuthorVanishLeavesEventsAlone() = + runBlocking { + val time = TimeUtils.now() + val a = signer.sign(TextNoteEvent.build("a", createdAt = time)) + observable.insert(a) + + val projection = projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + val seed = projection.items + + val foreignVanish = + otherSigner.sign( + RequestToVanishEvent.build( + "wss://quartz.local".normalizeRelayUrl(), + createdAt = time + 2, + ), + ) + observable.insert(foreignVanish) + + delay(150) + assertSame(seed, projection.items) + } + + /** + * NIP-40 expiration drops slots only when the application calls + * `deleteExpiredEvents()` on the observable store — projections + * no longer run their own ticker. + */ + @Test + fun nip40ExpirationDroppedOnStoreSweep() = + runBlocking { + val time = TimeUtils.now() + val safe = signer.sign(TextNoteEvent.build("safe", createdAt = time) { expiration(time + 100) }) + val short = signer.sign(TextNoteEvent.build("short", createdAt = time) { expiration(time + 1) }) + observable.insert(safe) + observable.insert(short) + + val projection = + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + assertEquals(2, projection.items.size) + + // Let the short expiration lapse, then ask the store to + // sweep — the projection drops the expired slot in + // response to the resulting StoreChange.Delete(Expired). + delay(2000) + observable.deleteExpiredEvents() + + val after = projection.awaitItems { it.size == 1 } + assertEquals(safe.id, after[0].value.id) + } + + /** + * `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 = + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND))) + projection.awaitLoaded() + assertEquals(3, projection.items.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) + } + + @Test + fun limitIsEnforcedOnInsertOverflow() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + val b = signer.sign(TextNoteEvent.build("b", createdAt = 200)) + observable.insert(a) + observable.insert(b) + + val projection = + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2)) + projection.awaitLoaded() + assertEquals(2, projection.items.size) + + val c = signer.sign(TextNoteEvent.build("c", createdAt = 300)) + observable.insert(c) + + val after = projection.awaitItems { it[0].value.id == c.id } + assertEquals(2, after.size) + assertEquals(c.id, after[0].value.id) + assertEquals(b.id, after[1].value.id) + } + + /** + * Per-filter limit: filter A caps at 2, filter B caps at 2, but + * the events they match are disjoint, so the projection's union + * is 4 (larger than any single filter's cap). + */ + @Test + fun perFilterLimitUnionExceedsSingleLimit() = + runBlocking { + val authorA = NostrSignerSync() + val authorB = NostrSignerSync() + val a1 = authorA.sign(TextNoteEvent.build("a1", createdAt = 100)) + val a2 = authorA.sign(TextNoteEvent.build("a2", createdAt = 200)) + val b1 = authorB.sign(TextNoteEvent.build("b1", createdAt = 110)) + val b2 = authorB.sign(TextNoteEvent.build("b2", createdAt = 210)) + observable.insert(a1) + observable.insert(a2) + observable.insert(b1) + observable.insert(b2) + + val filterA = Filter(kinds = listOf(TextNoteEvent.KIND), authors = listOf(authorA.pubKey), limit = 2) + val filterB = Filter(kinds = listOf(TextNoteEvent.KIND), authors = listOf(authorB.pubKey), limit = 2) + val projection = + projectionOf( + listOf(filterA, filterB), + ) + projection.awaitLoaded() + assertEquals(4, projection.items.size, "per-filter caps don't dedupe union") + } + + /** + * Each filter's cap is enforced on its own retained set: filter A + * keeps only its 2 newest matches even when more arrive. + */ + @Test + fun perFilterLimitEvictsOldestWithinFilter() = + runBlocking { + val a1 = signer.sign(TextNoteEvent.build("a1", createdAt = 100)) + val a2 = signer.sign(TextNoteEvent.build("a2", createdAt = 200)) + val a3 = signer.sign(TextNoteEvent.build("a3", createdAt = 300)) + observable.insert(a1) + observable.insert(a2) + + val projection = + projectionOf(Filter(kinds = listOf(TextNoteEvent.KIND), limit = 2)) + projection.awaitLoaded() + assertEquals(2, projection.items.size) + + observable.insert(a3) + val after = projection.awaitItems { it[0].value.id == a3.id } + assertEquals(2, after.size) + assertEquals(a3.id, after[0].value.id) + assertEquals(a2.id, after[1].value.id) + } + + @Test + fun cancellingScopeStopsListening() = + runBlocking { + val a = signer.sign(TextNoteEvent.build("a", createdAt = 100)) + observable.insert(a) + + // Sub-scope so we can cancel just the projection's collector + // without taking down the test's outer scope. + val collectorScope = CoroutineScope(SupervisorJob()) + val projection = + observable + .project(Filter(kinds = listOf(TextNoteEvent.KIND))) + .stateIn(collectorScope, SharingStarted.Eagerly, ProjectionState.Loading) + projection.awaitLoaded() + assertEquals(1, projection.items.size) + + collectorScope.cancel() + delay(50) + + // After the collector scope dies, new inserts don't update + // the StateFlow — its value is frozen at the last emission. + observable.insert(signer.sign(TextNoteEvent.build("b", createdAt = 200))) + delay(150) + assertEquals(1, projection.items.size) + assertEquals(a.id, projection.items[0].value.id) + } + + /** + * Ephemeral events (kinds `20000-29999`) are never persisted, so + * the inner SQLite store silently drops them. The + * `ObservableEventStore` wrapper still routes them onto its + * [events][com.vitorpamplona.quartz.nip01Core.store.projection.ObservableEventStore.changes] + * flow, so an open projection sees them while it's alive. They + * vanish from any future seed because the DB never had them. + */ + @Test + fun ephemeralEventsAppearInProjection() = + runBlocking { + val ephemeralKind = 22_000 + val projection = + projectionOf(Filter(kinds = listOf(ephemeralKind))) + projection.awaitLoaded() + assertTrue(projection.items.isEmpty()) + + val ephemeral: Event = + signer.sign( + TimeUtils.now(), + ephemeralKind, + arrayOf(emptyArray()), + "live", + ) + observable.insert(ephemeral) + + val after = projection.awaitItems { it.size == 1 } + assertEquals(ephemeral.id, after[0].value.id) + + // Confirmation that the inner SQLite store didn't persist. + val persisted = store.query(Filter(kinds = listOf(ephemeralKind))) + assertEquals(0, persisted.size) + + // A fresh projection on the same store gets nothing — the + // event was only ever live, not durable. + val freshProjection = + projectionOf(Filter(kinds = listOf(ephemeralKind))) + freshProjection.awaitLoaded() + assertTrue(freshProjection.items.isEmpty()) + } +} diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/AssertUtils.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/AssertUtils.kt index d7fa94ce5..57215924c 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/AssertUtils.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/AssertUtils.kt @@ -22,9 +22,10 @@ package com.vitorpamplona.quartz.nip01Core.store.sqlite import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.store.IEventStore import kotlin.test.assertEquals -suspend fun EventStore.assertQuery( +suspend fun IEventStore.assertQuery( expected: T?, filter: Filter, ) { @@ -40,7 +41,7 @@ suspend fun EventStore.assertQuery( } } -suspend fun EventStore.assertQuery( +suspend fun IEventStore.assertQuery( expected: List, filter: Filter, ) { diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.jvmAndroid.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.jvmAndroid.kt new file mode 100644 index 000000000..d393dd2c1 --- /dev/null +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.jvmAndroid.kt @@ -0,0 +1,78 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import java.lang.ref.WeakReference +import java.util.concurrent.ConcurrentHashMap + +/** + * JVM/Android actual: ConcurrentHashMap of WeakReference, pre-sized + * for the expected hot working set. Dead entries are reclaimed + * lazily on access (see [intern] and [get]) — when a weak ref's + * referent has been GC'd, the map slot is removed atomically and + * replaced if a fresh event takes its place. + */ +actual class EventInterner { + private val cache = ConcurrentHashMap>(INITIAL_CAPACITY, LOAD_FACTOR) + + actual fun intern(event: Event): Event { + // Fast path: existing canonical is still alive. + cache[event.id]?.get()?.let { return it } + + // Race-safe install: putIfAbsent ensures only one writer + // wins. If another thread beat us, return whatever they put + // (resolving the rare case where their ref was already GC'd + // by retrying through the slow path). + while (true) { + val ref = WeakReference(event) + val existing = cache.putIfAbsent(event.id, ref) + if (existing == null) return event + val canonical = existing.get() + if (canonical != null) return canonical + // Existing entry's referent was GC'd; drop it and retry. + cache.remove(event.id, existing) + } + } + + actual fun get(id: HexKey): Event? { + val ref = cache[id] ?: return null + val event = ref.get() + if (event != null) return event + // Self-cleaning: drop the dead entry so the map doesn't bloat. + cache.remove(id, ref) + return null + } + + actual fun size(): Int = cache.size + + actual fun clear() { + cache.clear() + } + + actual companion object { + actual val Default: EventInterner = EventInterner() + + private const val INITIAL_CAPACITY = 5_000 + private const val LOAD_FACTOR = 0.75f + } +} diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt index 8da5d14b5..c91cf7445 100644 --- a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt +++ b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt @@ -66,7 +66,7 @@ open class FsEventStore( * `null`, only "ALL_RELAYS" vanish requests cascade — matches * `SQLiteEventStore`'s relay arg semantics. */ - private val relay: NormalizedRelayUrl? = null, + override val relay: NormalizedRelayUrl? = null, /** * How to render an event to JSON before writing the canonical file. * Default is the compact NIP-01 form ([Event.toJson]); CLIs that diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInternerTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInternerTest.kt new file mode 100644 index 000000000..efde104d8 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInternerTest.kt @@ -0,0 +1,138 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import com.vitorpamplona.quartz.utils.Secp256k1Instance +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertSame +import kotlin.test.assertTrue + +class EventInternerTest { + private val signer = NostrSignerSync() + private lateinit var interner: EventInterner + + @BeforeTest + fun setUp() { + Secp256k1Instance + interner = EventInterner() + } + + @AfterTest + fun tearDown() { + interner.clear() + } + + @Test + fun internReturnsFirstInstance() { + val original = signer.sign(TextNoteEvent.build("hello", createdAt = 100)) + val canonical = interner.intern(original) + assertSame(original, canonical) + assertEquals(1, interner.size()) + } + + @Test + fun internCollapsesDuplicates() { + val a = signer.sign(TextNoteEvent.build("hello", createdAt = 100)) + // Round-trip through OptimizedJsonMapper directly (NOT + // Event.fromJson, which would already intern via Default) so + // we get a non-identical-but-equal event for our isolated + // interner to deduplicate. + val b = OptimizedJsonMapper.fromJson(a.toJson()) as TextNoteEvent + assertEquals(a.id, b.id) + + val firstCanonical = interner.intern(a) + val secondCanonical = interner.intern(b) + assertSame(firstCanonical, secondCanonical) + assertSame(a, secondCanonical) + assertEquals(1, interner.size()) + } + + @Test + fun getReturnsLiveEntry() { + val a = signer.sign(TextNoteEvent.build("hello", createdAt = 100)) + interner.intern(a) + assertSame(a, interner.get(a.id)) + } + + @Test + fun getReturnsNullForUnknownId() { + assertNull(interner.get("0".repeat(64))) + } + + @Test + fun getEvictsDeadEntries() { + // Confine the event to a helper so no strong ref leaks into + // this method's stack frame after it returns. The interner's + // WeakReference is then the only thing holding it. + val id = internAndDiscard(interner) + for (i in 0..40) { + System.gc() + Thread.sleep(20) + if (interner.get(id) == null) break + } + assertNull(interner.get(id)) + assertEquals(0, interner.size(), "dead entry should be evicted on access") + } + + private fun internAndDiscard(interner: EventInterner): HexKey { + val ev = signer.sign(TextNoteEvent.build("hello", createdAt = 100)) + interner.intern(ev) + return ev.id + } + + @Test + fun draftChurnDoesNotLeak() { + // 100 distinct drafts, no strong references retained. + val ids = ArrayList(100) + for (i in 0 until 100) { + val e = signer.sign(TextNoteEvent.build("draft-$i", createdAt = 1_000L + i)) + ids.add(e.id) + interner.intern(e) + } + // Force GC + sweep — calling get() on every id triggers + // the on-access cleanup path. + for (i in 0..40) { + System.gc() + Thread.sleep(20) + ids.forEach { interner.get(it) } + if (interner.size() == 0) break + } + assertTrue(interner.size() <= 5, "expected most drafts to be evicted, got ${interner.size()}") + } + + @Test + fun defaultInstanceIsShared() { + val a = signer.sign(TextNoteEvent.build("hello", createdAt = 100)) + val canonical = EventInterner.Default.intern(a) + // A second intern call from anywhere returns the same canonical. + val b = EventInterner.Default.intern(a) + assertSame(canonical, b) + EventInterner.Default.clear() + } +} diff --git a/quartz/src/linuxMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.linux.kt b/quartz/src/linuxMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.linux.kt new file mode 100644 index 000000000..8c6e6223b --- /dev/null +++ b/quartz/src/linuxMain/kotlin/com/vitorpamplona/quartz/nip01Core/cache/interning/EventInterner.linux.kt @@ -0,0 +1,39 @@ +/* + * 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.cache.interning + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** Linux native actual: passthrough. See `EventInterner.apple.kt`. */ +actual class EventInterner { + actual fun intern(event: Event): Event = event + + actual fun get(id: HexKey): Event? = null + + actual fun size(): Int = 0 + + actual fun clear() {} + + actual companion object { + actual val Default: EventInterner = EventInterner() + } +}