feat(cache): Phase 1 — cache-centric architecture foundation

- Add BoundedLargeCache: LargeCache wrapper with size enforcement (50k notes, 25k users, 10k addressable). Lock-free reads via ConcurrentSkipListMap, evicts 10% when cap exceeded. - Switch DesktopLocalCache from ConcurrentHashMap to BoundedLargeCache. Exposes filterIntoSet, values, etc. matching Android's query patterns. - Add consume methods for kinds 1 (TextNote), 7 (Reaction), 9734 (ZapRequest), 9735 (Zap). Uses tagsWithoutCitations() for reply parsing, originalPost()+taggedAddresses() for reactions, zappedPost() for zaps. Follows Android LocalCache pattern. - Add consumeEvent() bridge in coordinator with BasicBundledInsert (250ms batching). Routes relay onEvent callbacks to cache. - Fix SharedFlow: extraBufferCapacity=64, DROP_OLDEST. Prevents blocking emitters and dropped events. - Set -Xmx2g JVM arg for long desktop sessions. - Clear cache on logout (coordinator first, then cache). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-19 13:55:04 +02:00
parent 61119e6916
commit 543f7a329c
5 changed files with 300 additions and 16 deletions
@@ -80,6 +80,8 @@ compose.desktop {
        mainClass = "com.vitorpamplona.amethyst.desktop.MainKt"
        jvmArgs += "--add-opens=java.base/java.nio=ALL-UNNAMED"

+        jvmArgs += "-Xmx2g"
+
        nativeDistributions {
            appResourcesRootDir.set(project.layout.projectDirectory.dir("src/jvmMain/appResources"))
            targetFormats(TargetFormat.Dmg, TargetFormat.Msi, TargetFormat.Deb)
@@ -460,6 +460,14 @@ fun App(
            )
        }

+    // Clear cache and subscriptions on logout
+    LaunchedEffect(accountState) {
+        if (accountState is AccountState.LoggedOut) {
+            subscriptionsCoordinator.clear()
+            localCache.clear()
+        }
+    }
+
    // Try to load saved account on startup
    DisposableEffect(Unit) {
        relayManager.addDefaultRelays()
@@ -0,0 +1,93 @@
+/*
+ * 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.amethyst.desktop.cache
+
+import com.vitorpamplona.quartz.utils.cache.CacheCollectors
+import com.vitorpamplona.quartz.utils.cache.LargeCache
+
+/**
+ * A bounded wrapper around [LargeCache] that enforces a maximum size.
+ *
+ * When the cache exceeds [maxSize], the oldest entries (by key order) are evicted.
+ * Uses [LargeCache] (ConcurrentSkipListMap) for lock-free reads and rich query APIs
+ * (filterIntoSet, mapNotNull, etc.) matching Android's LocalCache patterns.
+ *
+ * Chosen over LruCache because:
+ * - Lock-free reads via ConcurrentSkipListMap (vs synchronized on every get())
+ * - Rich query API matching Android's filter patterns
+ * - No snapshot copy overhead for iteration
+ */
+class BoundedLargeCache<K : Comparable<K>, V>(
+    private val maxSize: Int,
+    private val evictPercent: Float = 0.1f,
+) {
+    private val inner = LargeCache<K, V>()
+
+    fun get(key: K): V? = inner.get(key)
+
+    fun put(
+        key: K,
+        value: V,
+    ) {
+        inner.put(key, value)
+        enforceSize()
+    }
+
+    fun getOrCreate(
+        key: K,
+        builder: (K) -> V,
+    ): V {
+        val result = inner.getOrCreate(key, builder)
+        enforceSize()
+        return result
+    }
+
+    fun remove(key: K): V? = inner.remove(key)
+
+    fun containsKey(key: K): Boolean = inner.containsKey(key)
+
+    fun size(): Int = inner.size()
+
+    fun isEmpty(): Boolean = inner.isEmpty()
+
+    fun clear() = inner.clear()
+
+    fun keys(): Set<K> = inner.keys()
+
+    fun values(): Iterable<V> = inner.values()
+
+    fun filterIntoSet(consumer: CacheCollectors.BiFilter<K, V>): Set<V> = inner.filterIntoSet(consumer)
+
+    fun <R> mapNotNull(consumer: CacheCollectors.BiMapper<K, V, R?>): List<R> = inner.mapNotNull(consumer)
+
+    fun forEach(consumer: java.util.function.BiConsumer<K, V>) = inner.forEach(consumer)
+
+    fun count(consumer: CacheCollectors.BiFilter<K, V>): Int = inner.count(consumer)
+
+    private fun enforceSize() {
+        val currentSize = inner.size()
+        if (currentSize > maxSize) {
+            val toRemove = (maxSize * evictPercent).toInt().coerceAtLeast(1)
+            val keys = inner.keys().take(toRemove)
+            keys.forEach { inner.remove(it) }
+        }
+    }
+}
@@ -32,12 +32,18 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.tags.aTag.taggedAddresses
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
 import com.vitorpamplona.quartz.nip19Bech32.decodePublicKeyAsHexOrNull
+import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
 import com.vitorpamplona.quartz.nip47WalletConnect.events.LnZapPaymentRequestEvent
 import com.vitorpamplona.quartz.nip47WalletConnect.events.LnZapPaymentResponseEvent
+import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent
+import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
 import com.vitorpamplona.quartz.utils.DualCase
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.launch
@@ -50,28 +56,33 @@ import java.util.concurrent.ConcurrentHashMap
 * Supports searching users by name prefix for the search functionality.
 */
 class DesktopLocalCache : ICacheProvider {
-    private val users = ConcurrentHashMap<HexKey, User>()
-    private val notes = ConcurrentHashMap<HexKey, Note>()
-    private val addressableNotes = ConcurrentHashMap<String, AddressableNote>()
+    val users = BoundedLargeCache<HexKey, User>(MAX_USERS)
+    val notes = BoundedLargeCache<HexKey, Note>(MAX_NOTES)
+    val addressableNotes = BoundedLargeCache<String, AddressableNote>(MAX_ADDRESSABLE)
    private val deletedEvents = ConcurrentHashMap.newKeySet<HexKey>()

-    private val eventStream = DesktopCacheEventStream()
+    val eventStream = DesktopCacheEventStream()
+
+    companion object {
+        const val MAX_NOTES = 50_000
+        const val MAX_USERS = 25_000
+        const val MAX_ADDRESSABLE = 10_000
+    }

    val paymentTracker = NwcPaymentTracker()

    // ----- User operations -----

-    override fun getUserIfExists(pubkey: HexKey): User? = users[pubkey]
+    override fun getUserIfExists(pubkey: HexKey): User? = users.get(pubkey)

    override fun getOrCreateUser(pubkey: HexKey): User =
-        users.getOrPut(pubkey) {
-            // Create placeholder notes for relay lists
+        users.getOrCreate(pubkey) {
            val nip65Note = getOrCreateNote("nip65:$pubkey")
            val dmNote = getOrCreateNote("dm:$pubkey")
            User(pubkey, nip65Note, dmNote)
        }

-    override fun countUsers(predicate: (String, User) -> Boolean): Int = users.count { (key, user) -> predicate(key, user) }
+    override fun countUsers(predicate: (String, User) -> Boolean): Int = users.count { key, user -> predicate(key, user) }

    override fun findUsersStartingWith(
        prefix: String,
@@ -92,7 +103,8 @@ class DesktopLocalCache : ICacheProvider {
            )

        // Search by name/displayName/nip05/lud16
-        return users.values
+        return users
+            .values()
            .filter { user ->
                val metadata = user.metadataOrNull()
                if (metadata == null) {
@@ -128,6 +140,134 @@ class DesktopLocalCache : ICacheProvider {
        }
    }

+    // ----- Event consumption (mirrors Android LocalCache pattern) -----
+
+    /**
+     * Routes an event to the appropriate consume method.
+     * Returns true if the event was consumed (new), false if already seen.
+     */
+    fun consume(
+        event: Event,
+        relay: NormalizedRelayUrl?,
+    ): Boolean =
+        when (event) {
+            is MetadataEvent -> {
+                consumeMetadata(event)
+                true
+            }
+
+            is TextNoteEvent -> {
+                consumeTextNote(event, relay)
+            }
+
+            is ReactionEvent -> {
+                consumeReaction(event, relay)
+            }
+
+            is LnZapRequestEvent -> {
+                consumeZapRequest(event, relay)
+            }
+
+            is LnZapEvent -> {
+                consumeZap(event, relay)
+            }
+
+            else -> {
+                false
+            }
+        }
+
+    /**
+     * Consumes a kind 1 text note event.
+     * Creates/updates Note in cache and links reply relationships.
+     */
+    private fun consumeTextNote(
+        event: TextNoteEvent,
+        relay: NormalizedRelayUrl?,
+    ): Boolean {
+        val note = getOrCreateNote(event.id)
+        if (note.event != null) return false
+        val author = getOrCreateUser(event.pubKey)
+        val repliesTo = event.tagsWithoutCitations().mapNotNull { getNoteIfExists(it) }
+        note.loadEvent(event, author, repliesTo)
+        relay?.let { note.addRelay(it) }
+        repliesTo.forEach { it.addReply(note) }
+        return true
+    }
+
+    /**
+     * Consumes a kind 7 reaction event.
+     * Links reaction to target notes via e-tags and a-tags.
+     */
+    private fun consumeReaction(
+        event: ReactionEvent,
+        relay: NormalizedRelayUrl?,
+    ): Boolean {
+        val note = getOrCreateNote(event.id)
+        if (note.event != null) return false
+        val author = getOrCreateUser(event.pubKey)
+        val reactedTo =
+            event.originalPost().mapNotNull { getNoteIfExists(it) } +
+                event.taggedAddresses().mapNotNull { addressableNotes.get(it.toValue()) }
+        note.loadEvent(event, author, reactedTo)
+        relay?.let { note.addRelay(it) }
+        reactedTo.forEach { it.addReaction(note) }
+        return true
+    }
+
+    /**
+     * Consumes a kind 9734 zap request event.
+     * Must be consumed before the corresponding LnZapEvent (kind 9735).
+     */
+    private fun consumeZapRequest(
+        event: LnZapRequestEvent,
+        relay: NormalizedRelayUrl?,
+    ): Boolean {
+        val note = getOrCreateNote(event.id)
+        if (note.event != null) return false
+        val author = getOrCreateUser(event.pubKey)
+        note.loadEvent(event, author, emptyList())
+        relay?.let { note.addRelay(it) }
+        return true
+    }
+
+    /**
+     * Consumes a kind 9735 zap receipt event.
+     * Links zap to target notes via the embedded zap request.
+     */
+    private fun consumeZap(
+        event: LnZapEvent,
+        relay: NormalizedRelayUrl?,
+    ): Boolean {
+        val note = getOrCreateNote(event.id)
+        if (note.event != null) return false
+        val author = getOrCreateUser(event.pubKey)
+
+        // Get or consume the embedded zap request
+        val zapRequestEvent = event.zapRequest
+        val zapRequestNote =
+            if (zapRequestEvent != null) {
+                consumeZapRequest(zapRequestEvent, relay)
+                getOrCreateNote(zapRequestEvent.id)
+            } else {
+                null
+            }
+
+        val zappedNotes =
+            event.zappedPost().mapNotNull { getNoteIfExists(it) } +
+                event.taggedAddresses().mapNotNull { addressableNotes.get(it.toValue()) }
+
+        note.loadEvent(event, author, zappedNotes)
+        relay?.let { note.addRelay(it) }
+
+        // Link zap to target notes
+        if (zapRequestNote != null) {
+            zappedNotes.forEach { it.addZap(zapRequestNote, note) }
+        }
+
+        return true
+    }
+
    // ----- NWC Payment operations -----

    /**
@@ -199,17 +339,17 @@ class DesktopLocalCache : ICacheProvider {

    // ----- Note operations -----

-    override fun getNoteIfExists(hexKey: HexKey): Note? = notes[hexKey]
+    override fun getNoteIfExists(hexKey: HexKey): Note? = notes.get(hexKey)

    override fun checkGetOrCreateNote(hexKey: HexKey): Note = getOrCreateNote(hexKey)

    fun getOrCreateNote(hexKey: HexKey): Note =
-        notes.getOrPut(hexKey) {
+        notes.getOrCreate(hexKey) {
            Note(hexKey)
        }

    override fun getOrCreateAddressableNote(key: Address): AddressableNote =
-        addressableNotes.getOrPut(key.toValue()) {
+        addressableNotes.getOrCreate(key.toValue()) {
            AddressableNote(key)
        }

@@ -260,9 +400,9 @@ class DesktopLocalCache : ICacheProvider {

    // ----- Stats -----

-    fun userCount(): Int = users.size
+    fun userCount(): Int = users.size()

-    fun noteCount(): Int = notes.size
+    fun noteCount(): Int = notes.size()

    fun clear() {
        users.clear()
@@ -276,8 +416,18 @@ class DesktopLocalCache : ICacheProvider {
 * Desktop implementation of ICacheEventStream.
 */
 class DesktopCacheEventStream : ICacheEventStream {
-    private val _newEventBundles = MutableSharedFlow<Set<Note>>(replay = 0)
-    private val _deletedEventBundles = MutableSharedFlow<Set<Note>>(replay = 0)
+    private val _newEventBundles =
+        MutableSharedFlow<Set<Note>>(
+            replay = 0,
+            extraBufferCapacity = 64,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )
+    private val _deletedEventBundles =
+        MutableSharedFlow<Set<Note>>(
+            replay = 0,
+            extraBufferCapacity = 64,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )

    override val newEventBundles: SharedFlow<Set<Note>> = _newEventBundles
    override val deletedEventBundles: SharedFlow<Set<Note>> = _deletedEventBundles
@@ -24,6 +24,7 @@ import com.vitorpamplona.amethyst.commons.model.Note
 import com.vitorpamplona.amethyst.commons.relayClient.assemblers.FeedMetadataCoordinator
 import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataPreloader
 import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataRateLimiter
+import com.vitorpamplona.amethyst.commons.service.BasicBundledInsert
 import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache
 import com.vitorpamplona.amethyst.desktop.model.DesktopDmRelayState
 import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -34,6 +35,8 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch

 /**
 * Desktop-specific relay subscriptions coordinator.
@@ -86,6 +89,34 @@ class DesktopRelaySubscriptionsCoordinator(
            },
        )

+    // Event bundler: batches consumed notes before emitting to SharedFlow
+    // 250ms for desktop (Android uses 1000ms to save battery)
+    private val eventBundler =
+        BasicBundledInsert<Note>(
+            delay = 250,
+            dispatcher = Dispatchers.IO,
+            scope = scope,
+        )
+
+    /**
+     * Central event router — consumes an event into the cache and emits to event stream.
+     * Called from relay onEvent callbacks. Non-blocking (launches on IO dispatcher).
+     */
+    fun consumeEvent(
+        event: Event,
+        relay: NormalizedRelayUrl?,
+    ) {
+        scope.launch(Dispatchers.IO) {
+            val consumed = localCache.consume(event, relay)
+            if (consumed) {
+                val note = localCache.getNoteIfExists(event.id) as? Note ?: return@launch
+                eventBundler.invalidateList(note) { batch ->
+                    localCache.eventStream.emitNewNotes(batch)
+                }
+            }
+        }
+    }
+
    /**
     * Start the coordinator.
     * Call once when app starts or user logs in.