From 543f7a329c7a864b8d8f907329ea7161fe7e6d98 Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Thu, 19 Mar 2026 13:55:04 +0200 Subject: [PATCH] =?UTF-8?q?feat(cache):=20Phase=201=20=E2=80=94=20cache-ce?= =?UTF-8?q?ntric=20architecture=20foundation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- desktopApp/build.gradle.kts | 2 + .../vitorpamplona/amethyst/desktop/Main.kt | 8 + .../desktop/cache/BoundedLargeCache.kt | 93 +++++++++ .../desktop/cache/DesktopLocalCache.kt | 182 ++++++++++++++++-- .../DesktopRelaySubscriptionsCoordinator.kt | 31 +++ 5 files changed, 300 insertions(+), 16 deletions(-) create mode 100644 desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/BoundedLargeCache.kt diff --git a/desktopApp/build.gradle.kts b/desktopApp/build.gradle.kts index 1c6264985..0ea2cde8f 100644 --- a/desktopApp/build.gradle.kts +++ b/desktopApp/build.gradle.kts @@ -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) diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt index dcc369138..9d659b67f 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt @@ -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() diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/BoundedLargeCache.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/BoundedLargeCache.kt new file mode 100644 index 000000000..576b50ae1 --- /dev/null +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/BoundedLargeCache.kt @@ -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, V>( + private val maxSize: Int, + private val evictPercent: Float = 0.1f, +) { + private val inner = LargeCache() + + 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 = inner.keys() + + fun values(): Iterable = inner.values() + + fun filterIntoSet(consumer: CacheCollectors.BiFilter): Set = inner.filterIntoSet(consumer) + + fun mapNotNull(consumer: CacheCollectors.BiMapper): List = inner.mapNotNull(consumer) + + fun forEach(consumer: java.util.function.BiConsumer) = inner.forEach(consumer) + + fun count(consumer: CacheCollectors.BiFilter): 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) } + } + } +} diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/DesktopLocalCache.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/DesktopLocalCache.kt index b41bda8de..367b128e6 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/DesktopLocalCache.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/cache/DesktopLocalCache.kt @@ -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() - private val notes = ConcurrentHashMap() - private val addressableNotes = ConcurrentHashMap() + val users = BoundedLargeCache(MAX_USERS) + val notes = BoundedLargeCache(MAX_NOTES) + val addressableNotes = BoundedLargeCache(MAX_ADDRESSABLE) private val deletedEvents = ConcurrentHashMap.newKeySet() - 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>(replay = 0) - private val _deletedEventBundles = MutableSharedFlow>(replay = 0) + private val _newEventBundles = + MutableSharedFlow>( + replay = 0, + extraBufferCapacity = 64, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + private val _deletedEventBundles = + MutableSharedFlow>( + replay = 0, + extraBufferCapacity = 64, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) override val newEventBundles: SharedFlow> = _newEventBundles override val deletedEventBundles: SharedFlow> = _deletedEventBundles diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt index e47b44169..7fe7e489f 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt @@ -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( + 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.