Fix m->d reaction deletion counting (#85)

Fixes a bug where, if multiple Matrix users had used the same reaction on a message, and then one of those Matrix users removed their reactions, the bot would forcibly remove all of that reactions. Now, we check and make sure there are no remaining reactions from Matrix before removal.

This also rewrote the retrigger system to be more generic and to use promises instead of re-entry (would lose call stack).

Co-authored-by: Cadence Ember <cadence@disroot.org>
Reviewed-on: https://gitdab.com/cadence/out-of-your-element/pulls/85

src/d2m/actions/retrigger.js

@@ -2,7 +2,15 @@
 
 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")
+
+/*
+	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 = false
 
@@ -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()
 
-/**
- * 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
- */
-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
-			}
+	/**
+	 * 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)
+	}
 
-	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}`)
+	/** @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)
 		}
-		emitter.removeAllListeners(inputID)
-	}, 60 * 1000) // 1 minute
-	return true // event was not found, then retrigger
+
+		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)
+		}
+	}
+}
+
+/**
+ * @param {string} id
+ * @param {(found: Boolean) => any} resolve
+ * @param {boolean} existsInDatabase
+ */
+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)
+	}
+
+	// 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")
 
@@ -316,7 +319,7 @@ module.exports = {
 		// @ts-ignore
 		await sendMessage.sendMessage(message, channel, guild, row)
 
-		retrigger.messageFinishedBridging(message.id)
+		retrigger.finishedBridging(message.id)
 	},
 
 	/**
@@ -337,7 +340,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} */
@@ -375,6 +378,16 @@ module.exports = {
 	 * @param {DiscordTypes.GatewayMessageReactionRemoveDispatchData | DiscordTypes.GatewayMessageReactionRemoveEmojiDispatchData | DiscordTypes.GatewayMessageReactionRemoveAllDispatchData} data
 	 */
 	async onSomeReactionsRemoved(client, data) {
+		// Don't attempt to double-bridge our own m2d deleted reactions back to Matrix
+		if ("user_id" in data && data.user_id === botID) {
+			const emojiIdOrName = data.emoji.id || data.emoji.name
+			const i = redact.m2dDeletedReactions.findIndex(x => data.message_id === x.messageID && emojiIdOrName === x.emojiIdOrName)
+			if (i !== -1) {
+				redact.m2dDeletedReactions.splice(i, 1)
+				return
+			}
+		}
+
 		await removeReaction.removeSomeReactions(data)
 	},
 
@@ -384,7 +397,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)
 	},
 
@@ -432,12 +445,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)
 	},