Phase 895: export mesh fetch mixin extract to RPC mixins (v0.8.862)
Non-breaking refactor — platform-audit-export-mesh-fetch-mixin.js and platform-automation-digest-mesh-fetch-mixin.js move mesh fetch orchestration out of the 1800+ line audit/automation export mixin into dedicated RPC mixins. Default gossip v1 unchanged; fleet v2 cutover still deferred. Verification: test:ci-phase895, pear run boot, pearcord.log scan.
This commit is contained in:
Raven Scott
@@ -2,10 +2,12 @@
|
||||
|
||||
const { platformAuditArchiveRpcRespondersMixin } = require('./platform-audit-archive-rpc-responders-mixin')
|
||||
const { platformTargetedExportRpcPullMixin } = require('./platform-targeted-export-rpc-pull-mixin')
|
||||
const { platformAuditExportMeshFetchMixin } = require('./platform-audit-export-mesh-fetch-mixin')
|
||||
|
||||
const platformAuditArchiveRpcMixin = {
|
||||
...platformAuditArchiveRpcRespondersMixin,
|
||||
...platformTargetedExportRpcPullMixin,
|
||||
...platformAuditExportMeshFetchMixin,
|
||||
|
||||
async _tryFetchAuditExportArchiveFromMeshRpc (archiveId, opts = {}) {
|
||||
return this._fetchTargetedExportFromMeshRpc('auditExportArchive', { archiveId }, opts)
|
||||
|
||||
@@ -27,12 +27,6 @@ const {
|
||||
buildActivityFeed, enrichActivityWithAttachments, IPC_CONTRACT_VERSION, Hyperswarm,
|
||||
id, now, guildTopic
|
||||
} = require('../../../platform-class-imports')
|
||||
const {
|
||||
markRpcResponse,
|
||||
fetchGuildGossipRpcThenMeshWait,
|
||||
runGuildExportMeshFetchWithRetry
|
||||
} = require('../../../gossip-rpc-pull')
|
||||
|
||||
|
||||
/** Phase 756 — audit log & automation digest export */
|
||||
const platformAuditAutomationExportMixin = {
|
||||
@@ -426,194 +420,6 @@ async _onAutomationDigestExportSnapshotRequestGossip (payload) {
|
||||
return res
|
||||
},
|
||||
|
||||
_resolveAutomationDigestSnapshotFetchWaiter (snap) {
|
||||
if (!snap?.id || !snap.body?.length) 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 (!sharedScope.roleHasPermission(roles, sharedScope.PERMISSION.MANAGE_GUILD)) {
|
||||
throw new Error('no permission to fetch automation digest snapshot')
|
||||
}
|
||||
return runGuildExportMeshFetchWithRetry(
|
||||
(id, o) => this._fetchAutomationDigestSnapshotFromMeshOnce(id, o),
|
||||
snapshotId,
|
||||
{
|
||||
...opts,
|
||||
notFoundMessage: 'automation digest snapshot not found on mesh',
|
||||
computeBackoffMs: deliveryScope.computeArchiveFetchBackoffMs
|
||||
}
|
||||
)
|
||||
},
|
||||
|
||||
async _fetchAutomationDigestSnapshotFromMeshOnce (snapshotId, opts = {}) {
|
||||
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,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
}
|
||||
}
|
||||
const rpcSnap = await fetchGuildGossipRpcThenMeshWait(
|
||||
() => this._tryFetchAutomationDigestSnapshotFromMeshRpc(snapshotId, opts),
|
||||
{
|
||||
isSuccess: (r) => Boolean(r?.body?.length),
|
||||
onRpcHit: async (rpcSnap) => {
|
||||
await this.deliveryReceipts.ingestAutomationDigestExportSnapshot(rpcSnap).catch(() => {})
|
||||
return markRpcResponse({
|
||||
snapshotId: rpcSnap.id || snapshotId,
|
||||
format: rpcSnap.format,
|
||||
body: rpcSnap.body,
|
||||
summaryLines: rpcSnap.summaryLines,
|
||||
exportedAt: rpcSnap.exportedAt,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
})
|
||||
},
|
||||
meshWait: async () => {
|
||||
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')
|
||||
})
|
||||
if (!fetched?.body?.length) {
|
||||
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,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
return rpcSnap
|
||||
},
|
||||
|
||||
_resolveAuditArchiveFetchWaiter (archive) {
|
||||
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)
|
||||
this._auditArchiveFetchWaiters.delete(key)
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
async fetchAuditExportArchiveFromMesh (archiveId, opts = {}) {
|
||||
if (!this.guild?.guild) throw new Error('no guild')
|
||||
const roles = await this._memberRoles()
|
||||
if (!sharedScope.roleHasPermission(roles, sharedScope.PERMISSION.MANAGE_GUILD)) {
|
||||
throw new Error('no permission to fetch audit export archive')
|
||||
}
|
||||
return runGuildExportMeshFetchWithRetry(
|
||||
(id, o) => this._fetchAuditExportArchiveFromMeshOnce(id, o),
|
||||
archiveId,
|
||||
{
|
||||
...opts,
|
||||
notFoundMessage: 'audit export archive not found on mesh',
|
||||
computeBackoffMs: deliveryScope.computeArchiveFetchBackoffMs
|
||||
}
|
||||
)
|
||||
},
|
||||
|
||||
async _fetchAuditExportArchiveFromMeshOnce (archiveId, opts = {}) {
|
||||
await this._initDeliveryReceipts(this.guild.guild.id)
|
||||
let row = await this.deliveryReceipts.getAuditExportArchive(archiveId)
|
||||
if (row?.entries?.length) {
|
||||
const out = await this.exportAuditExportArchive(archiveId, opts)
|
||||
return { ...out, fetchAttempt: (Number(opts.attempt) || 0) + 1 }
|
||||
}
|
||||
const rpcArch = await fetchGuildGossipRpcThenMeshWait(
|
||||
() => this._tryFetchAuditExportArchiveFromMeshRpc(archiveId, opts),
|
||||
{
|
||||
isSuccess: (r) => Boolean(r?.entries?.length),
|
||||
onRpcHit: async (rpcArch) => {
|
||||
await this.deliveryReceipts.ingestAuditExportArchive(rpcArch).catch(() => {})
|
||||
const out = await this.exportAuditExportArchive(rpcArch.id || archiveId, opts)
|
||||
return markRpcResponse({
|
||||
...out,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
})
|
||||
},
|
||||
meshWait: async () => {
|
||||
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._auditArchiveFetchWaiters.delete(archiveId)
|
||||
reject(new Error('audit archive fetch timed out'))
|
||||
}, timeoutMs)
|
||||
const handlers = this._auditArchiveFetchWaiters.get(archiveId) || []
|
||||
handlers.push((arch) => {
|
||||
clearTimeout(timer)
|
||||
resolve(arch)
|
||||
})
|
||||
this._auditArchiveFetchWaiters.set(archiveId, handlers)
|
||||
})
|
||||
if (this.guild.gossipAuditExportArchiveRequest) {
|
||||
this.guild.gossipAuditExportArchiveRequest({
|
||||
guildId: this.guild.guild.id,
|
||||
archiveId,
|
||||
requestId,
|
||||
requestedBy: this.identity.user?.id || null,
|
||||
targetMemberId: opts.targetMemberId || null,
|
||||
at: Date.now(),
|
||||
attempt: Number(opts.attempt) || 0
|
||||
})
|
||||
}
|
||||
const fetched = await waitPromise.catch(async () => {
|
||||
row = await this.deliveryReceipts.getAuditExportArchive(archiveId)
|
||||
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 }
|
||||
}
|
||||
}
|
||||
)
|
||||
return rpcArch
|
||||
},
|
||||
|
||||
async exportAuditExportArchive (archiveId, opts = {}) {
|
||||
if (!this.guild?.guild) throw new Error('no guild')
|
||||
const roles = await this._memberRoles()
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
'use strict'
|
||||
|
||||
const { deliveryScope, sharedScope, id } = require('../../../platform-class-imports')
|
||||
const {
|
||||
markRpcResponse,
|
||||
fetchGuildGossipRpcThenMeshWait,
|
||||
runGuildExportMeshFetchWithRetry
|
||||
} = require('../../../gossip-rpc-pull')
|
||||
|
||||
/** Audit export archive mesh fetch + gossip waiters (Phase 895). */
|
||||
const platformAuditExportMeshFetchMixin = {
|
||||
_resolveAuditArchiveFetchWaiter (archive) {
|
||||
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)
|
||||
this._auditArchiveFetchWaiters.delete(key)
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
async fetchAuditExportArchiveFromMesh (archiveId, opts = {}) {
|
||||
if (!this.guild?.guild) throw new Error('no guild')
|
||||
const roles = await this._memberRoles()
|
||||
if (!sharedScope.roleHasPermission(roles, sharedScope.PERMISSION.MANAGE_GUILD)) {
|
||||
throw new Error('no permission to fetch audit export archive')
|
||||
}
|
||||
return runGuildExportMeshFetchWithRetry(
|
||||
(id, o) => this._fetchAuditExportArchiveFromMeshOnce(id, o),
|
||||
archiveId,
|
||||
{
|
||||
...opts,
|
||||
notFoundMessage: 'audit export archive not found on mesh',
|
||||
computeBackoffMs: deliveryScope.computeArchiveFetchBackoffMs
|
||||
}
|
||||
)
|
||||
},
|
||||
|
||||
async _fetchAuditExportArchiveFromMeshOnce (archiveId, opts = {}) {
|
||||
await this._initDeliveryReceipts(this.guild.guild.id)
|
||||
let row = await this.deliveryReceipts.getAuditExportArchive(archiveId)
|
||||
if (row?.entries?.length) {
|
||||
const out = await this.exportAuditExportArchive(archiveId, opts)
|
||||
return { ...out, fetchAttempt: (Number(opts.attempt) || 0) + 1 }
|
||||
}
|
||||
const rpcArch = await fetchGuildGossipRpcThenMeshWait(
|
||||
() => this._tryFetchAuditExportArchiveFromMeshRpc(archiveId, opts),
|
||||
{
|
||||
isSuccess: (r) => Boolean(r?.entries?.length),
|
||||
onRpcHit: async (rpcArch) => {
|
||||
await this.deliveryReceipts.ingestAuditExportArchive(rpcArch).catch(() => {})
|
||||
const out = await this.exportAuditExportArchive(rpcArch.id || archiveId, opts)
|
||||
return markRpcResponse({
|
||||
...out,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
})
|
||||
},
|
||||
meshWait: async () => {
|
||||
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._auditArchiveFetchWaiters.delete(archiveId)
|
||||
reject(new Error('audit archive fetch timed out'))
|
||||
}, timeoutMs)
|
||||
const handlers = this._auditArchiveFetchWaiters.get(archiveId) || []
|
||||
handlers.push((arch) => {
|
||||
clearTimeout(timer)
|
||||
resolve(arch)
|
||||
})
|
||||
this._auditArchiveFetchWaiters.set(archiveId, handlers)
|
||||
})
|
||||
if (this.guild.gossipAuditExportArchiveRequest) {
|
||||
this.guild.gossipAuditExportArchiveRequest({
|
||||
guildId: this.guild.guild.id,
|
||||
archiveId,
|
||||
requestId,
|
||||
requestedBy: this.identity.user?.id || null,
|
||||
targetMemberId: opts.targetMemberId || null,
|
||||
at: Date.now(),
|
||||
attempt: Number(opts.attempt) || 0
|
||||
})
|
||||
}
|
||||
const fetched = await waitPromise.catch(async () => {
|
||||
row = await this.deliveryReceipts.getAuditExportArchive(archiveId)
|
||||
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 }
|
||||
}
|
||||
}
|
||||
)
|
||||
return rpcArch
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { platformAuditExportMeshFetchMixin }
|
||||
@@ -0,0 +1,113 @@
|
||||
'use strict'
|
||||
|
||||
const { deliveryScope, sharedScope, id } = require('../../../platform-class-imports')
|
||||
const {
|
||||
markRpcResponse,
|
||||
fetchGuildGossipRpcThenMeshWait,
|
||||
runGuildExportMeshFetchWithRetry
|
||||
} = require('../../../gossip-rpc-pull')
|
||||
|
||||
/** Automation digest snapshot mesh fetch + gossip waiters (Phase 895). */
|
||||
const platformAutomationDigestMeshFetchMixin = {
|
||||
_resolveAutomationDigestSnapshotFetchWaiter (snap) {
|
||||
if (!snap?.id || !snap.body?.length) 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 (!sharedScope.roleHasPermission(roles, sharedScope.PERMISSION.MANAGE_GUILD)) {
|
||||
throw new Error('no permission to fetch automation digest snapshot')
|
||||
}
|
||||
return runGuildExportMeshFetchWithRetry(
|
||||
(id, o) => this._fetchAutomationDigestSnapshotFromMeshOnce(id, o),
|
||||
snapshotId,
|
||||
{
|
||||
...opts,
|
||||
notFoundMessage: 'automation digest snapshot not found on mesh',
|
||||
computeBackoffMs: deliveryScope.computeArchiveFetchBackoffMs
|
||||
}
|
||||
)
|
||||
},
|
||||
|
||||
async _fetchAutomationDigestSnapshotFromMeshOnce (snapshotId, opts = {}) {
|
||||
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,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
}
|
||||
}
|
||||
const rpcSnap = await fetchGuildGossipRpcThenMeshWait(
|
||||
() => this._tryFetchAutomationDigestSnapshotFromMeshRpc(snapshotId, opts),
|
||||
{
|
||||
isSuccess: (r) => Boolean(r?.body?.length),
|
||||
onRpcHit: async (rpcSnap) => {
|
||||
await this.deliveryReceipts.ingestAutomationDigestExportSnapshot(rpcSnap).catch(() => {})
|
||||
return markRpcResponse({
|
||||
snapshotId: rpcSnap.id || snapshotId,
|
||||
format: rpcSnap.format,
|
||||
body: rpcSnap.body,
|
||||
summaryLines: rpcSnap.summaryLines,
|
||||
exportedAt: rpcSnap.exportedAt,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
})
|
||||
},
|
||||
meshWait: async () => {
|
||||
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')
|
||||
})
|
||||
if (!fetched?.body?.length) {
|
||||
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,
|
||||
fetchAttempt: (Number(opts.attempt) || 0) + 1
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
return rpcSnap
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { platformAutomationDigestMeshFetchMixin }
|
||||
@@ -4,10 +4,12 @@ const {
|
||||
platformAutomationDigestSnapshotRpcRespondersMixin
|
||||
} = require('./platform-automation-digest-snapshot-rpc-responders-mixin')
|
||||
const { platformTargetedExportRpcPullMixin } = require('./platform-targeted-export-rpc-pull-mixin')
|
||||
const { platformAutomationDigestMeshFetchMixin } = require('./platform-automation-digest-mesh-fetch-mixin')
|
||||
|
||||
const platformAutomationDigestSnapshotRpcMixin = {
|
||||
...platformAutomationDigestSnapshotRpcRespondersMixin,
|
||||
...platformTargetedExportRpcPullMixin,
|
||||
...platformAutomationDigestMeshFetchMixin,
|
||||
|
||||
async _tryFetchAutomationDigestSnapshotFromMeshRpc (snapshotId, opts = {}) {
|
||||
return this._fetchTargetedExportFromMeshRpc(
|
||||
|
||||
Reference in New Issue
Block a user