src/d2m/actions/retrigger.js

@@ -2,9 +2,17 @@
 
 const {EventEmitter} = require("events")
 const passthrough = require("../../passthrough")
-const {select} = passthrough
+const {select, sync, from} = passthrough
+/** @type {import("../../matrix/utils")} */
+const utils = sync.require("../../matrix/utils")
 
-const DEBUG_RETRIGGER = false
+/*
+	Due to Eventual Consistency(TM) an update/delete may arrive before the original message arrives
+	(or before the it has finished being bridged to an event).
+	In this case, wait until the original message has finished bridging, then retrigger the passed function.
+*/
+
+const DEBUG_RETRIGGER = true
 
 function debugRetrigger(message) {
 	if (DEBUG_RETRIGGER) {
@@ -12,81 +20,140 @@ function debugRetrigger(message) {
 	}
 }
 
-const paused = new Set()
-const emitter = new EventEmitter()
+const storage = new class {
+	/** @private @type {Set<string>} */
+	paused = new Set()
+	/** @private @type {Map<string, ((found: Boolean) => any)[]>} id -> list of resolvers */
+	resolves = new Map()
+	/** @private @type {Map<string, ReturnType<setTimeout>>} id -> timer */
+	timers = new Map()
+
+	/**
+	 * The purpose of storage is to store `resolve` and call it at a later time.
+	 * @param {string} id
+	 * @param {(found: Boolean) => any} resolve
+	 */
+	store(id, resolve) {
+		debugRetrigger(`[retrigger] STORE id = ${id}`)
+		this.resolves.set(id, (this.resolves.get(id) || []).concat(resolve)) // add to list in map value
+		if (!this.timers.has(id)) {
+			debugRetrigger(`[retrigger] SET TIMER id = ${id}`)
+			this.timers.set(id, setTimeout(() => this.resolve(id, false), 60 * 1000).unref()) // 1 minute
+		}
+	}
+	
+	/** @param {string} id */
+	isNotPaused(id) {
+		return !storage.paused.has(id)
+	}
+
+	/** @param {string} id */
+	pause(id) {
+		debugRetrigger(`[retrigger] PAUSE id = ${id}`)
+		this.paused.add(id)
+	}
+
+	/**
+	 * Go through `resolves` storage and resolve them all. (Also resets timer/paused.)
+	 * @param {string} id
+	 * @param {boolean} value
+	 */
+	resolve(id, value) {
+		if (this.paused.has(id)) {
+			debugRetrigger(`[retrigger] RESUME id = ${id}`)
+			this.paused.delete(id)
+		}
+
+		if (this.resolves.has(id)) {
+			debugRetrigger(`[retrigger] RESOLVE ${value} id = ${id}`)
+			const fns = this.resolves.get(id) || []
+			this.resolves.delete(id)
+			for (const fn of fns) {
+				fn(value)
+			}
+		}
+
+		if (this.timers.has(id)) {
+			clearTimeout(this.timers.get(id))
+			this.timers.delete(id)
+		}
+	}
+}
 
 /**
- * Due to Eventual Consistency(TM) an update/delete may arrive before the original message arrives
- * (or before the it has finished being bridged to an event).
- * In this case, wait until the original message has finished bridging, then retrigger the passed function.
- * @template {(...args: any[]) => any} T
- * @param {string} inputID
- * @param {T} fn
- * @param {Parameters<T>} rest
- * @returns {boolean} false if the event was found and the function will be ignored, true if the event was not found and the function will be retriggered
+ * @param {string} id
+ * @param {(found: Boolean) => any} resolve
+ * @param {boolean} existsInDatabase
  */
-function eventNotFoundThenRetrigger(inputID, fn, ...rest) {
-	if (!paused.has(inputID)) {
-		if (inputID.match(/^[0-9]+$/)) {
-			const eventID = select("event_message", "event_id", {message_id: inputID}).pluck().get()
-			if (eventID) {
-				debugRetrigger(`[retrigger] OK mid <-> eid = ${inputID} <-> ${eventID}`)
-				return false // event was found so don't retrigger
-			}
-		} else if (inputID.match(/^\$/)) {
-			const messageID = select("event_message", "message_id", {event_id: inputID}).pluck().get()
-			if (messageID) {
-				debugRetrigger(`[retrigger] OK eid <-> mid = ${inputID} <-> ${messageID}`)
-				return false // message was found so don't retrigger
-			}
-		}
+function waitFor(id, resolve, existsInDatabase) {
+	if (existsInDatabase && storage.isNotPaused(id)) { // if event already exists and isn't paused then resolve immediately
+		debugRetrigger(`[retrigger] EXISTS id = ${id}`)
+		return resolve(true)
 	}
 
-	debugRetrigger(`[retrigger] WAIT id = ${inputID}`)
-	emitter.once(inputID, () => {
-		debugRetrigger(`[retrigger] TRIGGER id = ${inputID}`)
-		fn(...rest)
-	})
-	// if the event never arrives, don't trigger the callback, just clean up
-	setTimeout(() => {
-		if (emitter.listeners(inputID).length) {
-			debugRetrigger(`[retrigger] EXPIRE id = ${inputID}`)
-		}
-		emitter.removeAllListeners(inputID)
-	}, 60 * 1000) // 1 minute
-	return true // event was not found, then retrigger
+	// doesn't exist. wait for it to exist. storage will resolve true if it exists or false if it timed out
+	return storage.store(id, resolve)
+}
+
+const GET_EVENT_PREPARED = from("event_message").select("event_id").and("WHERE event_id = ?").prepare().raw()
+/**
+ * @param {string} eventID
+ * @returns {Promise<boolean>} if true then the message did not arrive
+ */
+function waitForEvent(eventID) {
+	const {promise, resolve} = Promise.withResolvers()
+	waitFor(eventID, resolve, !!GET_EVENT_PREPARED.get(eventID))
+	return promise
+}
+
+const GET_MESSAGE_PREPARED = from("event_message").select("message_id").and("WHERE message_id = ?").prepare().raw()
+/**
+ * @param {string} messageID
+ * @returns {Promise<boolean>} if true then the message did not arrive
+ */
+function waitForMessage(messageID) {
+	const {promise, resolve} = Promise.withResolvers()
+	waitFor(messageID, resolve, !!GET_MESSAGE_PREPARED.get(messageID))
+	return promise
+}
+
+const GET_REACTION_EVENT_PREPARED = from("reaction").select("hashed_event_id").and("WHERE hashed_event_id = ?").prepare().raw()
+/**
+ * @param {string} eventID
+ * @returns {Promise<boolean>} if true then the message did not arrive
+ */
+function waitForReactionEvent(eventID) {
+	const {promise, resolve} = Promise.withResolvers()
+	waitFor(eventID, resolve, !!GET_REACTION_EVENT_PREPARED.get(utils.getEventIDHash(eventID)))
+	return promise
 }
 
 /**
  * Anything calling retrigger during the callback will be paused and retriggered after the callback resolves.
  * @template T
- * @param {string} messageID
+ * @param {string} id
  * @param {Promise<T>} promise
  * @returns {Promise<T>}
  */
-async function pauseChanges(messageID, promise) {
+async function pauseChanges(id, promise) {
 	try {
-		debugRetrigger(`[retrigger] PAUSE id = ${messageID}`)
-		paused.add(messageID)
+		storage.pause(id)
 		return await promise
 	} finally {
-		debugRetrigger(`[retrigger] RESUME id = ${messageID}`)
-		paused.delete(messageID)
-		messageFinishedBridging(messageID)
+		finishedBridging(id)
 	}
 }
 
 /**
  * Triggers any pending operations that were waiting on the corresponding event ID.
- * @param {string} messageID
+ * @param {string} id
  */
-function messageFinishedBridging(messageID) {
-	if (emitter.listeners(messageID).length) {
-		debugRetrigger(`[retrigger] EMIT id = ${messageID}`)
-	}
-	emitter.emit(messageID)
+function finishedBridging(id) {
+	storage.resolve(id, true)
 }
 
-module.exports.eventNotFoundThenRetrigger = eventNotFoundThenRetrigger
-module.exports.messageFinishedBridging = messageFinishedBridging
+module.exports.waitForMessage = waitForMessage
+module.exports.waitForEvent = waitForEvent
+module.exports.waitForReactionEvent = waitForReactionEvent
 module.exports.pauseChanges = pauseChanges
+module.exports.finishedBridging = finishedBridging

src/d2m/converters/remove-reaction.js

@@ -34,7 +34,7 @@ function removeReaction(data, reactions, key) {
 				// Even though the bridge bot only reacted once on Discord-side, multiple Matrix users may have
 				// reacted on Matrix-side. Semantically, we want to remove the reaction from EVERY Matrix user.
 				// Also need to clean up the database.
-				const hash = utils.getEventIDHash(event.event_id)
+				const hash = utils.getEventIDHash(eventID)
 				removals.push({eventID, mxid: null, hash})
 			}
 			if (!lookingAtMatrixReaction && !wantToRemoveMatrixReaction) {

src/d2m/event-dispatcher.js

@@ -2,6 +2,7 @@
 
 const assert = require("assert").strict
 const DiscordTypes = require("discord-api-types/v10")
+const {id: botID} = require("../../addbot")
 const {sync, db, select, from} = require("../passthrough")
 
 /** @type {import("./actions/send-message")}) */
@@ -38,6 +39,8 @@ const removeMember = sync.require("./actions/remove-member")
 const vote = sync.require("./actions/poll-vote")
 /** @type {import("../m2d/event-dispatcher")} */
 const matrixEventDispatcher = sync.require("../m2d/event-dispatcher")
+/** @type {import("../m2d/actions/redact.js")} */
+const redact = sync.require("../m2d/actions/redact.js")
 /** @type {import("../discord/interactions/matrix-info")} */
 const matrixInfoInteraction = sync.require("../discord/interactions/matrix-info")
 /** @type {import("../agi/listener")} */
@@ -321,7 +324,7 @@ module.exports = {
 		// @ts-ignore
 		await sendMessage.sendMessage(message, channel, guild, row)
 
-		retrigger.messageFinishedBridging(message.id)
+		retrigger.finishedBridging(message.id)
 	},
 
 	/**
@@ -342,7 +345,7 @@ module.exports = {
 
 		if (!row) {
 			// Check that the sending-to room exists, and deal with Eventual Consistency(TM)
-			if (retrigger.eventNotFoundThenRetrigger(data.id, module.exports.MESSAGE_UPDATE, client, data)) return
+			if (!await retrigger.waitForMessage(data.id)) return
 		}
 
 		/** @type {DiscordTypes.GatewayMessageCreateDispatchData} */
@@ -380,6 +383,16 @@ module.exports = {
 	 * @param {DiscordTypes.GatewayMessageReactionRemoveDispatchData | DiscordTypes.GatewayMessageReactionRemoveEmojiDispatchData | DiscordTypes.GatewayMessageReactionRemoveAllDispatchData} data
 	 */
 	async onSomeReactionsRemoved(client, data) {
+		// Don't attempt to double-bridge our own deleted reactions, this would go badly if there are race conditions
+		if ("user_id" in data && data.user_id === botID && data.emoji.name) { // sure hope data.emoji.name exists here
+			const encodedEmoji = encodeURIComponent(data.emoji.id ? `${data.emoji.name}:${data.emoji.id}` : data.emoji.name)
+			const i = redact.ourDeletedReactions.findIndex(x => data.message_id === x.message_id && encodedEmoji === x.encoded_emoji)
+			if (i !== -1) {
+				redact.ourDeletedReactions.splice(i, 1)
+				return
+			}
+		}
+
 		await removeReaction.removeSomeReactions(data)
 	},
 
@@ -389,7 +402,7 @@ module.exports = {
 	 */
 	async MESSAGE_DELETE(client, data) {
 		speedbump.onMessageDelete(data.id)
-		if (retrigger.eventNotFoundThenRetrigger(data.id, module.exports.MESSAGE_DELETE, client, data)) return
+		if (!await retrigger.waitForMessage(data.id)) return
 		await deleteMessage.deleteMessage(data)
 	},
 
@@ -437,12 +450,12 @@ module.exports = {
 	 * @param {DiscordTypes.GatewayMessagePollVoteDispatchData} data
 	 */
 	async MESSAGE_POLL_VOTE_ADD(client, data) {
-		if (retrigger.eventNotFoundThenRetrigger(data.message_id, module.exports.MESSAGE_POLL_VOTE_ADD, client, data)) return
+		if (!await retrigger.waitForMessage(data.message_id)) return
 		await vote.addVote(data)
 	},
 
 	async MESSAGE_POLL_VOTE_REMOVE(client, data) {
-		if (retrigger.eventNotFoundThenRetrigger(data.message_id, module.exports.MESSAGE_POLL_VOTE_REMOVE, client, data)) return
+		if (!await retrigger.waitForMessage(data.message_id)) return
 		await vote.removeVote(data)
 	},
 

src/db/orm.js

@@ -104,6 +104,16 @@ class From {
 		return r
 	}
 
+	pluckUnsafe(col) {
+		/** @type {Pluck<Table, any>} */
+		// @ts-ignore
+		const r = this
+		r.cols = [col]
+		r.makeColsSafe = false
+		r.isPluck = true
+		return r
+	}
+
 	/**
 	 * @param {string} sql
 	 */

src/db/orm.test.js

@@ -68,3 +68,8 @@ test("orm: select unsafe works (to select complex column names that can't be typ
 		.all()
 	t.equal(results[0].power_level, 150)
 })
+
+test("orm: pluck unsafe works (to select complex column names that can't be type verified)", t => {
+	const result = from("channel_room").where({guild_id: "112760669178241024"}).pluckUnsafe("count(*)").get()
+	t.equal(result, 7)
+})

src/m2d/actions/add-reaction.js

@@ -17,7 +17,7 @@ const retrigger = sync.require("../../d2m/actions/retrigger")
  */
 async function addReaction(event) {
 	// Wait until the corresponding channel and message have already been bridged
-	if (retrigger.eventNotFoundThenRetrigger(event.content["m.relates_to"].event_id, () => as.emit("type:m.reaction", event))) return
+	if (!await retrigger.waitForEvent(event.content["m.relates_to"].event_id)) return
 
 	// These will exist because it passed retrigger
 	const row = from("event_message").join("message_room", "message_id").join("historical_channel_room", "historical_room_index")
@@ -50,6 +50,8 @@ async function addReaction(event) {
 	}
 
 	db.prepare("REPLACE INTO reaction (hashed_event_id, message_id, encoded_emoji, original_encoding) VALUES (?, ?, ?, ?)").run(utils.getEventIDHash(event.event_id), messageID, discordPreferredEncoding, key)
+
+	retrigger.finishedBridging(event.event_id)
 }
 
 module.exports.addReaction = addReaction

src/m2d/actions/redact.js

@@ -10,6 +10,9 @@ const utils = sync.require("../../matrix/utils")
 /** @type {import("../../d2m/actions/retrigger")} */
 const retrigger = sync.require("../../d2m/actions/retrigger")
 
+/** @type {{message_id: string, encoded_emoji: string}[]} */
+const ourDeletedReactions = []
+
 /**
  * @param {Ty.Event.Outer_M_Room_Redaction} event
  */
@@ -24,6 +27,21 @@ async function deleteMessage(event) {
 	db.prepare("DELETE FROM message_room WHERE message_id = ?").run(rows[0].message_id)
 }
 
+/**
+ * @param {Ty.Event.Outer_M_Room_Redaction} event
+ */
+async function removeMessageEvent(event) {
+	// Could be for removing a message or suppressing embeds. For more information, the message needs to be bridged first.
+	if (!await retrigger.waitForEvent(event.redacts)) return
+
+	const row = select("event_message", ["event_type", "event_subtype", "part"], {event_id: event.redacts}).get()
+	if (row && row.event_type === "m.room.message" && row.event_subtype === "m.notice" && row.part === 1) {
+		await suppressEmbeds(event)
+	} else {
+		await deleteMessage(event)
+	}
+}
+
 /**
  * @param {Ty.Event.Outer_M_Room_Redaction} event
  */
@@ -41,11 +59,18 @@ async function suppressEmbeds(event) {
  * @param {Ty.Event.Outer_M_Room_Redaction} event
  */
 async function removeReaction(event) {
+	if (!await retrigger.waitForReactionEvent(event.redacts)) return
+
 	const hash = utils.getEventIDHash(event.redacts)
 	const row = from("reaction").join("message_room", "message_id").join("historical_channel_room", "historical_room_index")
 		.select("reference_channel_id", "message_id", "encoded_emoji").where({hashed_event_id: hash}).get()
 	if (!row) return
+	// See how many Matrix-side reactions there are, and delete if it's the last one
+	const numberOfReactions = from("reaction").where({message_id: row.message_id, encoded_emoji: row.encoded_emoji}).pluckUnsafe("count(*)").get()
+	if (numberOfReactions === 1) {
+		ourDeletedReactions.push(row)
 		await discord.snow.channel.deleteReactionSelf(row.reference_channel_id, row.message_id, row.encoded_emoji)
+	}
 	db.prepare("DELETE FROM reaction WHERE hashed_event_id = ?").run(hash)
 }
 
@@ -54,18 +79,12 @@ async function removeReaction(event) {
  * @param {Ty.Event.Outer_M_Room_Redaction} event
  */
 async function handle(event) {
-	// If this is for removing a reaction, try it
-	await removeReaction(event)
-
-	// Or, it might be for removing a message or suppressing embeds. But to do that, the message needs to be bridged first.
-	if (retrigger.eventNotFoundThenRetrigger(event.redacts, () => as.emit("type:m.room.redaction", event))) return
-
-	const row = select("event_message", ["event_type", "event_subtype", "part"], {event_id: event.redacts}).get()
-	if (row && row.event_type === "m.room.message" && row.event_subtype === "m.notice" && row.part === 1) {
-		await suppressEmbeds(event)
-	} else {
-		await deleteMessage(event)
-	}
+	// Don't know if it's a redaction for a reaction or an event, try both at the same time (otherwise waitFor will block)
+	await Promise.all([
+		removeMessageEvent(event),
+		removeReaction(event)
+	])
 }
 
 module.exports.handle = handle
+module.exports.ourDeletedReactions = ourDeletedReactions

src/m2d/event-dispatcher.js

@@ -211,7 +211,7 @@ async event => {
 		// @ts-ignore
 		await matrixCommandHandler.execute(event)
 	}
-	retrigger.messageFinishedBridging(event.event_id)
+	retrigger.finishedBridging(event.event_id)
 	await api.ackEvent(event)
 }))
 
@@ -222,7 +222,7 @@ sync.addTemporaryListener(as, "type:m.sticker", guard("m.sticker",
 async event => {
 	if (utils.eventSenderIsFromDiscord(event.sender)) return
 	const messageResponses = await sendEvent.sendEvent(event)
-	retrigger.messageFinishedBridging(event.event_id)
+	retrigger.finishedBridging(event.event_id)
 	await api.ackEvent(event)
 }))
 

src/stdin.js

@@ -15,12 +15,15 @@ const mreq = sync.require("./matrix/mreq")
 const api = sync.require("./matrix/api")
 const file = sync.require("./matrix/file")
 const sendEvent = sync.require("./m2d/actions/send-event")
+const redact = sync.require("./m2d/actions/redact")
 const eventDispatcher = sync.require("./d2m/event-dispatcher")
 const updatePins = sync.require("./d2m/actions/update-pins")
 const speedbump = sync.require("./d2m/actions/speedbump")
 const ks = sync.require("./matrix/kstate")
 const setPresence = sync.require("./d2m/actions/set-presence")
 const channelWebhook = sync.require("./m2d/actions/channel-webhook")
+const dUtils = sync.require("./discord/utils")
+const mxUtils = sync.require("./matrix/utils")
 const guildID = "112760669178241024"
 
 async function ping() {