Merge pull request #2664 from vitorpamplona/claude/quartz-eventstore-observer-o2f1O
Add event projection and interning for reactive event store queries
This commit is contained in:
Vitor Pamplona
+45
@@ -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()
|
||||
}
|
||||
}
|
||||
Vendored
+58
@@ -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
|
||||
}
|
||||
}
|
||||
+121
@@ -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<Event>()
|
||||
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 <T : Event> query(filter: Filter): List<T> = inner.query<T>(filter).map { interner.intern(it) as T }
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
override suspend fun <T : Event> query(filters: List<Filter>): List<T> = inner.query<T>(filters).map { interner.intern(it) as T }
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
override suspend fun <T : Event> query(
|
||||
filter: Filter,
|
||||
onEach: (T) -> Unit,
|
||||
) = inner.query<T>(filter) { onEach(interner.intern(it) as T) }
|
||||
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
override suspend fun <T : Event> query(
|
||||
filters: List<Filter>,
|
||||
onEach: (T) -> Unit,
|
||||
) = inner.query<T>(filters) { onEach(interner.intern(it) as T) }
|
||||
|
||||
override suspend fun count(filter: Filter): Int = inner.count(filter)
|
||||
|
||||
override suspend fun count(filters: List<Filter>): Int = inner.count(filters)
|
||||
|
||||
override suspend fun delete(filter: Filter) = inner.delete(filter)
|
||||
|
||||
override suspend fun delete(filters: List<Filter>) = inner.delete(filters)
|
||||
|
||||
override suspend fun deleteExpiredEvents() = inner.deleteExpiredEvents()
|
||||
|
||||
override fun close() = inner.close()
|
||||
}
|
||||
+436
@@ -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<out T : Event> {
|
||||
data object Loading : ProjectionState<Nothing>
|
||||
|
||||
data class Loaded<T : Event>(
|
||||
val items: List<MutableStateFlow<T>>,
|
||||
) : ProjectionState<T>
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<T : Event>(
|
||||
private val store: ObservableEventStore,
|
||||
private val filters: List<Filter>,
|
||||
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<HexKey, Slot<T>>()
|
||||
|
||||
/** Slots keyed by replaceable / addressable [Address] for in-place updates and supersession lookups. */
|
||||
private val byAddress = HashMap<Address, Slot<T>>()
|
||||
|
||||
/**
|
||||
* 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<Filter, java.util.SortedSet<Slot<T>>> =
|
||||
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<T>(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<T> {
|
||||
if (byId.isEmpty()) return ProjectionState.Loaded(emptyList())
|
||||
if (perFilter.size == 1) {
|
||||
return ProjectionState.Loaded(perFilter.values.first().map { it.flow })
|
||||
}
|
||||
val union = sortedSetOf(slotComparator<T>())
|
||||
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<T>,
|
||||
f: Filter,
|
||||
set: java.util.SortedSet<Slot<T>>,
|
||||
): 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<T>): 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<T>): 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<T : Event>(
|
||||
initial: T,
|
||||
) {
|
||||
val sortCreatedAt: Long = initial.createdAt
|
||||
val sortId: HexKey = initial.id
|
||||
val address: Address? = addressOf(initial)
|
||||
val flow: MutableStateFlow<T> = 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<Slot<*>>` 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<Slot<*>> =
|
||||
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 <T : Event> slotComparator(): Comparator<Slot<T>> = SLOT_COMPARATOR as Comparator<Slot<T>>
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<ProjectionState<T>> =
|
||||
flow {
|
||||
val projection = EventStoreProjection<T>(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 <T : Event> ObservableEventStore.project(filter: Filter): Flow<ProjectionState<T>> = project(listOf(filter))
|
||||
@@ -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 {
|
||||
|
||||
+166
@@ -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<StoreChange>(
|
||||
replay = 0,
|
||||
extraBufferCapacity = 256,
|
||||
onBufferOverflow = BufferOverflow.SUSPEND,
|
||||
)
|
||||
|
||||
/** Stream of mutations accepted by this layer. See [StoreChange] for the cases. */
|
||||
val changes: SharedFlow<StoreChange> = _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<Filter>,
|
||||
) : 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<Event>()
|
||||
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 <T : Event> query(filter: Filter): List<T> = inner.query(filter)
|
||||
|
||||
override suspend fun <T : Event> query(filters: List<Filter>): List<T> = inner.query(filters)
|
||||
|
||||
override suspend fun <T : Event> query(
|
||||
filter: Filter,
|
||||
onEach: (T) -> Unit,
|
||||
) = inner.query(filter, onEach)
|
||||
|
||||
override suspend fun <T : Event> query(
|
||||
filters: List<Filter>,
|
||||
onEach: (T) -> Unit,
|
||||
) = inner.query(filters, onEach)
|
||||
|
||||
override suspend fun count(filter: Filter): Int = inner.count(filter)
|
||||
|
||||
override suspend fun count(filters: List<Filter>): 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<Filter>) {
|
||||
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()
|
||||
}
|
||||
+6
-1
@@ -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)
|
||||
|
||||
+1
-1
@@ -232,7 +232,7 @@ class QueryBuilder(
|
||||
}
|
||||
}
|
||||
|
||||
private fun <T : Event> SQLiteStatement.toEvent() =
|
||||
private fun <T : Event> SQLiteStatement.toEvent(): T =
|
||||
EventFactory.create<T>(
|
||||
getText(0),
|
||||
getText(1),
|
||||
|
||||
+593
@@ -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 <T : Event> projectionOf(filter: Filter): StateFlow<ProjectionState<T>> = observable.project<T>(filter).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading)
|
||||
|
||||
private fun <T : Event> projectionOf(filters: List<Filter>): StateFlow<ProjectionState<T>> = observable.project<T>(filters).stateIn(scope, SharingStarted.Eagerly, ProjectionState.Loading)
|
||||
|
||||
/** Snapshot of the currently-loaded items, or empty if still seeding. */
|
||||
private val <T : Event> StateFlow<ProjectionState<T>>.items: List<MutableStateFlow<T>>
|
||||
get() = (value as? ProjectionState.Loaded)?.items.orEmpty()
|
||||
|
||||
/** Suspends until the seed completes; returns the loaded list. */
|
||||
private suspend fun <T : Event> StateFlow<ProjectionState<T>>.awaitLoaded(timeoutMs: Long = 5_000): List<MutableStateFlow<T>> =
|
||||
withTimeout(timeoutMs) {
|
||||
(first { it is ProjectionState.Loaded } as ProjectionState.Loaded).items
|
||||
}
|
||||
|
||||
private suspend fun <T : Event> StateFlow<ProjectionState<T>>.awaitItems(
|
||||
timeoutMs: Long = 5_000,
|
||||
predicate: (List<MutableStateFlow<T>>) -> Boolean,
|
||||
): List<MutableStateFlow<T>> =
|
||||
withTimeout(timeoutMs) {
|
||||
(first { it is ProjectionState.Loaded && predicate(it.items) } as ProjectionState.Loaded).items
|
||||
}
|
||||
|
||||
private suspend fun <T : Event> awaitFlow(
|
||||
flow: MutableStateFlow<T>,
|
||||
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<TextNoteEvent>(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<TextNoteEvent>(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<TextNoteEvent>(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<MetadataEvent>(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<LongTextNoteEvent>(
|
||||
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<LongTextNoteEvent>(
|
||||
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<MetadataEvent>(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<Event>(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<Event>(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<Event>(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<Event>(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<Event>(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<Event>(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<TextNoteEvent>(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<TextNoteEvent>(
|
||||
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<TextNoteEvent>(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<TextNoteEvent>(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<Event>(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<Event>(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<Event>(Filter(kinds = listOf(ephemeralKind)))
|
||||
freshProjection.awaitLoaded()
|
||||
assertTrue(freshProjection.items.isEmpty())
|
||||
}
|
||||
}
|
||||
+3
-2
@@ -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 <T : Event> EventStore.assertQuery(
|
||||
suspend fun <T : Event> IEventStore.assertQuery(
|
||||
expected: T?,
|
||||
filter: Filter,
|
||||
) {
|
||||
@@ -40,7 +41,7 @@ suspend fun <T : Event> EventStore.assertQuery(
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun <T : Event> EventStore.assertQuery(
|
||||
suspend fun <T : Event> IEventStore.assertQuery(
|
||||
expected: List<T>,
|
||||
filter: Filter,
|
||||
) {
|
||||
|
||||
+78
@@ -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<HexKey, WeakReference<Event>>(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
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -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
|
||||
|
||||
Vendored
+138
@@ -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<HexKey>(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()
|
||||
}
|
||||
}
|
||||
+39
@@ -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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user