Merge pull request #1395 from davotoula/refactor-decrypt-and-index-processor-light

Refactor decrypt and index processor
This commit is contained in:
Vitor Pamplona
2025-08-05 09:50:12 -04:00
committed by GitHub
2 changed files with 424 additions and 224 deletions
@@ -173,7 +173,7 @@ class AccountViewModel(
scope = viewModelScope,
)
val newNotesPreProcessor = DecryptAndIndexProcessor(account, LocalCache)
val newNotesPreProcessor = EventProcessor(account, LocalCache)
var firstRoute: Route? = null
@@ -949,7 +949,7 @@ class AccountViewModel(
}
}
suspend fun loadReactionTo(note: Note?): String? {
fun loadReactionTo(note: Note?): String? {
if (note == null) return null
return note.getReactionBy(userProfile())
@@ -1027,7 +1027,7 @@ class AccountViewModel(
fun getUserIfExists(hex: HexKey): User? = LocalCache.getUserIfExists(hex)
private suspend fun checkGetOrCreateNote(key: HexKey): Note? = LocalCache.checkGetOrCreateNote(key)
private fun checkGetOrCreateNote(key: HexKey): Note? = LocalCache.checkGetOrCreateNote(key)
override suspend fun getOrCreateNote(key: HexKey): Note = LocalCache.getOrCreateNote(key)
@@ -1102,11 +1102,11 @@ class AccountViewModel(
)
}
suspend fun checkGetOrCreatePublicChatChannel(key: HexKey): PublicChatChannel? = LocalCache.getOrCreatePublicChatChannel(key)
fun checkGetOrCreatePublicChatChannel(key: HexKey): PublicChatChannel? = LocalCache.getOrCreatePublicChatChannel(key)
suspend fun checkGetOrCreateLiveActivityChannel(key: Address): LiveActivitiesChannel? = LocalCache.getOrCreateLiveChannel(key)
fun checkGetOrCreateLiveActivityChannel(key: Address): LiveActivitiesChannel? = LocalCache.getOrCreateLiveChannel(key)
suspend fun checkGetOrCreateEphemeralChatChannel(key: RoomId): EphemeralChatChannel? = LocalCache.getOrCreateEphemeralChannel(key)
fun checkGetOrCreateEphemeralChatChannel(key: RoomId): EphemeralChatChannel? = LocalCache.getOrCreateEphemeralChannel(key)
fun checkGetOrCreateChannel(
key: HexKey,
@@ -1660,7 +1660,7 @@ class AccountViewModel(
}
}
suspend fun findUsersStartingWithSync(prefix: String) = LocalCache.findUsersStartingWith(prefix, account)
fun findUsersStartingWithSync(prefix: String) = LocalCache.findUsersStartingWith(prefix, account)
fun relayStatusFlow() = app.client.relayStatusFlow()
@@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.experimental.ephemChat.chat.EphemeralChatEvent
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip03Timestamp.OtsEvent
import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent
import com.vitorpamplona.quartz.nip17Dm.base.ChatroomKey
import com.vitorpamplona.quartz.nip17Dm.files.ChatMessageEncryptedFileHeaderEvent
import com.vitorpamplona.quartz.nip17Dm.messages.ChatMessageEvent
import com.vitorpamplona.quartz.nip28PublicChat.message.ChannelMessageEvent
@@ -39,241 +40,178 @@ import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
import com.vitorpamplona.quartz.nip59Giftwrap.seals.SealedRumorEvent
import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent
import kotlinx.coroutines.CancellationException
import kotlin.reflect.KClass
class DecryptAndIndexProcessor(
val account: Account,
val cache: LocalCache,
class EventProcessor(
private val account: Account,
private val cache: LocalCache,
) {
private val decryptionService = DecryptionService(account)
private val indexingService = IndexingService(account, cache)
private val chatroomService = ChatroomService(account)
private val eventHandlers = createEventHandlers()
suspend fun consume(note: Note) {
val noteEvent = note.event
if (noteEvent != null) {
note.event?.let { event ->
try {
consumeAlreadyVerified(noteEvent, note, note)
processEvent(event, note, note)
} catch (e: Exception) {
Log.e("PrecacheNewNotesProcessor", "Error processing note", e)
}
}
}
suspend fun consumeAlreadyVerified(
event: Event,
eventNote: Note,
publicNote: Note,
) {
when (event) {
is PrivateDmEvent -> {
if (event.canDecrypt(account.signer.pubKey)) {
val talkingWith = event.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(talkingWith, eventNote)
}
}
is ChatMessageEvent -> {
if (event.isIncluded(account.signer.pubKey)) {
val key = event.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(key, eventNote)
}
}
is ChatMessageEncryptedFileHeaderEvent -> {
if (event.isIncluded(account.signer.pubKey)) {
val key = event.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(key, eventNote)
}
}
is OtsEvent -> {
// verifies new OTS upon arrival
Amethyst.instance.otsVerifCache.cacheVerify(event, account.otsResolverBuilder)
}
is DraftEvent -> {
// Avoid decrypting over and over again if the event already exist.
if (event.pubKey == account.signer.pubKey) {
if (!event.isDeleted()) {
val rumor = account.draftsDecryptionCache.preCachedDraft(event) ?: account.draftsDecryptionCache.cachedDraft(event)
if (rumor != null) {
indexDraftAsRealEvent(eventNote, rumor)
}
}
}
}
is GiftWrapEvent -> {
if (event.recipientPubKey() == account.signer.pubKey) {
val innerGiftId = event.innerEventId
if (innerGiftId == null) {
event.unwrapOrNull(account.signer)?.let { innerGift ->
// clear the encrypted payload to save memory
eventNote.event = event.copyNoContent()
if (cache.justConsume(innerGift, null, false)) {
cache.copyRelaysFromTo(publicNote, innerGift)
val innerGiftNote = cache.getOrCreateNote(innerGift.id)
consumeAlreadyVerified(innerGift, innerGiftNote, publicNote)
}
}
} else {
cache.copyRelaysFromTo(publicNote, innerGiftId)
val innerGiftNote = cache.getOrCreateNote(innerGiftId)
val innerGift = innerGiftNote.event
if (innerGift != null) {
consumeAlreadyVerified(innerGift, innerGiftNote, publicNote)
}
}
}
}
is SealedRumorEvent -> {
val rumorId = event.innerEventId
if (rumorId == null) {
event.unsealOrNull(account.signer)?.let { innerRumor ->
// clear the encrypted payload to save memory
eventNote.event = event.copyNoContent()
// rumors cannot be verified
cache.justConsume(innerRumor, null, true)
cache.copyRelaysFromTo(publicNote, innerRumor)
val innerRumorNote = cache.getOrCreateNote(innerRumor.id)
consumeAlreadyVerified(innerRumor, innerRumorNote, publicNote)
}
} else {
cache.copyRelaysFromTo(publicNote, rumorId)
val innerRumorNote = cache.getOrCreateNote(rumorId)
val innerRumor = innerRumorNote.event
if (innerRumor != null) {
consumeAlreadyVerified(innerRumor, innerRumorNote, publicNote)
}
}
}
is LnZapRequestEvent -> {
if (account.privateZapsDecryptionCache.cachedPrivateZap(event) == null && event.isPrivateZap()) {
account.privateZapsDecryptionCache.decryptPrivateZap(event)
}
Log.e("EventProcessor", "Error processing note", e)
}
}
}
suspend fun delete(note: Note) {
val noteEvent = note.event
if (noteEvent != null) {
note.event?.let { event ->
try {
delete(noteEvent, note)
deleteEvent(event, note)
} catch (e: Exception) {
Log.e("PrecacheNewNotesProcessor", "Error processing note", e)
Log.e("EventProcessor", "Error deleting note", e)
}
}
}
suspend fun delete(
internal suspend fun processEvent(
event: Event,
eventNote: Note,
publicNote: Note,
) {
eventHandlers[event::class]?.process(event, eventNote, publicNote)
}
internal suspend fun deleteEvent(
event: Event,
eventNote: Note,
) {
when (event) {
is PrivateDmEvent -> {
if (event.canDecrypt(account.signer.pubKey)) {
// Avoid decrypting over and over again if the event already exist.
val talkingWith = event.chatroomKey(account.signer.pubKey)
account.chatroomList.removeMessage(talkingWith, eventNote)
}
}
eventHandlers[event::class]?.delete(event, eventNote)
}
is ChatMessageEvent -> {
if (event.isIncluded(account.signer.pubKey)) {
val key = event.chatroomKey(account.signer.pubKey)
account.chatroomList.removeMessage(key, eventNote)
}
}
private fun createEventHandlers(): Map<KClass<out Event>, EventHandler> =
mapOf(
PrivateDmEvent::class to PrivateDmHandler(chatroomService),
ChatMessageEvent::class to ChatMessageHandler(chatroomService),
ChatMessageEncryptedFileHeaderEvent::class to ChatMessageEncryptedFileHeaderHandler(chatroomService),
OtsEvent::class to OtsEventHandler(account),
DraftEvent::class to DraftEventHandler(decryptionService, indexingService),
GiftWrapEvent::class to GiftWrapEventHandler(decryptionService, cache, this),
SealedRumorEvent::class to SealedRumorEventHandler(decryptionService, cache, this),
LnZapRequestEvent::class to LnZapRequestEventHandler(decryptionService),
LnZapEvent::class to LnZapEventHandler(decryptionService),
)
is ChatMessageEncryptedFileHeaderEvent -> {
if (event.isIncluded(account.signer.pubKey)) {
val key = event.chatroomKey(account.signer.pubKey)
account.chatroomList.removeMessage(key, eventNote)
}
}
is DraftEvent -> {
// Avoid decrypting over and over again if the event already exist.
if (event.pubKey == account.signer.pubKey) {
// already deindexed by LocalCache
// deindexDraftAsRealEvent(eventNote)
}
}
is GiftWrapEvent -> {
if (event.recipientPubKey() == account.signer.pubKey) {
val innerGiftId = event.innerEventId
if (innerGiftId != null) {
val innerGiftNote = cache.getNoteIfExists(innerGiftId)
val innerGift = innerGiftNote?.event
if (innerGift != null) {
delete(innerGift, innerGiftNote)
}
}
}
}
is SealedRumorEvent -> {
val rumorId = event.innerEventId
if (rumorId != null) {
val innerRumorNote = cache.getNoteIfExists(rumorId)
val innerRumor = innerRumorNote?.event
if (innerRumor != null) {
delete(innerRumor, innerRumorNote)
}
}
}
is LnZapEvent -> {
event.zapRequest?.let { req ->
if (req.isPrivateZap()) {
// We can't know which account this was for without going through it.
account.privateZapsDecryptionCache.delete(req)
}
}
}
is LnZapRequestEvent -> {
if (event.isPrivateZap()) {
account.privateZapsDecryptionCache.delete(event)
}
}
// ..
suspend fun runNew(newNotes: Set<Note>) {
try {
newNotes.forEach { consume(it) }
handleDeletedDrafts(newNotes)
} catch (e: Exception) {
if (e is CancellationException) throw e
Log.e("EventProcessor", "Error processing batch", e)
}
}
suspend fun runDeleted(notes: Set<Note>) {
try {
notes.forEach { delete(it) }
} catch (e: Exception) {
if (e is CancellationException) throw e
Log.e("EventProcessor", "Error deleting batch", e)
}
}
private suspend fun handleDeletedDrafts(newNotes: Set<Note>) {
val deletedDrafts =
newNotes.mapNotNull { note ->
val event = note.event
if (event is DraftEvent && event.isDeleted()) note else null
}
if (deletedDrafts.isNotEmpty()) {
Log.w("EventProcessor", "Deleting ${deletedDrafts.size} draft notes")
account.delete(deletedDrafts)
}
}
}
class DecryptionService(
val account: Account,
) {
suspend fun unwrapGiftWrap(event: GiftWrapEvent): Event? = event.unwrapOrNull(account.signer)
suspend fun unsealRumor(event: SealedRumorEvent): Event? = event.unsealOrNull(account.signer)
suspend fun decryptDraft(event: DraftEvent): Event? =
account.draftsDecryptionCache.preCachedDraft(event)
?: account.draftsDecryptionCache.cachedDraft(event)
suspend fun handlePrivateZap(event: LnZapRequestEvent) {
if (account.privateZapsDecryptionCache.cachedPrivateZap(event) == null && event.isPrivateZap()) {
account.privateZapsDecryptionCache.decryptPrivateZap(event)
}
}
fun deletePrivateZap(event: LnZapRequestEvent) {
if (event.isPrivateZap()) {
account.privateZapsDecryptionCache.delete(event)
}
}
fun deletePrivateZapFromZapEvent(event: LnZapEvent) {
event.zapRequest?.let { req ->
if (req.isPrivateZap()) {
account.privateZapsDecryptionCache.delete(req)
}
}
}
}
class IndexingService(
private val account: Account,
private val cache: LocalCache,
) {
fun indexDraftAsRealEvent(
draftEventWrap: Note,
rumor: Event,
) {
setupReplyRelationships(draftEventWrap, rumor)
indexByEventType(draftEventWrap, rumor)
}
private fun setupReplyRelationships(
draftEventWrap: Note,
rumor: Event,
) {
draftEventWrap.replyTo = cache.computeReplyTo(rumor)
draftEventWrap.replyTo?.forEach { it.addReply(draftEventWrap) }
}
private fun indexByEventType(
draftEventWrap: Note,
rumor: Event,
) {
val chatroomService = ChatroomService(account)
when (rumor) {
is PrivateDmEvent -> {
if (rumor.canDecrypt(account.signer)) {
val talkingWith = rumor.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(talkingWith, draftEventWrap)
val key = rumor.chatroomKey(account.signer.pubKey)
chatroomService.addMessage(key, draftEventWrap)
}
}
is ChatMessageEvent -> {
if (rumor.isIncluded(account.signer.pubKey)) {
val key = rumor.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(key, draftEventWrap)
chatroomService.addMessage(key, draftEventWrap)
}
}
is ChatMessageEncryptedFileHeaderEvent -> {
if (rumor.isIncluded(account.signer.pubKey)) {
val key = rumor.chatroomKey(account.signer.pubKey)
account.chatroomList.addMessage(key, draftEventWrap)
chatroomService.addMessage(key, draftEventWrap)
}
}
is EphemeralChatEvent -> {
rumor.roomId()?.let {
val channel = cache.getOrCreateEphemeralChannel(it)
rumor.roomId()?.let { roomId ->
val channel = cache.getOrCreateEphemeralChannel(roomId)
channel.addNote(draftEventWrap, null)
}
}
@@ -291,41 +229,303 @@ class DecryptAndIndexProcessor(
}
}
}
}
suspend fun runNew(newNotes: Set<Note>) {
try {
newNotes.forEach {
consume(it)
}
class ChatroomService(
val account: Account,
) {
fun addMessage(
chatroomKey: ChatroomKey,
note: Note,
) {
account.chatroomList.addMessage(chatroomKey, note)
}
val toDelete =
newNotes.mapNotNull {
val noteEvent = it.event
if (noteEvent is DraftEvent && noteEvent.isDeleted()) {
it
} else {
null
}
}
fun removeMessage(
chatroomKey: ChatroomKey,
note: Note,
) {
account.chatroomList.removeMessage(chatroomKey, note)
}
}
if (toDelete.isNotEmpty()) {
Log.w("PrecacheNewNotesProcessor", "Deleting ${toDelete.size} draft notes that should have been deleted already")
account.delete(toDelete)
}
} catch (e: Exception) {
if (e is CancellationException) throw e
Log.e("PrecacheNewNotesProcessor", "This shouldn't happen", e)
interface EventHandler {
suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {}
suspend fun delete(
event: Event,
eventNote: Note,
) {}
}
class PrivateDmHandler(
private val chatroomService: ChatroomService,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as PrivateDmEvent
if (event.canDecrypt(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.addMessage(key, eventNote)
}
}
suspend fun runDeleted(newNotes: Set<Note>) {
try {
newNotes.forEach {
delete(it)
}
} catch (e: Exception) {
if (e is CancellationException) throw e
Log.e("PrecacheNewNotesProcessor", "This shouldn't happen", e)
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as PrivateDmEvent
if (event.canDecrypt(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.removeMessage(key, eventNote)
}
}
}
class ChatMessageHandler(
private val chatroomService: ChatroomService,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as ChatMessageEvent
if (event.isIncluded(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.addMessage(key, eventNote)
}
}
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as ChatMessageEvent
if (event.isIncluded(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.removeMessage(key, eventNote)
}
}
}
class ChatMessageEncryptedFileHeaderHandler(
private val chatroomService: ChatroomService,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as ChatMessageEncryptedFileHeaderEvent
if (event.isIncluded(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.addMessage(key, eventNote)
}
}
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as ChatMessageEncryptedFileHeaderEvent
if (event.isIncluded(chatroomService.account.signer.pubKey)) {
val key = event.chatroomKey(chatroomService.account.signer.pubKey)
chatroomService.removeMessage(key, eventNote)
}
}
}
class OtsEventHandler(
private val account: Account,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as OtsEvent
Amethyst.instance.otsVerifCache.cacheVerify(event, account.otsResolverBuilder)
}
}
class DraftEventHandler(
private val decryptionService: DecryptionService,
private val indexingService: IndexingService,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as DraftEvent
if (event.pubKey == decryptionService.account.signer.pubKey && !event.isDeleted()) {
val rumor = decryptionService.decryptDraft(event)
rumor?.let { indexingService.indexDraftAsRealEvent(eventNote, it) }
}
}
}
class GiftWrapEventHandler(
private val decryptionService: DecryptionService,
private val cache: LocalCache,
private val eventProcessor: EventProcessor,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as GiftWrapEvent
if (event.recipientPubKey() != decryptionService.account.signer.pubKey) return
val innerGiftId = event.innerEventId
if (innerGiftId == null) {
processNewGiftWrap(event, eventNote, publicNote)
} else {
processExistingGiftWrap(innerGiftId, publicNote)
}
}
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as GiftWrapEvent
if (event.recipientPubKey() != decryptionService.account.signer.pubKey) return
event.innerEventId?.let { innerGiftId ->
val innerGiftNote = cache.getNoteIfExists(innerGiftId)
innerGiftNote?.event?.let { innerGift ->
eventProcessor.deleteEvent(innerGift, innerGiftNote)
}
}
}
private suspend fun processNewGiftWrap(
event: GiftWrapEvent,
eventNote: Note,
publicNote: Note,
) {
val innerGift = decryptionService.unwrapGiftWrap(event) ?: return
eventNote.event = event.copyNoContent()
if (cache.justConsume(innerGift, null, false)) {
cache.copyRelaysFromTo(publicNote, innerGift)
val innerGiftNote = cache.getOrCreateNote(innerGift.id)
eventProcessor.processEvent(innerGift, innerGiftNote, publicNote)
}
}
private suspend fun processExistingGiftWrap(
innerGiftId: String,
publicNote: Note,
) {
cache.copyRelaysFromTo(publicNote, innerGiftId)
val innerGiftNote = cache.getOrCreateNote(innerGiftId)
innerGiftNote.event?.let { innerGift ->
eventProcessor.processEvent(innerGift, innerGiftNote, publicNote)
}
}
}
class SealedRumorEventHandler(
private val decryptionService: DecryptionService,
private val cache: LocalCache,
private val eventProcessor: EventProcessor,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as SealedRumorEvent
val rumorId = event.innerEventId
if (rumorId == null) {
processNewSealedRumor(event, eventNote, publicNote)
} else {
processExistingSealedRumor(rumorId, publicNote)
}
}
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as SealedRumorEvent
event.innerEventId?.let { rumorId ->
val innerRumorNote = cache.getNoteIfExists(rumorId)
innerRumorNote?.event?.let { innerRumor ->
eventProcessor.deleteEvent(innerRumor, innerRumorNote)
}
}
}
private suspend fun processNewSealedRumor(
event: SealedRumorEvent,
eventNote: Note,
publicNote: Note,
) {
val innerRumor = decryptionService.unsealRumor(event) ?: return
eventNote.event = event.copyNoContent()
cache.justConsume(innerRumor, null, true)
cache.copyRelaysFromTo(publicNote, innerRumor)
val innerRumorNote = cache.getOrCreateNote(innerRumor.id)
eventProcessor.processEvent(innerRumor, innerRumorNote, publicNote)
}
private suspend fun processExistingSealedRumor(
rumorId: String,
publicNote: Note,
) {
cache.copyRelaysFromTo(publicNote, rumorId)
val innerRumorNote = cache.getOrCreateNote(rumorId)
innerRumorNote.event?.let { innerRumor ->
eventProcessor.processEvent(innerRumor, innerRumorNote, publicNote)
}
}
}
class LnZapRequestEventHandler(
private val decryptionService: DecryptionService,
) : EventHandler {
override suspend fun process(
event: Event,
eventNote: Note,
publicNote: Note,
) {
event as LnZapRequestEvent
decryptionService.handlePrivateZap(event)
}
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as LnZapRequestEvent
decryptionService.deletePrivateZap(event)
}
}
class LnZapEventHandler(
private val decryptionService: DecryptionService,
) : EventHandler {
override suspend fun delete(
event: Event,
eventNote: Note,
) {
event as LnZapEvent
decryptionService.deletePrivateZapFromZapEvent(event)
}
}