Phase 874: protomux-rpc audit export archive pull (v0.8.841)

Extract platform-audit-archive-rpc-mixin for AUDIT_EXPORT_ARCHIVE_REQUEST
RPC-first path in fetchAuditExportArchiveFromMesh; register guild RPC handler.
Harden archive fetch waiters to require full entries (non-breaking fix).
Smokes, CI bundle test:ci-phase874, docs, version 0.8.841.

mixins/runtime/features/platform-audit-archive-rpc-mixin.js

@@ -0,0 +1,95 @@
+'use strict'
+
+const b4a = require('b4a')
+const { id } = require('../../../platform-class-imports')
+const sharedScope = require('../../../platform-pearcord-shared-imports')
+
+function meshPeerIdHex (pk) {
+  if (pk == null) return null
+  if (typeof pk === 'string') {
+    const s = pk.trim()
+    if (/^[0-9a-f]{64}$/i.test(s)) return s.toLowerCase()
+  }
+  if (b4a.isBuffer(pk) || pk instanceof Uint8Array) {
+    return b4a.toString(pk, 'hex')
+  }
+  return String(pk).trim() || null
+}
+
+const platformAuditArchiveRpcMixin = {
+  async _buildAuditExportArchiveResponsePayload (payload) {
+    if (!payload?.guildId || !payload?.archiveId) return null
+    if (this.guild?.guild?.id !== payload.guildId) return null
+    const selfId = this.identity.user?.id || null
+    if (payload.targetMemberId && payload.targetMemberId !== selfId) return null
+    await this._initDeliveryReceipts(payload.guildId)
+    const row = await this.deliveryReceipts.getAuditExportArchive(payload.archiveId)
+    if (!row?.entries?.length) return null
+    return {
+      ...row,
+      respondedBy: selfId,
+      requestId: payload.requestId || null
+    }
+  },
+
+  async _handleAuditExportArchiveRequestRpc (payload) {
+    const res = await this._buildAuditExportArchiveResponsePayload(payload)
+    if (res) res.viaRpc = true
+    return res
+  },
+
+  async _tryFetchAuditExportArchiveFromMeshRpc (archiveId, opts = {}) {
+    if (!sharedScope.isRpcGossipRequestEnabled?.()) return null
+    if (!this.guild?.guild || !this.guild?.requestGossipFromPeer) return null
+    const guildId = this.guild.guild.id
+    let peerIds = [...(this.guild.peers?.keys() || [])]
+    if (!opts.targetMemberId) {
+      let hostPk = meshPeerIdHex(await this._lookupGuildHostPublicKey(this.guild.guild))
+      if (hostPk && !this.guild.peers?.has(hostPk) && peerIds.length === 1) {
+        hostPk = peerIds[0]
+      }
+      if (hostPk && peerIds.includes(hostPk)) {
+        peerIds = [hostPk]
+      } else if (hostPk && !peerIds.length) {
+        peerIds = [hostPk]
+      }
+    }
+    if (!peerIds.length) return null
+    await this._flushGuildGossipOutbox().catch(() => {})
+    try {
+      const { ensureGuildWireChannels } = require('pearcord-guild/mesh')
+      ensureGuildWireChannels(this.guild)
+    } catch {
+      // ignore
+    }
+    const requestId = id()
+    const payload = {
+      guildId,
+      archiveId,
+      requestId,
+      requestedBy: this.identity.user?.id || null,
+      targetMemberId: opts.targetMemberId || null,
+      at: Date.now(),
+      attempt: Number(opts.attempt) || 0
+    }
+    const rpcMs = Number(process.env.PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_MS) || 12000
+    const wireMs =
+      Number(process.env.PEARCORD_AUDIT_EXPORT_ARCHIVE_RPC_WIRE_READY_MS) || 8000
+    const responses = await Promise.all(
+      peerIds.map((peerId) =>
+        this.guild
+          .requestGossipFromPeer(
+            peerId,
+            sharedScope.RPC.AUDIT_EXPORT_ARCHIVE_REQUEST,
+            payload,
+            { timeout: rpcMs, wireReadyMs: wireMs }
+          )
+          .catch(() => null)
+      )
+    )
+    const hit = responses.find((r) => r && Array.isArray(r.entries) && r.entries.length)
+    return hit || null
+  }
+}
+
+module.exports = { platformAuditArchiveRpcMixin }

mixins/runtime/features/platform-audit-automation-export-mixin.js

