Retrigger m->d reactions and removals

src/d2m/actions/retrigger.js

@@ -20,31 +20,39 @@ const emitter = new EventEmitter()
  * (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[]) => Promise<any>} T
- * @param {string} messageID
+ * @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(messageID, fn, ...rest) {
+function eventNotFoundThenRetrigger(inputID, fn, ...rest) {
-	if (!paused.has(messageID)) {
+	if (!paused.has(inputID)) {
-		const eventID = select("event_message", "event_id", {message_id: messageID}).pluck().get()
+		if (inputID.match(/^[0-9]+$/)) {
+			const eventID = select("event_message", "event_id", {message_id: inputID}).pluck().get()
 			if (eventID) {
-			debugRetrigger(`[retrigger] OK mid <-> eid = ${messageID} <-> ${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
+			}
+		}
 	}
 
-	debugRetrigger(`[retrigger] WAIT mid = ${messageID}`)
+	debugRetrigger(`[retrigger] WAIT id = ${inputID}`)
-	emitter.once(messageID, () => {
+	emitter.once(inputID, () => {
-		debugRetrigger(`[retrigger] TRIGGER mid = ${messageID}`)
+		debugRetrigger(`[retrigger] TRIGGER id = ${inputID}`)
 		fn(...rest)
 	})
 	// if the event never arrives, don't trigger the callback, just clean up
 	setTimeout(() => {
-		if (emitter.listeners(messageID).length) {
+		if (emitter.listeners(inputID).length) {
-			debugRetrigger(`[retrigger] EXPIRE mid = ${messageID}`)
+			debugRetrigger(`[retrigger] EXPIRE id = ${inputID}`)
 		}
-		emitter.removeAllListeners(messageID)
+		emitter.removeAllListeners(inputID)
 	}, 60 * 1000) // 1 minute
 	return true // event was not found, then retrigger
 }
@@ -58,11 +66,11 @@ function eventNotFoundThenRetrigger(messageID, fn, ...rest) {
  */
 async function pauseChanges(messageID, promise) {
 	try {
-		debugRetrigger(`[retrigger] PAUSE mid = ${messageID}`)
+		debugRetrigger(`[retrigger] PAUSE id = ${messageID}`)
 		paused.add(messageID)
 		return await promise
 	} finally {
-		debugRetrigger(`[retrigger] RESUME mid = ${messageID}`)
+		debugRetrigger(`[retrigger] RESUME id = ${messageID}`)
 		paused.delete(messageID)
 		messageFinishedBridging(messageID)
 	}
@@ -74,7 +82,7 @@ async function pauseChanges(messageID, promise) {
  */
 function messageFinishedBridging(messageID) {
 	if (emitter.listeners(messageID).length) {
-		debugRetrigger(`[retrigger] EMIT mid = ${messageID}`)
+		debugRetrigger(`[retrigger] EMIT id = ${messageID}`)
 	}
 	emitter.emit(messageID)
 }

src/m2d/actions/add-reaction.js

@@ -4,20 +4,27 @@ const assert = require("assert").strict
 const Ty = require("../../types")
 
 const passthrough = require("../../passthrough")
-const {discord, sync, db, select} = passthrough
+const {discord, as, sync, db, select, from} = passthrough
 /** @type {import("../../matrix/utils")} */
 const utils = sync.require("../../matrix/utils")
 /** @type {import("../converters/emoji")} */
 const emoji = sync.require("../converters/emoji")
+/** @type {import("../../d2m/actions/retrigger")} */
+const retrigger = sync.require("../../d2m/actions/retrigger")
 
 /**
  * @param {Ty.Event.Outer<Ty.Event.M_Reaction>} event
  */
 async function addReaction(event) {
-	const channelID = select("historical_channel_room", "reference_channel_id", {room_id: event.room_id}).pluck().get()
+	// Wait until the corresponding channel and message have already been bridged
-	if (!channelID) return // We just assume the bridge has already been created
+	if (retrigger.eventNotFoundThenRetrigger(event.content["m.relates_to"].event_id, as.emit.bind(as, "type:m.reaction", event))) return
-	const messageID = select("event_message", "message_id", {event_id: event.content["m.relates_to"].event_id}, "ORDER BY reaction_part").pluck().get()
+
-	if (!messageID) return // Nothing can be done if the parent message was never bridged.
+	// These will exist because it passed retrigger
+	const row = from("event_message").join("message_room", "message_id").join("historical_channel_room", "historical_room_index")
+		.select("message_id", "reference_channel_id").where({event_id: event.content["m.relates_to"].event_id}).and("ORDER BY reaction_part ASC").get()
+	assert(row)
+	const messageID = row.message_id
+	const channelID = row.reference_channel_id
 
 	const key = event.content["m.relates_to"].key
 	const discordPreferredEncoding = await emoji.encodeEmoji(key, event.content.shortcode)
@@ -35,6 +42,10 @@ async function addReaction(event) {
 			// happens if a matrix user tries to add on to a super reaction
 			return
 		}
+		if (e.message?.includes("Unknown Message")) {
+			// happens under a race condition where a message is deleted after it passes the database check above
+			return
+		}
 		throw e
 	}
 

src/m2d/actions/redact.js

@@ -4,9 +4,11 @@ const DiscordTypes = require("discord-api-types/v10")
 const Ty = require("../../types")
 
 const passthrough = require("../../passthrough")
-const {discord, sync, db, select, from} = passthrough
+const {discord, as, sync, db, select, from} = passthrough
 /** @type {import("../../matrix/utils")} */
 const utils = sync.require("../../matrix/utils")
+/** @type {import("../../d2m/actions/retrigger")} */
+const retrigger = sync.require("../../d2m/actions/retrigger")
 
 /**
  * @param {Ty.Event.Outer_M_Room_Redaction} event
@@ -52,13 +54,18 @@ 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.bind(as, "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)
 	}
-	await removeReaction(event)
 }
 
 module.exports.handle = handle

src/m2d/event-dispatcher.js

@@ -28,6 +28,8 @@ const api = sync.require("../matrix/api")
 const createRoom = sync.require("../d2m/actions/create-room")
 /** @type {import("../matrix/room-upgrade")} */
 const roomUpgrade = require("../matrix/room-upgrade")
+/** @type {import("../d2m/actions/retrigger")} */
+const retrigger = sync.require("../d2m/actions/retrigger")
 const {reg} = require("../matrix/read-registration")
 
 let lastReportedEvent = 0
@@ -201,6 +203,7 @@ async event => {
 		// @ts-ignore
 		await matrixCommandHandler.execute(event)
 	}
+	retrigger.messageFinishedBridging(event.event_id)
 	await api.ackEvent(event)
 }))
 
@@ -211,6 +214,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)
 	await api.ackEvent(event)
 }))