diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt index b316174452..920631cf99 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt @@ -2645,20 +2645,58 @@ object LocalCache : ILocalCache, ICacheProvider { } chatroomList.forEach { userHex, room -> + // History floors are pinned per scope on first advance; null means that window never paged + // history, so its cursors hold no position to misalign and nothing needs rewinding. Only the + // bands strictly BELOW a floor are this window's responsibility — a pruned message newer than + // the floor is the always-on live tail's concern, and rewinding history for it would needlessly + // re-page (and, for a busy room straddling the floor, mis-set the boundary). Hence the per-floor + // filter when accumulating below. + val giftWrapFloor = room.giftWrapHistory.floor + val accountNip04Floor = room.nip04History.floor + room.rooms.map { key, chatroom -> val toBeRemoved = chatroom.pruneMessagesToTheLatestOnly() val childrenToBeRemoved = mutableListOf() - toBeRemoved.forEach { - childrenToBeRemoved.addAll(removeIfWrap(it)) - unlinkAndRemove(it) + // Newest pruned `created_at` per relay, in each window's cursor space, capped at < floor. + // Gift wraps page by the OUTER wrap time (from the rumor's host stub); NIP-04 by the event's + // own time, and a kind:4 belongs to BOTH the account (rooms-list) and per-conversation cursor. + val giftWrapPruned = HashMap() + val accountNip04Pruned = HashMap() + val roomNip04Pruned = HashMap() + // chatroom.nip04History is lazy — only touch (allocate) it when this room actually drops a + // kind:4 message, so rooms that never paged conversation history pay nothing. + val roomNip04Floor = if (toBeRemoved.any { it.event is PrivateDmEvent }) chatroom.nip04History.floor else null - childrenToBeRemoved.addAll(it.clearChildLinks()) + toBeRemoved.forEach { note -> + when (val ev = note.event) { + is WrappedEvent -> + if (giftWrapFloor != null) { + val outerUntil = ev.host?.createdAt ?: ev.createdAt + if (outerUntil < giftWrapFloor) note.relays.forEach { giftWrapPruned.merge(it, outerUntil, ::maxOf) } + } + is PrivateDmEvent -> { + val until = ev.createdAt + if (accountNip04Floor != null && until < accountNip04Floor) note.relays.forEach { accountNip04Pruned.merge(it, until, ::maxOf) } + if (roomNip04Floor != null && until < roomNip04Floor) note.relays.forEach { roomNip04Pruned.merge(it, until, ::maxOf) } + } + } + + childrenToBeRemoved.addAll(removeIfWrap(note)) + unlinkAndRemove(note) + + childrenToBeRemoved.addAll(note.clearChildLinks()) } unlinkAndRemove(childrenToBeRemoved) + // Realign the windows so a relay that already paged past (or `done` below) the dropped band + // re-requests it on the next demand-advance instead of skipping the hole. + if (giftWrapPruned.isNotEmpty()) room.giftWrapHistory.rewindTo(giftWrapPruned) + if (accountNip04Pruned.isNotEmpty()) room.nip04History.rewindTo(accountNip04Pruned) + if (roomNip04Pruned.isNotEmpty()) chatroom.nip04History.rewindTo(roomNip04Pruned) + if (toBeRemoved.size > 1) { println( "PRUNE: ${toBeRemoved.size} private messages from $userHex to ${key.users.joinToString()} removed. ${chatroom.messages.size} kept", diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/RelayLoadingCursorsTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/RelayLoadingCursorsTest.kt index 7f5237e7dc..b297dabc9f 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/RelayLoadingCursorsTest.kt +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/RelayLoadingCursorsTest.kt @@ -106,4 +106,101 @@ class RelayLoadingCursorsTest { val cursors = RelayLoadingCursors() assertEquals(null, cursors.deepestReached(emptyList(), start)) } + + // ── rewindTo: realign the window after the cache prunes messages out of it ── + + @Test + fun rewindReopensThePrunedBandAndResumesFromItOnNextAdvance() { + val cursors = RelayLoadingCursors() + cursors.floor = start + + // page deep: floor 1000 → reached 200 + cursors.advance(relayA, start) + cursors.onEvent(relayA, 900) + cursors.onEvent(relayA, 200) + cursors.onEose(relayA) + assertEquals(200L, cursors.reachedUntilFor(relayA, start)) + + // prune drops everything older than 700 (newest pruned = 700) + cursors.rewindTo(mapOf(relayA to 700L)) + + // reached pulled up to just above the pruned band, not done, and un-armed (demand-driven) + assertEquals(701L, cursors.reachedUntilFor(relayA, start)) + assertFalse(cursors.isDone(relayA)) + assertEquals(emptyList(), cursors.armedRelays(listOf(relayA))) + + // the next advance resumes at the boundary and re-requests the pruned band (until = 700), + // NOT from the floor (which would re-stream the still-held tail above 700) + assertTrue(cursors.advance(relayA, start)) + assertEquals(700L, cursors.requestedUntilFor(relayA)) + } + + @Test + fun rewindClearsDoneSoAnExhaustedRelayCanReFetch() { + val cursors = RelayLoadingCursors() + cursors.floor = start + + cursors.advance(relayA, start) + cursors.onEvent(relayA, 300) + cursors.onEose(relayA) // reached 300 + cursors.advance(relayA, start) + cursors.onEose(relayA) // empty page → done + assertTrue(cursors.isDone(relayA)) + + cursors.rewindTo(mapOf(relayA to 500L)) + + assertFalse("a pruned relay must be re-fetchable even after it reached the bottom", cursors.isDone(relayA)) + assertEquals(501L, cursors.reachedUntilFor(relayA, start)) + assertTrue(cursors.advance(relayA, start)) + assertEquals(500L, cursors.requestedUntilFor(relayA)) + } + + @Test + fun rewindNeverClimbsAboveTheFloor() { + val cursors = RelayLoadingCursors() + cursors.floor = start + + cursors.advance(relayA, start) + cursors.onEvent(relayA, 300) + cursors.onEose(relayA) // reached 300 + + // a boundary at/above the floor clamps to the floor (history lives strictly below it) + cursors.rewindTo(mapOf(relayA to start)) + assertEquals(start, cursors.reachedUntilFor(relayA, start)) + } + + @Test + fun rewindSkipsRelaysWithoutACursorOrShallowerThanThePrunedBand() { + val cursors = RelayLoadingCursors() + cursors.floor = start + + // A delivered to 200; B never paged + cursors.advance(relayA, start) + cursors.onEvent(relayA, 200) + cursors.onEose(relayA) + + // B has no cursor (never paged) → skipped, no entry minted; A's reach (200) is already shallower + // than a boundary of 150 (target 151), so it needs no rewind either. + cursors.rewindTo(mapOf(relayB to 500L, relayA to 150L)) + + // B: untouched (still at the floor, unarmed) + assertEquals(start, cursors.reachedUntilFor(relayB, start)) + assertEquals(emptyList(), cursors.armedRelays(listOf(relayB))) + // A: unchanged + assertEquals(200L, cursors.reachedUntilFor(relayA, start)) + } + + @Test + fun rewindIsANoOpWhenTheWindowNeverPagedHistory() { + val cursors = RelayLoadingCursors() + // floor is null (never advanced any history page) + cursors.advance(relayA, start) + cursors.onEvent(relayA, 200) + cursors.onEose(relayA) + + cursors.rewindTo(mapOf(relayA to 150L)) + + // unchanged: with no pinned floor there is no history window to realign + assertEquals(200L, cursors.reachedUntilFor(relayA, start)) + } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/privateChats/Chatroom.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/privateChats/Chatroom.kt index 995e5ae27c..13d604208d 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/privateChats/Chatroom.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/privateChats/Chatroom.kt @@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.paging.RelayLoadingCursors import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent import com.vitorpamplona.quartz.nip14Subject.subject +import com.vitorpamplona.quartz.nip59Giftwrap.WrappedEvent import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.MutableSharedFlow @@ -145,7 +146,11 @@ class Chatroom : NotesGatherer { } else { // Old messages, keep the last one. sorted.take(1).toSet() - } + sorted.filter { it.flowSet?.isInUse() ?: false } + sorted.filter { it.event !is PrivateDmEvent } + } + sorted.filter { it.flowSet?.isInUse() ?: false } + sorted.filter { it.event !is PrivateDmEvent && it.event !is WrappedEvent } + // Both DM protocols are pruned by the recency rule above: NIP-04 (PrivateDmEvent) and NIP-17 + // (WrappedEvent rumors — ChatMessageEvent / file headers). Anything else that ever lands in a + // room is kept. The caller realigns the per-relay download window for the dropped messages so + // they can be paged again later (see LocalCache.pruneOldMessages + RelayLoadingCursors.rewindTo). val toRemove = messages.minus(toKeep) messages = toKeep diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/RelayLoadingCursors.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/RelayLoadingCursors.kt index 90ea6be155..cb720f48f9 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/RelayLoadingCursors.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/RelayLoadingCursors.kt @@ -103,11 +103,18 @@ class RelayLoadingCursors { ): Boolean { val c = cursor(relay) if (c.done) return false + val reached = c.reachedUntil c.requestedUntil = - if (c.requestedUntil == null) { - start - } else { - (c.reachedUntil ?: start) - 1 + when { + // Resume just below the oldest event already delivered. Covers both normal page-to-page + // advance and a post-[rewindTo] resume (which un-arms the relay — requestedUntil back to + // null — but keeps the rewound reached point, so the next page picks up at the boundary + // instead of restarting at the floor and re-streaming the still-held tail). + reached != null -> reached - 1 + // Very first page for this relay (nothing delivered, nothing requested yet). + c.requestedUntil == null -> start + // Armed but still mid-page (no EOSE yet) — keep asking from the same top. + else -> start } c.pageCount = 0 c.pageOldest = Long.MAX_VALUE @@ -147,6 +154,45 @@ class RelayLoadingCursors { } } + /** + * Realigns the window after the cache prunes messages out of it: for each `relay → newestPrunedUntil` + * entry, rewinds that relay so it no longer claims to hold anything at or below [newestPrunedUntil] + * (the newest cursor-space `created_at` among the messages pruned from that relay — for gift wraps the + * **outer-wrap** time, recovered from the rumor's [host][com.vitorpamplona.quartz.nip59Giftwrap.HostStub]). + * + * Without this, a relay that already paged past the pruned band — or reached `done` — would never + * re-request the dropped messages: its [reachedUntil] still points below them, so the next [advance] + * starts even older and skips the hole entirely. + * + * The rewind pulls [reachedUntil] back up to just above [newestPrunedUntil] (so the next page's + * `until` re-includes it), clears [done] (there *is* older data to re-fetch again), and un-arms the + * relay (requested cursor back to null) so paging stays demand-driven — the dropped band comes back + * only when the on-screen marker advances the relay again, not eagerly on the next re-subscribe. + * + * Bounds: + * - A relay with no cursor yet (never paged) is skipped — there is no window position to misalign. + * - The rewind never moves [reachedUntil] above the pinned [floor] (history lives strictly below it; + * a pruned message newer than the floor is the live tail's concern, not this window's). + * - A relay whose reached point is already shallower than the pruned band needs no rewind. + */ + fun rewindTo(newestPrunedUntil: Map) { + val floorAt = floor ?: return + newestPrunedUntil.forEach { (relay, prunedUntil) -> + val c = cursors.get(relay) ?: return@forEach + val reached = c.reachedUntil ?: return@forEach + // Re-include the newest pruned event: the next page asks `until = reached - 1`, so reached must + // sit one tick above it. Never climb above the floor. + val target = minOf(prunedUntil + 1, floorAt) + if (reached < target) { + c.reachedUntil = target + c.requestedUntil = null + c.done = false + c.pageCount = 0 + c.pageOldest = Long.MAX_VALUE + } + } + } + /** Relays from [all] that have been armed (advanced at least once) and are not yet [isDone]. */ fun armedRelays(all: Collection): List = all.filter { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/HostStub.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/HostStub.kt index b971711e4b..8fdd3c92d8 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/HostStub.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/HostStub.kt @@ -22,8 +22,19 @@ package com.vitorpamplona.quartz.nip59Giftwrap import com.vitorpamplona.quartz.nip01Core.core.HexKey +/** + * A lightweight reference to the host event a [WrappedEvent] was extracted from — kept on the inner + * event so callers can broadcast / delete / locate the outer wrap without holding the full event. + * + * [createdAt] is the host's own `created_at` (e.g. the kind:1059 gift-wrap timestamp, randomized per + * NIP-59), carried here so a decrypted rumor self-describes its outer-wrap time. The history pager + * cursors page gift wraps by that outer time, so the prune path uses it to realign the per-relay + * download window when a wrapped message is pruned (the chatroom only keeps the inner rumor, whose + * `created_at` is the real message time, not the wrap time). + */ class HostStub( val id: HexKey, val pubKey: HexKey, val kind: Int, + val createdAt: Long, ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/seals/SealedRumorEvent.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/seals/SealedRumorEvent.kt index 3591a5d7a1..ad2c9cf873 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/seals/SealedRumorEvent.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/seals/SealedRumorEvent.kt @@ -70,7 +70,7 @@ class SealedRumorEvent( val event = rumor.mergeWith(this) if (event is WrappedEvent) { - event.host = host ?: HostStub(this.id, this.pubKey, this.kind) + event.host = host ?: HostStub(this.id, this.pubKey, this.kind, this.createdAt) } innerEventId = event.id diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/wraps/GiftWrapEvent.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/wraps/GiftWrapEvent.kt index af7298adf8..7d19424ddc 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/wraps/GiftWrapEvent.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/wraps/GiftWrapEvent.kt @@ -74,7 +74,7 @@ open class GiftWrapEvent( val gift = fromJson(giftStr) if (gift is WrappedEvent) { - gift.host = HostStub(this.id, this.pubKey, this.kind) + gift.host = HostStub(this.id, this.pubKey, this.kind, this.createdAt) } innerEventId = gift.id