feat(platform): digest snapshot mesh fetch and archive peer last seen

Implement fetchAutomationDigestSnapshotFromMesh with waiters, gossip handlers, scheduled export body gossip, GUILD_SYNC snapshot ingest, listAuditArchiveFetchPeers lastSeenAt, and _touchArchivePeerSeen on presence/mesh join. Phase 106 v0.8.70. Co-authored-by: Cursor <[email protected]>
2026-05-22 05:31:54 -04:00
parent 7adf714b8a
commit 702377b36c
1 changed files with 144 additions and 3 deletions
@@ -259,6 +259,7 @@ class PearcordPlatform extends EventEmitter {
    this.workerReleaseHints = null
    this._auditScheduleTimer = null
    this._auditArchiveFetchWaiters = new Map()
+    this._automationDigestSnapshotFetchWaiters = new Map()
    this.appVersion = opts.appVersion || process.env.PEARCORD_APP_VERSION || '0.0.0'
    this.discovery = null
    this.contacts = null
@@ -1362,6 +1363,14 @@ class PearcordPlatform extends EventEmitter {
        await this._initDeliveryReceipts(payload.guildId)
        await this.deliveryReceipts.ingestAuditExportArchive(archive)
      }
+      for (const snap of payload.automationDigestExportSnapshots || []) {
+        if (!snap?.id) continue
+        await this._initDeliveryReceipts(payload.guildId)
+        await this.deliveryReceipts.ingestAutomationDigestExportSnapshot({
+          ...snap,
+          guildId: payload.guildId
+        })
+      }
      if (payload.hookFailureDigest?.digest) {
        await this._initDeliveryReceipts(payload.guildId)
        await this.deliveryReceipts.ingestHookFailureDigestSlice(
@@ -1469,6 +1478,9 @@ class PearcordPlatform extends EventEmitter {
    guildInstance.on('message-delete', (p) => this.emit('message-delete', p))
    guildInstance.on('presence', (payload) => {
      this.presence?.ingest(payload)
+      if (payload?.userId && this.guild?.guild) {
+        this._touchArchivePeerSeen(payload.userId, 'presence').catch(() => {})
+      }
      this.emit('presence', payload)
    })
    guildInstance.on('profile-cosmetic', (payload) => {
@@ -1502,6 +1514,9 @@ class PearcordPlatform extends EventEmitter {
      this._ingestGuildSync(payload).catch(() => {})
    })
    guildInstance.on('peer', ({ type }) => {
+      if (type === 'join' && this.identity?.user?.id) {
+        this._touchArchivePeerSeen(this.identity.user.id, 'mesh').catch(() => {})
+      }
      if (type !== 'join') return
      setTimeout(() => this._pushGuildSyncToMesh().catch(() => {}), 900)
    })
@@ -1636,6 +1651,12 @@ class PearcordPlatform extends EventEmitter {
    guildInstance.on('audit-export-archive-request', (payload) => {
      this._onAuditExportArchiveRequestGossip(payload).catch(() => {})
    })
+    guildInstance.on('automation-digest-export-snapshot', (payload) => {
+      this._onAutomationDigestExportSnapshotGossip(payload).catch(() => {})
+    })
+    guildInstance.on('automation-digest-export-snapshot-request', (payload) => {
+      this._onAutomationDigestExportSnapshotRequestGossip(payload).catch(() => {})
+    })
    guildInstance.on('audit', async (p) => {
      if (p?.guildId && p?.id) {
        const existing = await this.db.get(COLLECTIONS.AUDIT_LOG, { id: p.id })
@@ -2071,16 +2092,26 @@ class PearcordPlatform extends EventEmitter {
    return row
  }

+  async _touchArchivePeerSeen (userId, source = 'presence') {
+    if (!this.guild?.guild || !userId) return null
+    await this._initDeliveryReceipts(this.guild.guild.id)
+    return this.deliveryReceipts.touchArchivePeerSeen(userId, source)
+  }
+
  async listAuditArchiveFetchPeers () {
    if (!this.guild?.guild) return { meshPeerCount: 0, meshLive: false, options: [] }
    const meshPeerCount = this.guild.getStats?.()?.peers || 0
    const meshLive = meshPeerCount > 0
    const selfId = this.identity.user?.id || null
    const presenceSnap = this.presence?.snapshot?.() || { peers: [] }
-    const onlineIds = new Set(
-      (presenceSnap.peers || []).map((p) => p.userId).filter(Boolean)
+    const presenceSeen = new Map(
+      (presenceSnap.peers || []).map((p) => [p.userId, p.at || now()])
    )
-    if (selfId) onlineIds.add(selfId)
+    if (selfId) {
+      await this._touchArchivePeerSeen(selfId, 'mesh').catch(() => {})
+      presenceSeen.set(selfId, now())
+    }
+    const onlineIds = new Set([...presenceSeen.keys()].filter(Boolean))
    const options = [
      {
        memberId: null,
@@ -2089,10 +2120,12 @@ class PearcordPlatform extends EventEmitter {
          : 'Any peer (mesh offline)',
        online: meshLive,
        meshConnected: meshLive,
+        lastSeenAt: meshLive ? now() : null,
        isSelf: false
      }
    ]
    const members = await this.guild.listMembers()
+    await this._initDeliveryReceipts(this.guild.guild.id)
    for (const m of members) {
      const userId = m.userId || m.id
      if (!userId) continue
@@ -2100,17 +2133,122 @@ class PearcordPlatform extends EventEmitter {
      const name = user?.displayName || user?.username || userId.slice(0, 8)
      const presenceOnline = onlineIds.has(userId)
      const meshConnected = meshLive && presenceOnline
+      const stored = await this.deliveryReceipts.getArchivePeerLastSeen(userId)
+      const lastSeenAt = Math.max(
+        presenceSeen.get(userId) || 0,
+        stored?.lastSeenAt || 0
+      ) || null
      options.push({
        memberId: userId,
        label: userId === selfId ? `${name} (you)` : name,
        online: presenceOnline || (userId === selfId && meshLive),
        meshConnected,
+        lastSeenAt,
        isSelf: userId === selfId
      })
    }
    return { meshPeerCount, meshLive, options }
  }

+  async _onAutomationDigestExportSnapshotGossip (payload) {
+    if (!payload?.guildId || !payload?.id) return null
+    if (this.guild?.guild?.id !== payload.guildId) return null
+    await this._initDeliveryReceipts(payload.guildId)
+    const row = await this.deliveryReceipts.ingestAutomationDigestExportSnapshot(payload)
+    if (row?.body?.length) {
+      this._resolveAutomationDigestSnapshotFetchWaiter(row)
+      this.emit('automation-digest-export-snapshot', row)
+    }
+    return row
+  }
+
+  async _onAutomationDigestExportSnapshotRequestGossip (payload) {
+    if (!payload?.guildId || !payload?.snapshotId) 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.getAutomationDigestExportSnapshot(
+      payload.snapshotId
+    )
+    if (!row?.body?.length) return null
+    if (this.guild.gossipAutomationDigestExportSnapshot) {
+      this.guild.gossipAutomationDigestExportSnapshot({
+        ...row,
+        respondedBy: selfId,
+        requestId: payload.requestId || null
+      })
+    }
+    return row
+  }
+
+  _resolveAutomationDigestSnapshotFetchWaiter (snap) {
+    if (!snap?.id) return
+    const handlers = this._automationDigestSnapshotFetchWaiters.get(snap.id)
+    if (!handlers) return
+    for (const fn of handlers) fn(snap)
+    this._automationDigestSnapshotFetchWaiters.delete(snap.id)
+  }
+
+  async fetchAutomationDigestSnapshotFromMesh (snapshotId, opts = {}) {
+    if (!this.guild?.guild) throw new Error('no guild')
+    const roles = await this._memberRoles()
+    if (!roleHasPermission(roles, PERMISSION.MANAGE_GUILD)) {
+      throw new Error('no permission to fetch automation digest snapshot')
+    }
+    await this._initDeliveryReceipts(this.guild.guild.id)
+    let row = await this.deliveryReceipts.getAutomationDigestExportSnapshot(snapshotId)
+    if (row?.body?.length) {
+      return {
+        snapshotId: row.id,
+        format: row.format,
+        body: row.body,
+        summaryLines: row.summaryLines,
+        exportedAt: row.exportedAt,
+        fetchMesh: true,
+        fetchAttempts: 1
+      }
+    }
+    const timeoutMs = Math.min(12000, Math.max(800, Number(opts.timeoutMs) || 3500))
+    const requestId = id()
+    const waitPromise = new Promise((resolve, reject) => {
+      const timer = setTimeout(() => {
+        this._automationDigestSnapshotFetchWaiters.delete(snapshotId)
+        reject(new Error('automation digest snapshot fetch timed out'))
+      }, timeoutMs)
+      this._automationDigestSnapshotFetchWaiters.set(snapshotId, [
+        (snap) => {
+          clearTimeout(timer)
+          resolve(snap)
+        }
+      ])
+    })
+    if (this.guild.gossipAutomationDigestExportSnapshotRequest) {
+      this.guild.gossipAutomationDigestExportSnapshotRequest({
+        guildId: this.guild.guild.id,
+        snapshotId,
+        requestId,
+        requestedBy: this.identity.user?.id || null,
+        targetMemberId: opts.targetMemberId || null,
+        at: Date.now()
+      })
+    }
+    const fetched = await waitPromise.catch(async () => {
+      row = await this.deliveryReceipts.getAutomationDigestExportSnapshot(snapshotId)
+      if (row?.body?.length) return row
+      throw new Error('automation digest snapshot not found on mesh')
+    })
+    return {
+      snapshotId: fetched.id,
+      format: fetched.format,
+      body: fetched.body,
+      summaryLines: fetched.summaryLines,
+      exportedAt: fetched.exportedAt,
+      fetchMesh: true,
+      fetchAttempts: 1
+    }
+  }
+
  _resolveAuditArchiveFetchWaiter (archive) {
    if (!archive?.id) return
    for (const [key, handlers] of this._auditArchiveFetchWaiters) {
@@ -2458,6 +2596,9 @@ class PearcordPlatform extends EventEmitter {
      lastExportAt: exported.exportedAt
    })
    this.emit('automation-digest-export-scheduled', { exported, snap, schedule })
+    if (snap?.body?.length && this.guild.gossipAutomationDigestExportSnapshot) {
+      this.guild.gossipAutomationDigestExportSnapshot(snap)
+    }
    setTimeout(() => this._pushGuildSyncToMesh().catch(() => {}), 400)
    return { ...exported, snapshotId: snap?.id || null }
  }