@@ -304,7 +304,9 @@ async _onAuditExportArchiveGossip (payload) {
   await this._initDeliveryReceipts(payload.guildId)
   const row = await this.deliveryReceipts.ingestAuditExportArchive(payload)
   if (row) {
-    this._resolveAuditArchiveFetchWaiter(row)
+    if (row.entries?.length) {
+      this._resolveAuditArchiveFetchWaiter(row)
+    }
     this.emit('audit-export-archive', row)
   }
   return row
@@ -317,21 +319,12 @@ async listAuditExportArchives (limit = 8) {
 },
 
 async _onAuditExportArchiveRequestGossip (payload) {
-  if (!payload?.guildId || !payload?.archiveId) return null
-  if (this.guild?.guild?.id !== payload.guildId) return null
-  const selfId = this.identity.user?.id || null
-  if (payload.targetMemberId && payload.targetMemberId !== selfId) return null
-  await this._initDeliveryReceipts(payload.guildId)
-  const row = await this.deliveryReceipts.getAuditExportArchive(payload.archiveId)
-  if (!row?.entries?.length) return null
+  const res = await this._buildAuditExportArchiveResponsePayload(payload)
+  if (!res) return null
   if (this.guild.gossipAuditExportArchive) {
-    this.guild.gossipAuditExportArchive({
-      ...row,
-      respondedBy: selfId,
-      requestId: payload.requestId || null
-    })
+    this.guild.gossipAuditExportArchive(res)
   }
-  return row
+  return res
 },
 
 async _touchArchivePeerSeen (userId, source = 'presence') {
@@ -536,7 +529,7 @@ async _fetchAutomationDigestSnapshotFromMeshOnce (snapshotId, opts = {}) {
 },
 
 _resolveAuditArchiveFetchWaiter (archive) {
-  if (!archive?.id) return
+  if (!archive?.id || !archive.entries?.length) return
   for (const [key, handlers] of this._auditArchiveFetchWaiters) {
     if (key === archive.id || key.startsWith(`${archive.id}:`)) {
       for (const fn of handlers) fn(archive)
@@ -588,6 +581,18 @@ async _fetchAuditExportArchiveFromMeshOnce (archiveId, opts = {}) {
     const out = await this.exportAuditExportArchive(archiveId, opts)
     return { ...out, fetchAttempt: (Number(opts.attempt) || 0) + 1 }
   }
+  if (sharedScope.isRpcGossipRequestEnabled?.()) {
+    const rpcArch = await this._tryFetchAuditExportArchiveFromMeshRpc(archiveId, opts)
+    if (rpcArch?.entries?.length) {
+      await this.deliveryReceipts.ingestAuditExportArchive(rpcArch).catch(() => {})
+      const out = await this.exportAuditExportArchive(rpcArch.id || archiveId, opts)
+      return {
+        ...out,
+        fetchAttempt: (Number(opts.attempt) || 0) + 1,
+        viaRpc: true
+      }
+    }
+  }
   const timeoutMs = Math.min(12000, Math.max(800, Number(opts.timeoutMs) || 3500))
   const requestId = id()
   const waitPromise = new Promise((resolve, reject) => {
@@ -618,6 +623,9 @@ async _fetchAuditExportArchiveFromMeshOnce (archiveId, opts = {}) {
     if (row?.entries?.length) return row
     throw new Error('audit export archive not found on mesh')
   })
+  if (!fetched?.entries?.length) {
+    throw new Error('audit export archive not found on mesh')
+  }
   const out = await this.exportAuditExportArchive(fetched.id || archiveId, opts)
   return { ...out, fetchAttempt: (Number(opts.attempt) || 0) + 1 }
 },

mixins/runtime/guild-sync/platform-guild-wire-mixin.js

@@ -37,7 +37,9 @@ _bindGuildGossipRpcHandlers (guildInstance) {
     ? {
         GUILD_SYNC_REQUEST: (payload) => this._handleGuildSyncRequestRpc(payload),
         MEMBER_PAGE_REQUEST: (payload) => this._handleMemberPageRequestRpc(payload),
-        MESSAGE_SEARCH_REQUEST: (payload) => this._handleMessageSearchRequestRpc(payload)
+        MESSAGE_SEARCH_REQUEST: (payload) => this._handleMessageSearchRequestRpc(payload),
+        AUDIT_EXPORT_ARCHIVE_REQUEST: (payload) =>
+          this._handleAuditExportArchiveRequestRpc(payload)
       }
     : null
   const hadHandlers = !!guildInstance._gossipRpcHandlers