Fix non-PK messages from PK users being flaky

This commit is contained in:
Cadence Ember
2026-07-28 01:12:39 +12:00
parent b765f564b3
commit 544c5206c7
4 changed files with 70 additions and 49 deletions
+4 -4
View File
@@ -16,9 +16,9 @@ const mreq = sync.require("../../matrix/mreq")
/** /**
* @param {import("discord-api-types/v10").GatewayMessageCreateDispatchData} message * @param {import("discord-api-types/v10").GatewayMessageCreateDispatchData} message
* @param {import("discord-api-types/v10").APIGuild} guild * @param {import("discord-api-types/v10").APIGuild} guild
* @param {{speedbump_id: string, speedbump_webhook_id: string} | null} row data about the webhook which is proxying messages in this channel * @param {{userID: string, webhookID: string} | null} proxyWebhook data about the webhook which is proxying messages in this channel
*/ */
async function editMessage(message, guild, row) { async function editMessage(message, guild, proxyWebhook) {
const historicalRoomOfMessage = from("message_room").join("historical_channel_room", "historical_room_index").where({message_id: message.id}).select("room_id").get() const historicalRoomOfMessage = from("message_room").join("historical_channel_room", "historical_room_index").where({message_id: message.id}).select("room_id").get()
const currentRoom = from("channel_room").join("historical_channel_room", "room_id").where({channel_id: message.channel_id}).select("room_id", "historical_room_index").get() const currentRoom = from("channel_room").join("historical_channel_room", "room_id").where({channel_id: message.channel_id}).select("room_id", "historical_room_index").get()
if (!currentRoom) return if (!currentRoom) return
@@ -27,9 +27,9 @@ async function editMessage(message, guild, row) {
let {roomID, eventsToRedact, eventsToReplace, eventsToSend, senderMxid, promotions} = await editToChanges.editToChanges(message, guild, api) let {roomID, eventsToRedact, eventsToReplace, eventsToSend, senderMxid, promotions} = await editToChanges.editToChanges(message, guild, api)
if (row && row.speedbump_webhook_id === message.webhook_id) { if (proxyWebhook && proxyWebhook.webhookID === message.webhook_id) {
// Handle the PluralKit public instance // Handle the PluralKit public instance
if (row.speedbump_id === "466378653216014359") { if (proxyWebhook.userID === "466378653216014359") {
senderMxid = await registerPkUser.syncUser(message.id, message.author, roomID, true) senderMxid = await registerPkUser.syncUser(message.id, message.author, roomID, true)
} }
} }
+4 -4
View File
@@ -28,9 +28,9 @@ const channelWebhook = sync.require("../../m2d/actions/channel-webhook")
* @param {DiscordTypes.GatewayMessageCreateDispatchData} message * @param {DiscordTypes.GatewayMessageCreateDispatchData} message
* @param {DiscordTypes.APIGuildChannel} channel * @param {DiscordTypes.APIGuildChannel} channel
* @param {DiscordTypes.APIGuild} guild * @param {DiscordTypes.APIGuild} guild
* @param {{speedbump_id: string, speedbump_webhook_id: string} | null} row data about the webhook which is proxying messages in this channel * @param {{userID: string, webhookID: string} | null} proxyWebhook data about the webhook which is proxying messages in this channel
*/ */
async function sendMessage(message, channel, guild, row) { async function sendMessage(message, channel, guild, proxyWebhook) {
const roomID = await createRoom.ensureRoom(message.channel_id) const roomID = await createRoom.ensureRoom(message.channel_id)
const historicalRoomIndex = select("historical_channel_room", "historical_room_index", {room_id: roomID}).pluck().get() const historicalRoomIndex = select("historical_channel_room", "historical_room_index", {room_id: roomID}).pluck().get()
assert(historicalRoomIndex) assert(historicalRoomIndex)
@@ -38,9 +38,9 @@ async function sendMessage(message, channel, guild, row) {
let senderMxid = null let senderMxid = null
if (dUtils.isWebhookMessage(message)) { if (dUtils.isWebhookMessage(message)) {
const useWebhookProfile = select("guild_space", "webhook_profile", {guild_id: guild.id}).pluck().get() ?? 0 const useWebhookProfile = select("guild_space", "webhook_profile", {guild_id: guild.id}).pluck().get() ?? 0
if (row && row.speedbump_webhook_id === message.webhook_id) { if (proxyWebhook && proxyWebhook.webhookID === message.webhook_id) {
// Handle the PluralKit public instance // Handle the PluralKit public instance
if (row.speedbump_id === "466378653216014359") { if (proxyWebhook.userID === "466378653216014359") {
senderMxid = await registerPkUser.syncUser(message.id, message.author, roomID, true) senderMxid = await registerPkUser.syncUser(message.id, message.author, roomID, true)
} }
} else if (useWebhookProfile) { } else if (useWebhookProfile) {
+46 -28
View File
@@ -5,9 +5,9 @@ const {discord, select, db} = passthrough
const DEBUG_SPEEDBUMP = false const DEBUG_SPEEDBUMP = false
function debugSpeedbump(message) { function debugSpeedbump(...args) {
if (DEBUG_SPEEDBUMP) { if (DEBUG_SPEEDBUMP) {
console.log(message) console.log(...args)
} }
} }
@@ -19,6 +19,11 @@ const KNOWN_BOTS = new Set([
"466378653216014359" // PluralKit "466378653216014359" // PluralKit
]) ])
const CONTEXT_ENUM = {
create: "create",
update: "update"
}
/** /**
* Fetch new speedbump data for the channel and put it in the database as cache * Fetch new speedbump data for the channel and put it in the database as cache
* @param {string} channelID * @param {string} channelID
@@ -34,55 +39,68 @@ async function updateCache(channelID, lastChecked) {
db.prepare("UPDATE channel_room SET speedbump_id = ?, speedbump_webhook_id = ?, speedbump_checked = ? WHERE channel_id = ?").run(foundApplication, foundWebhook, now, channelID) db.prepare("UPDATE channel_room SET speedbump_id = ?, speedbump_webhook_id = ?, speedbump_checked = ? WHERE channel_id = ?").run(foundApplication, foundWebhook, now, channelID)
} }
/** @type {Map<string, number>} messageID -> number of gateway events currently bumping */ /**
* @typedef BumpingEntry
* @prop {number} number number of gateway events currently bumping for this message ID
* @prop {boolean} hasCreate whether there was a message create within the events currently bumping
*/
/** @type {Map<string, BumpingEntry>} messageID -> BumpingEntry */
const bumping = new Map() const bumping = new Map()
/** /**
* Slow down a message. After it passes the speedbump, return whether it's okay or if it's been deleted. * Slow down a message. After it passes the speedbump, return whether it's okay or if it's been deleted.
* @param {keyof typeof CONTEXT_ENUM} context
* @param {string} messageID * @param {string} messageID
* @returns whether it was deleted * @returns whether it was deleted
*/ */
async function doSpeedbump(messageID) { async function doSpeedbump(context, messageID) {
let value = (bumping.get(messageID) ?? 0) + 1 const entry = bumping.get(messageID) ?? (() => {
bumping.set(messageID, value) const entry = {number: 0, hasCreate: false}
debugSpeedbump(`[speedbump] WAIT ${messageID}++ = ${value}`) bumping.set(messageID, entry)
return entry
})()
entry.number++
if (context === "create") entry.hasCreate = true
debugSpeedbump(`[speedbump] WAIT ${messageID}++ =`, entry)
await new Promise(resolve => setTimeout(resolve, SPEEDBUMP_SPEED)) await new Promise(resolve => setTimeout(resolve, SPEEDBUMP_SPEED))
if (!bumping.has(messageID)) { if (!bumping.has(messageID)) {
debugSpeedbump(`[speedbump] DELETED ${messageID}`) debugSpeedbump(`[speedbump] DELETED ${messageID}`)
return true return {skip: true, hasCreate: null}
} }
value = (bumping.get(messageID) ?? 0) - 1
if (value <= 0) { if (--entry.number <= 0) {
debugSpeedbump(`[speedbump] OK ${messageID}-- = ${value}`) debugSpeedbump(`[speedbump] OK ${messageID}-- =`, entry)
bumping.delete(messageID) bumping.delete(messageID)
return false return {skip: false, hasCreate: entry.hasCreate}
} else { } else {
debugSpeedbump(`[speedbump] MULTI ${messageID}-- = ${value}`) debugSpeedbump(`[speedbump] MULTI ${messageID}-- =`, entry)
bumping.set(messageID, value) return {skip: true, hasCreate: null}
return true
} }
} }
/** /**
* Check whether to slow down a message, and do it. After it passes the speedbump, return whether it's okay or if it's been deleted. * Check whether to slow down a message, and do it. After it passes the speedbump, return whether it's okay or if it's been deleted.
* @param {string} channelID * @param {keyof typeof CONTEXT_ENUM} context
* @param {string} messageID * @param {{id: string, channel_id: string, author: {id: string}, backfill?: boolean}} message uses the ID to identify, and the userID to only slow down the message when the user has used PK before
* @param {string} [userID] if provided, only slow down the message when the user has used PK before
* @returns whether it was deleted, and data about the channel's (not thread's) speedbump * @returns whether it was deleted, and data about the channel's (not thread's) speedbump
*/ */
async function maybeDoSpeedbump(channelID, messageID, userID) { async function maybeDoSpeedbump(context, message) {
let row = select("channel_room", ["room_id", "thread_parent", "speedbump_id", "speedbump_webhook_id"], {channel_id: channelID}).get() let row = select("channel_room", ["room_id", "thread_parent", "speedbump_id", "speedbump_webhook_id"], {channel_id: message.channel_id}).get()
if (row?.thread_parent) row = select("channel_room", ["room_id", "thread_parent", "speedbump_id", "speedbump_webhook_id"], {channel_id: row.thread_parent}).get() // webhooks belong to the channel, not the thread if (row?.thread_parent) row = select("channel_room", ["room_id", "thread_parent", "speedbump_id", "speedbump_webhook_id"], {channel_id: row.thread_parent}).get() // webhooks belong to the channel, not the thread
if (!row?.speedbump_webhook_id) return {affected: false, row: null} // channel not affected, no speedbump if (!row?.speedbump_webhook_id || !row?.speedbump_id) return {skip: false, proxyWebhook: null} // channel not affected, no speedbump
if (userID) {
if (row.speedbump_webhook_id === userID) return {affected: false, row} // shortcut let proxyWebhook = {userID: row.speedbump_id, webhookID: row.speedbump_webhook_id}
const userHasProxy = select("sim_proxy", "user_id", {proxy_owner_id: userID}).pluck().get() if (message.backfill) return {skip: false, proxyWebhook} // don't slow messages during backfill
if (!userHasProxy) return {affected: false, row} // user has not used PK before, no speedbump if (proxyWebhook.webhookID === message.author.id) return {skip: false, proxyWebhook} // shortcut
} const userHasProxy = select("sim_proxy", "user_id", {proxy_owner_id: message.author.id}).pluck().get()
const affected = await doSpeedbump(messageID) if (!userHasProxy) return {skip: false, proxyWebhook} // user has not used PK before, no speedbump
return {affected, row} // maybe affected, and there is a speedbump const {skip, hasCreate} = await doSpeedbump(context, message.id)
return {skip, hasCreate, proxyWebhook} // maybe affected, and there is a speedbump
} }
/** /**
+16 -13
View File
@@ -288,7 +288,6 @@ module.exports = {
if (!guildID) return // channel must have been a DM channel or something if (!guildID) return // channel must have been a DM channel or something
const roomID = select("channel_room", "room_id", {channel_id: channel.id}).pluck().get() const roomID = select("channel_room", "room_id", {channel_id: channel.id}).pluck().get()
if (!roomID) return // channel wasn't being bridged in the first place if (!roomID) return // channel wasn't being bridged in the first place
// @ts-ignore
await createRoom.unbridgeChannel(channel, guildID) await createRoom.unbridgeChannel(channel, guildID)
}, },
@@ -313,11 +312,10 @@ module.exports = {
if (!createRoom.existsOrAutocreatable(channel, guild.id)) return // Check that the sending-to room exists or is autocreatable if (!createRoom.existsOrAutocreatable(channel, guild.id)) return // Check that the sending-to room exists or is autocreatable
const {affected, row} = await speedbump.maybeDoSpeedbump(message.channel_id, message.id, message.author.id) const {skip, proxyWebhook} = await speedbump.maybeDoSpeedbump("create", message)
if (affected) return if (skip) return
// @ts-ignore await sendMessage.sendMessage(message, channel, guild, proxyWebhook)
await sendMessage.sendMessage(message, channel, guild, row)
retrigger.finishedBridging(message.id) retrigger.finishedBridging(message.id)
}, },
@@ -335,22 +333,27 @@ module.exports = {
if (dUtils.isEphemeralMessage(data)) return // Ephemeral messages are for the eyes of the receiver only! if (dUtils.isEphemeralMessage(data)) return // Ephemeral messages are for the eyes of the receiver only!
// Edits need to go through the speedbump as well. If the message is delayed but the edit isn't, we don't have anything to edit from. // Edits need to go through the speedbump as well. If the message is delayed but the edit isn't, we don't have anything to edit from.
const {affected, row} = await speedbump.maybeDoSpeedbump(data.channel_id, data.id, data.author.id) const {skip, hasCreate, proxyWebhook} = await speedbump.maybeDoSpeedbump("update", data)
if (affected) return if (skip) return
// Check that the sending-to room exists, and deal with Eventual Consistency(TM)
if (!await retrigger.waitForMessage(data.id)) return
/** @type {DiscordTypes.GatewayMessageCreateDispatchData} */ /** @type {DiscordTypes.GatewayMessageCreateDispatchData} */
// @ts-ignore
const message = data const message = data
const channel = client.channels.get(message.channel_id) const channel = client.channels.get(message.channel_id)
if (!channel || !("guild_id" in channel) || !channel.guild_id) return // Nothing we can do in direct messages. if (!channel || !("guild_id" in channel) || !channel.guild_id) return // Nothing we can do in direct messages.
const guild = client.guilds.get(channel.guild_id) const guild = client.guilds.get(channel.guild_id)
assert(guild) assert(guild)
// @ts-ignore if (!hasCreate) {
await retrigger.pauseChanges(message.id, editMessage.editMessage(message, guild, row)) // Standard path for most message updates
// Check that the target message already exists, and deal with Eventual Consistency(TM)
if (!await retrigger.waitForMessage(data.id)) return
await retrigger.pauseChanges(message.id, editMessage.editMessage(message, guild, proxyWebhook))
}
else {
// Path for edit packets that were speedbumped into the latest copy of a message that needs to be created
// Just pretend to be MESSAGE_CREATE
await sendMessage.sendMessage(message, channel, guild, proxyWebhook)
}
}, },
/** /**