diff --git a/index.js b/index.js index 124cf11..c6415db 100644 --- a/index.js +++ b/index.js @@ -567,6 +567,7 @@ class PearcordPlatform extends EventEmitter { this._meshAutoHealCheckTimer = null this._gossipOutboxReplayByGuild = new Map() this._lastGuildSyncChannelCoreKeys = [] + this._forumIndexByGuild = new Map() this._guildOpenDurationSamples = [] this._guildOpenFallbackThrottledUntil = 0 } @@ -1344,6 +1345,198 @@ class PearcordPlatform extends EventEmitter { } } + async _resumeSparseAckCursorsAfterHeal (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid) return { resumed: 0, sparsePulled: 0 } + const cursors = { ...(this._sparseAckCursorByGuild.get(gid) || {}) } + const channels = (await this.guild?.listChannels?.().catch(() => [])) || [] + let resumed = 0 + for (const ch of channels) { + if (!['text', 'announcement', 'thread', 'forum'].includes(ch.type)) continue + const rows = await this.db.find(COLLECTIONS.MESSAGES, { channelId: ch.id }).catch(() => []) + const localTail = rows.length > 0 ? rows.length - 1 : 0 + const prev = Number(cursors[ch.id]) || 0 + const next = Math.max(prev, localTail) + if (next !== prev) resumed++ + cursors[ch.id] = next + } + this._sparseAckCursorByGuild.set(gid, cursors) + let sparsePulled = 0 + const rep = await this._getGuildReplicator(gid).catch(() => null) + if (rep) { + const keys = rep.exportChannelCoreKeys(Object.keys(cursors)) + if (keys.length) { + const opened = await this._eagerOpenAnnouncedCores(gid, keys).catch(() => 0) + sparsePulled = Number(opened) || 0 + } + } + const channelsList = channels + const threadSlice = this._exportThreadArchiveSlice(channelsList, 0) + const threadApplied = await this._ingestThreadArchiveSlice(gid, threadSlice).catch(() => 0) + const forumRows = await this._exportForumIndexForSyncBundle(gid, channelsList, 0) + const forumApplied = await this._ingestForumIndexBundle(gid, forumRows).catch(() => 0) + this._patchGuildSyncHealth({ + guildId: gid, + sparseAckCursorCount: Object.keys(cursors).length + }) + return { resumed, sparsePulled, threadApplied, forumApplied } + } + + async _rebuildHyperbeeIndexesOnHeal (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid || !GuildReplicator.isAvailable()) { + return { channelCount: 0, migrated: 0 } + } + const userId = this.identity?.user?.id + const isOwner = userId && this.guild?.guild?.ownerId === userId + if (isOwner) return { channelCount: 0, migrated: 0, skipped: 'owner' } + const cursors = this._sparseAckCursorByGuild.get(gid) || {} + const channelIds = Object.keys(cursors).slice(0, 24) + if (!channelIds.length) return { channelCount: 0, migrated: 0 } + const rep = await this._getGuildReplicator(gid).catch(() => null) + if (!rep?.rebuildIndexesForChannels) return { channelCount: 0, migrated: 0 } + const batch = await rep + .rebuildIndexesForChannels(channelIds, { dryRun: false }) + .catch(() => ({ results: [] })) + const migrated = (batch.results || []).filter((r) => (Number(r.migrated) || 0) > 0).length + return { channelCount: channelIds.length, migrated, batch } + } + + async _healMemberRolesLww (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid || !this.guildRoles) return { links: 0 } + const roles = await this.guildRoles.listRoles(gid).catch(() => []) + const members = await this.guild?.listMembers?.().catch(() => []) + let links = 0 + for (const m of members || []) { + const memberId = m.userId || m.id + if (!memberId) continue + const row = await this.db.get(COLLECTIONS.MEMBERS, { guildId: gid, userId: memberId }) + if (!row) continue + const at = Number(row.updatedAt) || Number(row.joinedAt) || 0 + const custom = Array.isArray(row.customRoleIds) ? row.customRoleIds : [] + for (const roleId of custom) { + if (!roles.some((r) => r.id === roleId)) continue + links++ + } + if (at > 0) { + await this.db + .insert(COLLECTIONS.MEMBERS, { ...row, updatedAt: at }) + .catch(() => null) + } + } + if (this._memberPageBurstRemaining() > 0) { + this._recordMemberPageBurst() + } + return { links, roleCount: roles.length } + } + + async _healForumSearchIndexOnPartition (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid) return { forumRows: 0, searchUpdated: false } + const channels = (await this.guild?.listChannels?.().catch(() => [])) || [] + const forumRows = await this._exportForumIndexForSyncBundle(gid, channels, 0) + const forumApplied = await this._ingestForumIndexBundle(gid, forumRows) + let searchUpdated = false + if (typeof this.rebuildGuildSearchIndex === 'function') { + await this.rebuildGuildSearchIndex(gid).catch(() => null) + searchUpdated = true + } + return { forumRows: forumRows.length, forumApplied, searchUpdated } + } + + async _healSettingsMeshDeviceSync (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid) return null + await this.syncSettingsMesh({ guildId: gid }).catch(() => null) + const prefs = await this.getPrefs().catch(() => null) + const lastChannelId = + prefs?.lastChannelId || prefs?.lastGuildChannelId || this.activeChannelId || null + const lastGuildId = prefs?.lastGuildId || gid + if (lastChannelId && this.notifications) { + const { RPC } = require('pearcord-shared') + const { broadcastSettingsGossip } = require('pearcord-notifications/mesh') + await broadcastSettingsGossip(this.notifications, RPC.SETTINGS_UPDATE, { + userId: this.identity?.user?.id, + deviceId: this.identity?.deviceId, + updatedAt: Date.now(), + prefs: { + lastGuildId, + lastChannelId, + partitionHealAt: Date.now() + } + }).catch(() => null) + } + return { lastChannelId, lastGuildId, settingsMesh: true } + } + + async _healPartitionExtendedSlices (guildId) { + const gid = guildId || this.guild?.guild?.id + if (!gid) return {} + const channels = (await this.guild?.listChannels?.().catch(() => [])) || [] + const voiceRows = await this._exportVoiceOccupancyForSync(gid) + const voiceApplied = await this._ingestVoiceOccupancyBundle(gid, voiceRows).catch(() => 0) + let emojiSlots = 0 + if (this.emojiRegistry && this.guild?.gossipEmojiUpsert) { + const emojis = (await this.emojiRegistry.list().catch(() => [])).filter( + (r) => !r.guildId || r.guildId === gid + ) + for (const row of emojis.slice(0, 64)) { + this.guild.gossipEmojiUpsert(row) + emojiSlots++ + } + } + let automodReconciled = false + if (this.automod) { + await this.automod.getConfig().catch(() => null) + automodReconciled = true + } + const store = await this._getGuildSidecar(gid).catch(() => null) + if (store?.patchAuditRecent) { + const recent = await this._exportAuditSyncSlice(gid, 16, 0) + if (recent.length) await store.patchAuditRecent(recent).catch(() => null) + } + return { voiceApplied, emojiSlots, automodReconciled } + } + + _fanoutGuildSyncRequestWithAckWatermarks (channelId) { + const guildId = this.guild?.guild?.id + const userId = this.identity?.user?.id + if (!guildId || !userId || !this.guild?.gossipGuildSyncRequest) { + return { requested: false, peerCount: 0 } + } + const wm = this._getGuildSyncWatermark(guildId) + const ch = channelId || this.activeChannelId || null + const peerIds = [...(this.guild.peers?.keys() || [])] + void this._flushGuildGossipOutbox().catch(() => {}) + this.guild + .gossipGuildSyncRequest({ + guildId, + userId, + channelId: ch, + sinceTimestamp: wm.sinceTimestamp || 0, + sinceMessageId: wm.sinceMessageId || null, + fanoutPeerCount: peerIds.length + }) + .catch(() => {}) + const ackMap = this._guildSyncAcks.get(guildId) + return { + requested: true, + peerCount: peerIds.length, + watermark: wm, + ackCount: ackMap?.size ?? 0, + acks: ackMap + ? [...ackMap.entries()].map(([uid, row]) => ({ userId: uid, ...row })) + : [] + } + } + + requestGuildSyncFanout () { + const out = this._fanoutGuildSyncRequestWithAckWatermarks(this.activeChannelId) + this._scheduleGuildSyncRequestBurst() + return out + } + _meshPeerQualityForView (guildId) { const gid = guildId || this.guild?.guild?.id if (!gid) return [] @@ -2618,6 +2811,7 @@ class PearcordPlatform extends EventEmitter { const force = opts.force === true const loadGen = opts.loadGen ?? null const bounded = opts.bounded !== false + const joinMeshCapMs = opts.joinMeshCapMs ?? null const fallbackThrottled = opts.fallbackThrottled === true const g = this.guild if (!g?.guild) { @@ -2660,7 +2854,8 @@ class PearcordPlatform extends EventEmitter { loadGen, bounded, guildId, - fallbackThrottled + fallbackThrottled, + joinMeshCapMs }) this._guildMeshJoinCoalesce = { guildId, topic, guild: g, loadGen, promise } return promise @@ -2672,19 +2867,23 @@ class PearcordPlatform extends EventEmitter { loadGen, bounded, guildId, - fallbackThrottled = false + fallbackThrottled = false, + joinMeshCapMs: joinMeshCapMsOpt = null }) { const meshFlushMs = Number( process.env.PEARCORD_GUILD_SWARM_FLUSH_MS ?? process.env.PEARCORD_SWARM_FLUSH_MS ?? 2500 ) - const joinMeshCapMs = fallbackThrottled - ? Math.max( - 500, - Number(process.env.PEARCORD_GUILD_OPEN_THROTTLE_MESH_CAP_MS) || 1500 - ) - : meshFlushMs + 5000 + const joinMeshCapMs = + Number(joinMeshCapMsOpt) > 0 + ? Number(joinMeshCapMsOpt) + : fallbackThrottled + ? Math.max( + 500, + Number(process.env.PEARCORD_GUILD_OPEN_THROTTLE_MESH_CAP_MS) || 1500 + ) + : meshFlushMs + 5000 try { const g = this.guild if (!g?.guild || g.guild.id !== guildId) { @@ -5385,7 +5584,18 @@ class PearcordPlatform extends EventEmitter { const flush = await this._flushGuildGossipOutbox().catch(() => ({ flushed: 0 })) - this._requestGuildSyncFromHost(this.activeChannelId) + const userId = this.identity?.user?.id + const isOwner = userId && this.guild.guild.ownerId === userId + if (!isOwner) { + const pk = await this._lookupGuildHostPublicKey(this.guild.guild).catch(() => null) + if (pk) { + await this._wireGuildMeshToHost( + { creatorPublicKey: pk, hostPublicKey: pk }, + { hostPublicKey: pk } + ).catch(() => null) + } + } + this._fanoutGuildSyncRequestWithAckWatermarks(this.activeChannelId) this._scheduleGuildSyncRequestBurst([800, 2200, 5000]) span.end({ flushed: flush?.flushed || 0, @@ -6524,7 +6734,23 @@ class PearcordPlatform extends EventEmitter { async _ingestForumIndexBundle (guildId, rows = []) { if (!guildId || !rows?.length) return 0 - return rows.length + let map = this._forumIndexByGuild.get(guildId) + if (!map) { + map = new Map() + this._forumIndexByGuild.set(guildId, map) + } + let applied = 0 + for (const row of rows) { + if (!row?.channelId) continue + const key = `${row.parentId || ''}:${row.channelId}` + const at = Number(row.createdAt) || 0 + const prev = map.get(key) + if (!prev || at >= (Number(prev.createdAt) || 0)) { + map.set(key, { ...row, guildId }) + applied++ + } + } + return applied } async _ingestVoiceOccupancyBundle (guildId, rows = []) { @@ -8340,33 +8566,22 @@ class PearcordPlatform extends EventEmitter { const userId = this.identity?.user?.id if (!guildId || !userId || !this.guild?.gossipGuildSyncRequest) return if (this.guild.guild.ownerId === userId) return - void this._flushGuildGossipOutbox().catch(() => {}) const debKey = `${guildId}:${userId}` if (this._guildSyncRequestDebounce.has(debKey)) return this._guildSyncRequestDebounce.set(debKey, Date.now()) setTimeout(() => this._guildSyncRequestDebounce.delete(debKey), 4000) - const ch = channelId || this.activeChannelId || null - const wm = this._getGuildSyncWatermark(guildId) - this.guild - .gossipGuildSyncRequest({ - guildId, - userId, - channelId: ch, - sinceTimestamp: wm.sinceTimestamp || 0, - sinceMessageId: wm.sinceMessageId || null - }) - .catch(() => {}) + this._fanoutGuildSyncRequestWithAckWatermarks(channelId) } requestGuildSyncFromHost () { const guildId = this.guild?.guild?.id if (!guildId) throw new Error('no guild') - this._requestGuildSyncFromHost(this.activeChannelId) - this._scheduleGuildSyncRequestBurst() + const fanout = this.requestGuildSyncFanout() return { guildId, channelId: this.activeChannelId || null, - requested: true + requested: !!fanout?.requested, + ...fanout } } @@ -8476,11 +8691,12 @@ class PearcordPlatform extends EventEmitter { } } this._sparseAckCursorByGuild.set(guildId, cursors) + const resume = await this._resumeSparseAckCursorsAfterHeal(guildId) this._patchGuildSyncHealth({ guildId, - sparseAckCursorCount: Object.keys(cursors).length + sparseAckCursorCount: Object.keys(this._sparseAckCursorByGuild.get(guildId) || {}).length }) - return { pruned, remaining: Object.keys(cursors).length } + return { pruned, remaining: Object.keys(cursors).length, ...resume } } async _healSidecarExportCursor (guildId) { @@ -8545,7 +8761,10 @@ class PearcordPlatform extends EventEmitter { const presenceMerged = await this._healPresenceVectorPartition(guildId) const readReceiptsHealed = await this._healReadReceiptVectors(guildId) const sparseAckReconciled = await this._reconcileSparseAckCursorsAfterHeal(guildId) + const hyperbeeRebuilt = await this._rebuildHyperbeeIndexesOnHeal(guildId) const channelMetadataHealed = await this._healPartitionChannelMetadata(guildId) + const forumSearchHealed = await this._healForumSearchIndexOnPartition(guildId) + const memberRolesHealed = await this._healMemberRolesLww(guildId) const sidecarHealed = await this._healSidecarExportCursor(guildId) await this._persistGossipOutboxReplayWatermark(guildId).catch(() => null) const memberPageResumed = this._resumeMemberPageChainAfterHeal(guildId) @@ -8553,6 +8772,8 @@ class PearcordPlatform extends EventEmitter { const contactsReconciled = await this._reconcileContactsPresenceWithGuild(guildId).catch( () => 0 ) + const extendedHeal = await this._healPartitionExtendedSlices(guildId) + const settingsDeviceSync = await this._healSettingsMeshDeviceSync(guildId).catch(() => null) const req = this.requestGuildSyncFromHost() const burst = this.requestGuildSyncBurst({ clearPushHalt: false, forcePush: true }) const discoveryRefresh = await this._refreshDiscoveryAfterGuildMeshAction( @@ -8566,12 +8787,18 @@ class PearcordPlatform extends EventEmitter { presenceMerged, readReceiptsHealed, sparseAckReconciled, + hyperbeeRebuilt, channelMetadataHealed, + forumSearchHealed, + memberRolesHealed, sidecarHealed, memberPageResumed, watermark, contactsReconciled, + extendedHeal, + settingsDeviceSync, syncRequested: !!req?.requested, + syncFanoutAckCount: req?.ackCount ?? 0, burst, settingsMesh: !!settingsMesh, discoveryScoped: !!discoveryRefresh?.scoped @@ -14800,7 +15027,16 @@ class PearcordPlatform extends EventEmitter { } } this._wireGuild(this.guild) - await this._requestGuildMeshJoin({ source: 'guild-create', bounded: true }) + const createMeshRtt = + Number(this._guildSyncBurstRttMs) || + Number(process.env.PEARCORD_GUILD_MESH_RTT_MS) || + 120 + const createCaps = adaptiveBurstDelays(createMeshRtt, [8500, 12000, 16000]) + await this._requestGuildMeshJoin({ + source: 'guild-create', + bounded: true, + joinMeshCapMs: createCaps[0] + }) await this._initGuildVoice(result.guild.id) this.messages = new PearcordMessage({ authorId: user.id, db: this.db }) await this.messages.ready()