perf(nests): index room presence separately from chat for O(speakers) scans
LiveActivitiesChannel.notes is mixed-kind (chat, zaps, raids, clips, presence), and chat dominates by volume in active rooms. The Nests feed filter scanned it twice per room per recompute -- once for any-fresh-presence, once for fresh on-stage presence -- doing an `is MeetingRoomPresenceEvent` cast on every chat message just to find the speakers. Add a presenceNotes index on LiveActivitiesChannel keyed by author pubkey (presence is replaceable per author, so the key auto-collapses heartbeats). Populate it from LocalCache.consume(MeetingRoomPresenceEvent). Switch NestsFeedFilter to a single hasFreshSpeakers() pass over the index, dropping the now-redundant isLiveByPresence() check (hasFreshSpeakers implies it). Migrate NestsFeedLoaded's latest-presence flow to the same index.
This commit is contained in:
@@ -1556,6 +1556,10 @@ object LocalCache : ILocalCache, ICacheProvider {
|
|||||||
* presence event needs to be in there alongside chat. Without
|
* presence event needs to be in there alongside chat. Without
|
||||||
* this, presence-driven inclusion can't see follows broadcasting
|
* this, presence-driven inclusion can't see follows broadcasting
|
||||||
* in the room (only chat-driven inclusion would fire).
|
* in the room (only chat-driven inclusion would fire).
|
||||||
|
*
|
||||||
|
* Also indexed under [LiveActivitiesChannel.presenceNotes] keyed
|
||||||
|
* by author so the Nests feed can answer "are there speakers on
|
||||||
|
* stage?" without scanning the chat-dominated `notes` map.
|
||||||
*/
|
*/
|
||||||
fun consume(
|
fun consume(
|
||||||
event: MeetingRoomPresenceEvent,
|
event: MeetingRoomPresenceEvent,
|
||||||
@@ -1573,6 +1577,7 @@ object LocalCache : ILocalCache, ICacheProvider {
|
|||||||
val channel = getOrCreateLiveChannel(roomAddress)
|
val channel = getOrCreateLiveChannel(roomAddress)
|
||||||
val versionNote = getOrCreateNote(event.id)
|
val versionNote = getOrCreateNote(event.id)
|
||||||
channel.addNote(versionNote, relay)
|
channel.addNote(versionNote, relay)
|
||||||
|
channel.addPresenceNote(versionNote)
|
||||||
|
|
||||||
return new
|
return new
|
||||||
}
|
}
|
||||||
|
|||||||
+2
-2
@@ -291,9 +291,9 @@ private fun observeRoomLatestPresence(
|
|||||||
channel
|
channel
|
||||||
.flow()
|
.flow()
|
||||||
.notes.stateFlow
|
.notes.stateFlow
|
||||||
.mapLatest { state ->
|
.mapLatest { _ ->
|
||||||
var max: Long? = null
|
var max: Long? = null
|
||||||
state.channel.notes.forEach { _, note ->
|
channel.presenceNotes.forEach { _, note ->
|
||||||
val event = note.event
|
val event = note.event
|
||||||
if (event is MeetingRoomPresenceEvent) {
|
if (event is MeetingRoomPresenceEvent) {
|
||||||
val createdAt = event.createdAt
|
val createdAt = event.createdAt
|
||||||
|
|||||||
+15
-40
@@ -90,7 +90,6 @@ class NestsFeedFilter(
|
|||||||
val noteEvent = it.event as? MeetingSpaceEvent ?: return@filterTo false
|
val noteEvent = it.event as? MeetingSpaceEvent ?: return@filterTo false
|
||||||
if (!hasMinimumNestFields(noteEvent)) return@filterTo false
|
if (!hasMinimumNestFields(noteEvent)) return@filterTo false
|
||||||
if (!isWithinPlannedWindow(noteEvent, now)) return@filterTo false
|
if (!isWithinPlannedWindow(noteEvent, now)) return@filterTo false
|
||||||
if (!isLiveByPresence(noteEvent, presenceCutoff)) return@filterTo false
|
|
||||||
if (!hasFreshSpeakers(noteEvent, presenceCutoff)) return@filterTo false
|
if (!hasFreshSpeakers(noteEvent, presenceCutoff)) return@filterTo false
|
||||||
|
|
||||||
if (filterParams.match(noteEvent, it.relays)) return@filterTo true
|
if (filterParams.match(noteEvent, it.relays)) return@filterTo true
|
||||||
@@ -151,47 +150,23 @@ class NestsFeedFilter(
|
|||||||
starts < now + PLANNED_MAX_FUTURE_SECONDS
|
starts < now + PLANNED_MAX_FUTURE_SECONDS
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Drop OPEN/PRIVATE rooms that have no kind-10312 presence in the
|
|
||||||
* last [PRESENCE_FRESHNESS_WINDOW_SECONDS] AND were not created in
|
|
||||||
* the same window. The "created recently" grace lets brand-new
|
|
||||||
* rooms surface before any speaker has had time to publish their
|
|
||||||
* first heartbeat. Mirrors the NostrNests lobby gate — needs the
|
|
||||||
* lobby-wide kind-10312 REQ added in `filterNestsGlobal` to be
|
|
||||||
* meaningful.
|
|
||||||
*
|
|
||||||
* CLOSED and PLANNED rooms bypass this gate: CLOSED rooms may
|
|
||||||
* carry a recording (EGG-11), and PLANNED rooms have not started
|
|
||||||
* yet so no presence is expected.
|
|
||||||
*/
|
|
||||||
private fun isLiveByPresence(
|
|
||||||
event: MeetingSpaceEvent,
|
|
||||||
presenceCutoff: Long,
|
|
||||||
): Boolean {
|
|
||||||
val status = event.status()
|
|
||||||
if (status != StatusTag.STATUS.LIVE && status != StatusTag.STATUS.PRIVATE) return true
|
|
||||||
if (event.createdAt > presenceCutoff) return true
|
|
||||||
|
|
||||||
val channel = LocalCache.getLiveActivityChannelIfExists(event.address()) ?: return false
|
|
||||||
var fresh = false
|
|
||||||
channel.notes.forEach { _, note ->
|
|
||||||
if (fresh) return@forEach
|
|
||||||
val e = note.event
|
|
||||||
if (e is MeetingRoomPresenceEvent && e.createdAt > presenceCutoff) fresh = true
|
|
||||||
}
|
|
||||||
return fresh
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Drop OPEN/PRIVATE rooms whose live speaker slate is empty. A room
|
* Drop OPEN/PRIVATE rooms whose live speaker slate is empty. A room
|
||||||
* with no fresh kind-10312 presence carrying `onstage=1` has no one
|
* with no fresh kind-10312 presence carrying `onstage=1` published
|
||||||
* left on stage — even if the kind-30312 status still says `live`,
|
* in the last [PRESENCE_FRESHNESS_WINDOW_SECONDS] has no one left
|
||||||
* there is nothing to listen to and the room has effectively ended.
|
* on stage — even if the kind-30312 status still says `live`,
|
||||||
|
* there is nothing to listen to and the room has effectively
|
||||||
|
* ended. Mirrors (and tightens) the NostrNests lobby gate.
|
||||||
*
|
*
|
||||||
* Same created-at grace as [isLiveByPresence] so brand-new rooms
|
* Brand-new rooms get a created-at grace so they surface before
|
||||||
* surface before the first speaker heartbeat arrives. CLOSED and
|
* the first speaker heartbeat arrives. CLOSED rooms bypass this
|
||||||
* PLANNED rooms bypass this gate for the same reasons listed on
|
* gate (they may carry a recording — EGG-11), as do PLANNED rooms
|
||||||
* [isLiveByPresence].
|
* (not started yet, no presence expected).
|
||||||
|
*
|
||||||
|
* Reads the room's [LiveActivitiesChannel.presenceNotes] index
|
||||||
|
* (keyed by author, populated by
|
||||||
|
* `LocalCache.consume(MeetingRoomPresenceEvent)`) so the scan is
|
||||||
|
* O(speakers) instead of O(all chat + zaps + presence).
|
||||||
*/
|
*/
|
||||||
private fun hasFreshSpeakers(
|
private fun hasFreshSpeakers(
|
||||||
event: MeetingSpaceEvent,
|
event: MeetingSpaceEvent,
|
||||||
@@ -203,7 +178,7 @@ class NestsFeedFilter(
|
|||||||
|
|
||||||
val channel = LocalCache.getLiveActivityChannelIfExists(event.address()) ?: return false
|
val channel = LocalCache.getLiveActivityChannelIfExists(event.address()) ?: return false
|
||||||
var hasSpeaker = false
|
var hasSpeaker = false
|
||||||
channel.notes.forEach { _, note ->
|
channel.presenceNotes.forEach { _, note ->
|
||||||
if (hasSpeaker) return@forEach
|
if (hasSpeaker) return@forEach
|
||||||
val e = note.event
|
val e = note.event
|
||||||
if (e is MeetingRoomPresenceEvent && e.createdAt > presenceCutoff && e.onstage() == true) {
|
if (e is MeetingRoomPresenceEvent && e.createdAt > presenceCutoff && e.onstage() == true) {
|
||||||
|
|||||||
+20
@@ -26,9 +26,11 @@ import com.vitorpamplona.amethyst.commons.model.Note
|
|||||||
import com.vitorpamplona.amethyst.commons.model.User
|
import com.vitorpamplona.amethyst.commons.model.User
|
||||||
import com.vitorpamplona.amethyst.commons.util.toShortDisplay
|
import com.vitorpamplona.amethyst.commons.util.toShortDisplay
|
||||||
import com.vitorpamplona.quartz.nip01Core.core.Address
|
import com.vitorpamplona.quartz.nip01Core.core.Address
|
||||||
|
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||||
import com.vitorpamplona.quartz.nip01Core.tags.aTag.ATag
|
import com.vitorpamplona.quartz.nip01Core.tags.aTag.ATag
|
||||||
import com.vitorpamplona.quartz.nip19Bech32.entities.NAddress
|
import com.vitorpamplona.quartz.nip19Bech32.entities.NAddress
|
||||||
import com.vitorpamplona.quartz.nip53LiveActivities.streaming.LiveActivitiesEvent
|
import com.vitorpamplona.quartz.nip53LiveActivities.streaming.LiveActivitiesEvent
|
||||||
|
import com.vitorpamplona.quartz.utils.cache.LargeCache
|
||||||
|
|
||||||
@Stable
|
@Stable
|
||||||
class LiveActivitiesChannel(
|
class LiveActivitiesChannel(
|
||||||
@@ -40,6 +42,24 @@ class LiveActivitiesChannel(
|
|||||||
// Important to keep this long-term reference because LocalCache uses WeakReferences.
|
// Important to keep this long-term reference because LocalCache uses WeakReferences.
|
||||||
var infoNote: Note? = null
|
var infoNote: Note? = null
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Audio-room presence index (NIP-53 kind-10312) keyed by author
|
||||||
|
* pubkey. Presence is replaceable (one per author), so keying on
|
||||||
|
* the author auto-collapses heartbeat versions instead of growing
|
||||||
|
* unbounded the way `notes` does. Empty for streaming channels
|
||||||
|
* (kind-30311) — only kind-30312 rooms publish presence.
|
||||||
|
*
|
||||||
|
* Lets feeds answer "is anyone live on stage in this room?"
|
||||||
|
* without scanning the chat-dominated `notes` map. See
|
||||||
|
* [com.vitorpamplona.amethyst.ui.screen.loggedIn.nests.dal.NestsFeedFilter].
|
||||||
|
*/
|
||||||
|
val presenceNotes = LargeCache<HexKey, Note>()
|
||||||
|
|
||||||
|
fun addPresenceNote(note: Note) {
|
||||||
|
val author = note.author?.pubkeyHex ?: return
|
||||||
|
presenceNotes.put(author, note)
|
||||||
|
}
|
||||||
|
|
||||||
fun address() = address
|
fun address() = address
|
||||||
|
|
||||||
override fun relays() = info?.allRelayUrls()?.toSet()?.ifEmpty { null } ?: super.relays()
|
override fun relays() = info?.allRelayUrls()?.toSet()?.ifEmpty { null } ?: super.relays()
|
||||||
|
|||||||
Reference in New Issue
Block a user