pearcord/pearcord-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 885: targeted RPC pull orchestrator + audit/digest responders extract (v0.8.852)

Browse Source 
Non-breaking refactor: fetchTargetedGuildGossipRpcPullHit and
fetchGuildGossipRpcPullByConfig orchestrators; shared responder helpers in
gossip-rpc-responders.js; extract audit/digest RPC responders; registry-driven
pull dispatch for mesh/search paths. Default gossip v1 unchanged; fleet v2
cutover still deferred.
This commit is contained in:
Raven Scott
2026-06-04 18:34:55 -04:00
parent f3a4e2a700
commit 7a9fe34478
10 changed files with 177 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2
View File
@@ -92,6 +92,8 @@ Application facade: one `PearcordPlatform` class that wires identity, database,
**Phase 822 (v0.8.789):** Triple-peer topology via `guild-mesh-topology` + `getMeshStabilityStats`. Bundle: `npm run test:ci-phase822`.
**Phase 885 (v0.8.852):** Targeted pull orchestrator + registry dispatch + audit/digest responders extract. Bundle: `npm run test:ci-phase885`.
**Phase 884 (v0.8.851):** `buildGuildGossipRpcPullOpts()` + host/search payload builders; `platform-search-rpc-responders-mixin.js` extract. Bundle: `npm run test:ci-phase884`.
**Phase 883 (v0.8.850):** Gossip RPC handler-config registry + shared fetch payload helper; `platform-guild-sync-rpc-responders-mixin.js` extract. Bundle: `npm run test:ci-phase883`.
gossip-rpc-orchestrators.js
+41 -1
View File
@@ -10,6 +10,9 @@ const {
requestGuildGossipRpcFromPeers
} = require('./gossip-rpc-session')
const { pickFirstGuildRpcResponse } = require('./gossip-rpc-validators')
const { buildGuildGossipRpcPullOpts } = require('./gossip-rpc-pull-config')
const { buildTargetedGuildGossipRpcFetchPayload } = require('./gossip-rpc-payload')
const { lookupGuildGossipRpcPullRegistryEntryByConfig } = require('./gossip-rpc-registry')
/** Shared peer resolve + wire prep for guild gossip RPC fanout (Phase 880). */
async function prepareGuildGossipRpcFanoutRequest (platform, opts = {}) {
@@ -117,9 +120,46 @@ async function fetchGuildGossipRpcPullMerged (platform, opts = {}) {
return mergeFn(responses, prepared.guildId)
}
/** Targeted audit/digest pick-first pull via shared payload builder (Phase 885). */
async function fetchTargetedGuildGossipRpcPullHit (
platform,
configKey,
{ resourceIdField, rpcMethod, opts = {}, nextRequestId }
) {
const requestIdFn = typeof nextRequestId === 'function' ? nextRequestId : () => nextRequestId
return fetchGuildGossipRpcPullHit(
platform,
buildGuildGossipRpcPullOpts(configKey, {
targetMemberId: opts.targetMemberId || null,
rpcMethod,
buildPayload: async ({ guildId }) =>
buildTargetedGuildGossipRpcFetchPayload({
guildId,
resourceIdField,
requestId: requestIdFn(),
requestedBy: platform.identity?.user?.id || null,
opts
})
})
)
}
/** Dispatch pull orchestrator by registry config key (Phase 885). */
async function fetchGuildGossipRpcPullByConfig (platform, configKey, overrides = {}) {
const entry = lookupGuildGossipRpcPullRegistryEntryByConfig(configKey)
if (!entry) return null
const opts = buildGuildGossipRpcPullOpts(configKey, overrides)
if (!opts) return null
if (entry.pullKind === 'hostHit') return fetchGuildGossipRpcFromHostHit(platform, opts)
if (entry.pullKind === 'pullMerged') return fetchGuildGossipRpcPullMerged(platform, opts)
return fetchGuildGossipRpcPullHit(platform, opts)
}
module.exports = {
prepareGuildGossipRpcFanoutRequest,
fetchGuildGossipRpcPullHit,
fetchGuildGossipRpcFromHostHit,
fetchGuildGossipRpcPullMerged
fetchGuildGossipRpcPullMerged,
fetchTargetedGuildGossipRpcPullHit,
fetchGuildGossipRpcPullByConfig
}
gossip-rpc-pull.js
+3 -1
View File
@@ -9,6 +9,7 @@ const viewSnapshot = require('./gossip-rpc-view-snapshot')
const pullConfig = require('./gossip-rpc-pull-config')
const payload = require('./gossip-rpc-payload')
const registry = require('./gossip-rpc-registry')
const responders = require('./gossip-rpc-responders')
module.exports = {
...peer,
@@ -18,5 +19,6 @@ module.exports = {
...viewSnapshot,
...pullConfig,
...payload,
...registry
...registry,
...responders
}
gossip-rpc-responders.js
+24
View File
@@ -0,0 +1,24 @@
'use strict'
const { matchesGuildRpcTargetMember } = require('./gossip-rpc-validators')
/** True when payload guild/targetMember matches platform responder context (Phase 885). */
function assertGuildRpcTargetResponder (payload, platform) {
return matchesGuildRpcTargetMember(
payload,
platform.guild?.guild?.id,
platform.identity?.user?.id || null
)
}
/** Standard guild gossip RPC delivery response envelope (Phase 885). */
function buildGuildRpcDeliveryResponseEnvelope (row, platform, payload) {
if (!row) return null
return {
...row,
respondedBy: platform.identity?.user?.id || null,
requestId: payload?.requestId || null
}
}
module.exports = { assertGuildRpcTargetResponder, buildGuildRpcDeliveryResponseEnvelope }
mixins/runtime/features/platform-audit-archive-rpc-mixin.js
+9 -46
View File
@@ -2,56 +2,19 @@
const { id } = require('../../../platform-class-imports')
const sharedScope = require('../../../platform-pearcord-shared-imports')
const {
matchesGuildRpcTargetMember,
markRpcResponse,
fetchGuildGossipRpcPullHit,
buildGuildGossipRpcPullOpts,
buildTargetedGuildGossipRpcFetchPayload
} = require('../../../gossip-rpc-pull')
const { fetchTargetedGuildGossipRpcPullHit } = require('../../../gossip-rpc-pull')
const { platformAuditArchiveRpcRespondersMixin } = require('./platform-audit-archive-rpc-responders-mixin')
const platformAuditArchiveRpcMixin = {
async _buildAuditExportArchiveResponsePayload (payload) {
if (!payload?.archiveId) return null
if (
!matchesGuildRpcTargetMember(
payload,
this.guild?.guild?.id,
this.identity.user?.id || null
)
) {
return null
}
await this._initDeliveryReceipts(payload.guildId)
const row = await this.deliveryReceipts.getAuditExportArchive(payload.archiveId)
if (!row?.entries?.length) return null
return {
...row,
respondedBy: this.identity.user?.id || null,
requestId: payload.requestId || null
}
},
async _handleAuditExportArchiveRequestRpc (payload) {
return markRpcResponse(await this._buildAuditExportArchiveResponsePayload(payload))
},
...platformAuditArchiveRpcRespondersMixin,
async _tryFetchAuditExportArchiveFromMeshRpc (archiveId, opts = {}) {
return fetchGuildGossipRpcPullHit(
this,
buildGuildGossipRpcPullOpts('auditExportArchive', {
targetMemberId: opts.targetMemberId || null,
rpcMethod: sharedScope.RPC.AUDIT_EXPORT_ARCHIVE_REQUEST,
buildPayload: async ({ guildId }) =>
buildTargetedGuildGossipRpcFetchPayload({
guildId,
resourceIdField: { archiveId },
requestId: id(),
requestedBy: this.identity.user?.id || null,
opts
})
})
)
return fetchTargetedGuildGossipRpcPullHit(this, 'auditExportArchive', {
resourceIdField: { archiveId },
rpcMethod: sharedScope.RPC.AUDIT_EXPORT_ARCHIVE_REQUEST,
opts,
nextRequestId: () => id()
})
}
}
mixins/runtime/features/platform-audit-archive-rpc-responders-mixin.js
+25
View File
@@ -0,0 +1,25 @@
'use strict'
const {
assertGuildRpcTargetResponder,
buildGuildRpcDeliveryResponseEnvelope,
markRpcResponse
} = require('../../../gossip-rpc-pull')
/** Audit export archive protomux-rpc responders (Phase 885 extract). */
const platformAuditArchiveRpcRespondersMixin = {
async _buildAuditExportArchiveResponsePayload (payload) {
if (!payload?.archiveId) return null
if (!assertGuildRpcTargetResponder(payload, this)) return null
await this._initDeliveryReceipts(payload.guildId)
const row = await this.deliveryReceipts.getAuditExportArchive(payload.archiveId)
if (!row?.entries?.length) return null
return buildGuildRpcDeliveryResponseEnvelope(row, this, payload)
},
async _handleAuditExportArchiveRequestRpc (payload) {
return markRpcResponse(await this._buildAuditExportArchiveResponsePayload(payload))
}
}
module.exports = { platformAuditArchiveRpcRespondersMixin }
mixins/runtime/features/platform-automation-digest-snapshot-rpc-mixin.js
+10 -47
View File
@@ -2,58 +2,21 @@
const { id } = require('../../../platform-class-imports')
const sharedScope = require('../../../platform-pearcord-shared-imports')
const { fetchTargetedGuildGossipRpcPullHit } = require('../../../gossip-rpc-pull')
const {
matchesGuildRpcTargetMember,
markRpcResponse,
fetchGuildGossipRpcPullHit,
buildGuildGossipRpcPullOpts,
buildTargetedGuildGossipRpcFetchPayload
} = require('../../../gossip-rpc-pull')
platformAutomationDigestSnapshotRpcRespondersMixin
} = require('./platform-automation-digest-snapshot-rpc-responders-mixin')
const platformAutomationDigestSnapshotRpcMixin = {
async _buildAutomationDigestExportSnapshotResponsePayload (payload) {
if (!payload?.snapshotId) return null
if (
!matchesGuildRpcTargetMember(
payload,
this.guild?.guild?.id,
this.identity.user?.id || null
)
) {
return null
}
await this._initDeliveryReceipts(payload.guildId)
const row = await this.deliveryReceipts.getAutomationDigestExportSnapshot(payload.snapshotId)
if (!row?.body?.length) return null
return {
...row,
respondedBy: this.identity.user?.id || null,
requestId: payload.requestId || null
}
},
async _handleAutomationDigestExportSnapshotRequestRpc (payload) {
return markRpcResponse(
await this._buildAutomationDigestExportSnapshotResponsePayload(payload)
)
},
...platformAutomationDigestSnapshotRpcRespondersMixin,
async _tryFetchAutomationDigestSnapshotFromMeshRpc (snapshotId, opts = {}) {
return fetchGuildGossipRpcPullHit(
this,
buildGuildGossipRpcPullOpts('automationDigestSnapshot', {
targetMemberId: opts.targetMemberId || null,
rpcMethod: sharedScope.RPC.AUTOMATION_DIGEST_EXPORT_SNAPSHOT_REQUEST,
buildPayload: async ({ guildId }) =>
buildTargetedGuildGossipRpcFetchPayload({
guildId,
resourceIdField: { snapshotId },
requestId: id(),
requestedBy: this.identity.user?.id || null,
opts
})
})
)
return fetchTargetedGuildGossipRpcPullHit(this, 'automationDigestSnapshot', {
resourceIdField: { snapshotId },
rpcMethod: sharedScope.RPC.AUTOMATION_DIGEST_EXPORT_SNAPSHOT_REQUEST,
opts,
nextRequestId: () => id()
})
}
}
mixins/runtime/features/platform-automation-digest-snapshot-rpc-responders-mixin.js
+27
View File
@@ -0,0 +1,27 @@
'use strict'
const {
assertGuildRpcTargetResponder,
buildGuildRpcDeliveryResponseEnvelope,
markRpcResponse
} = require('../../../gossip-rpc-pull')
/** Automation digest snapshot protomux-rpc responders (Phase 885 extract). */
const platformAutomationDigestSnapshotRpcRespondersMixin = {
async _buildAutomationDigestExportSnapshotResponsePayload (payload) {
if (!payload?.snapshotId) return null
if (!assertGuildRpcTargetResponder(payload, this)) return null
await this._initDeliveryReceipts(payload.guildId)
const row = await this.deliveryReceipts.getAutomationDigestExportSnapshot(payload.snapshotId)
if (!row?.body?.length) return null
return buildGuildRpcDeliveryResponseEnvelope(row, this, payload)
},
async _handleAutomationDigestExportSnapshotRequestRpc (payload) {
return markRpcResponse(
await this._buildAutomationDigestExportSnapshotResponsePayload(payload)
)
}
}
module.exports = { platformAutomationDigestSnapshotRpcRespondersMixin }
mixins/runtime/mesh/platform-mesh-rpc-pull-mixin.js
+24 -31
View File
@@ -2,10 +2,9 @@
const sharedScope = require('../../../platform-pearcord-shared-imports')
const {
fetchGuildGossipRpcFromHostHit,
fetchGuildGossipRpcPullByConfig,
resolveGuildGossipRpcHostPeerId,
markRpcResponse,
buildGuildGossipRpcPullOpts,
buildGuildSyncHostRpcFetchPayload,
buildMemberPageHostRpcFetchPayload
} = require('../../../gossip-rpc-pull')
@@ -17,18 +16,15 @@ const platformMeshRpcPullMixin = {
const userId = this.identity?.user?.id
if (!guildId || !userId) return null
const off = Math.max(0, Number(offset) || 0)
const page = await fetchGuildGossipRpcFromHostHit(
this,
buildGuildGossipRpcPullOpts('memberPageHost', {
rpcMethod: sharedScope.RPC.MEMBER_PAGE_REQUEST,
buildPayload: async ({ guildId: gid }) =>
buildMemberPageHostRpcFetchPayload({
guildId: gid,
requesterId: userId,
offset: off
})
})
)
const page = await fetchGuildGossipRpcPullByConfig(this, 'memberPageHost', {
rpcMethod: sharedScope.RPC.MEMBER_PAGE_REQUEST,
buildPayload: async ({ guildId: gid }) =>
buildMemberPageHostRpcFetchPayload({
guildId: gid,
requesterId: userId,
offset: off
})
})
if (!page) return null
const added = await this._ingestMemberPage(page)
return markRpcResponse({
@@ -46,23 +42,20 @@ const platformMeshRpcPullMixin = {
const wm = this._getGuildSyncWatermark(guildId)
const ch = channelId || this.activeChannelId || null
const hostPublicKey = await this._lookupGuildHostPublicKey(this.guild.guild)
const bundle = await fetchGuildGossipRpcFromHostHit(
this,
buildGuildGossipRpcPullOpts('guildSyncHost', {
hostPublicKey,
rpcMethod: sharedScope.RPC.GUILD_SYNC_REQUEST,
beforeRequest: async (platform) =>
platform.guild.guild.ownerId === platform.identity?.user?.id,
buildPayload: async ({ guildId: gid }) =>
buildGuildSyncHostRpcFetchPayload({
guildId: gid,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null
})
})
)
const bundle = await fetchGuildGossipRpcPullByConfig(this, 'guildSyncHost', {
hostPublicKey,
rpcMethod: sharedScope.RPC.GUILD_SYNC_REQUEST,
beforeRequest: async (platform) =>
platform.guild.guild.ownerId === platform.identity?.user?.id,
buildPayload: async ({ guildId: gid }) =>
buildGuildSyncHostRpcFetchPayload({
guildId: gid,
userId,
channelId: ch,
sinceTimestamp: wm.sinceTimestamp || 0,
sinceMessageId: wm.sinceMessageId || null
})
})
if (!bundle) return null
await this._ingestGuildSync(bundle)
const hostPk = resolveGuildGossipRpcHostPeerId(this.guild, hostPublicKey)
mixins/runtime/search-discovery/platform-search-mesh-gossip-mixin.js
+12 -16
View File
@@ -4,8 +4,7 @@ const searchScope = require('pearcord-search')
const sharedScope = require('../../../platform-pearcord-shared-imports')
const { id } = require('../../../platform-class-imports')
const {
fetchGuildGossipRpcPullMerged,
buildGuildGossipRpcPullOpts,
fetchGuildGossipRpcPullByConfig,
buildMessageSearchGuildGossipRpcFetchPayload
} = require('../../../gossip-rpc-pull')
const { platformSearchRpcRespondersMixin } = require('./platform-search-rpc-responders-mixin')
@@ -16,20 +15,17 @@ const platformSearchMeshGossipMixin = {
async _fetchGuildSearchFromMeshRpc (guildId, query, opts = {}) {
if (!this.guild?.guild || this.guild.guild.id !== guildId) return null
const requestId = id()
return fetchGuildGossipRpcPullMerged(
this,
buildGuildGossipRpcPullOpts('messageSearchMerge', {
rpcMethod: sharedScope.RPC.MESSAGE_SEARCH_REQUEST,
buildPayload: async ({ guildId: gid }) =>
buildMessageSearchGuildGossipRpcFetchPayload({
guildId: gid,
requestId,
query,
opts,
requestedBy: this.identity.user?.id || null
})
})
)
return fetchGuildGossipRpcPullByConfig(this, 'messageSearchMerge', {
rpcMethod: sharedScope.RPC.MESSAGE_SEARCH_REQUEST,
buildPayload: async ({ guildId: gid }) =>
buildMessageSearchGuildGossipRpcFetchPayload({
guildId: gid,
requestId,
query,
opts,
requestedBy: this.identity.user?.id || null
})
})
},
async _fetchGuildSearchFromMeshOnce (guildId, query, opts = {}) {