Hello thread PR, my old friend...

I've come to fix conflicts, and then pretend you don't exist, again.

scripts/http-interrupt-test-client.js

@@ -0,0 +1,31 @@
+// @ts-check
+
+const crypto = require("crypto")
+const fs = require("fs")
+const assert = require("assert")
+const stream = require("stream")
+const {scheduler} = require("timers/promises")
+
+;(async () => {
+	try {
+		const res = await fetch("http://localhost:6694", {
+			keepalive: true
+		})
+		assert(res.body)
+
+		// const data = await res.blob().then(blob => blob.bytes())
+		// console.log(data)
+
+		const c = crypto.createCipheriv("aes-256-ctr", Buffer.from("oKInhenybApUF-X4ahfQi0dIql8fyTXDVjfiEZvJquE", "base64url"), Buffer.from("Qt4ZM5g1zDQAAAAAAAAAAA", "base64url"))
+		// @ts-ignore - close enough
+		// const nodeBody = stream.Readable.fromWeb(res.body)
+		console.log(await new Response(res.body.pipeThrough(stream.Transform.toWeb(c))).bytes())
+
+
+		console.log("all done")
+	} catch (e) {
+		console.error(e)
+	}
+
+	console.log("didn't crash")
+})()

scripts/http-interrupt-test-server.js

@@ -0,0 +1,63 @@
+// @ts-check
+
+const net = require("net")
+const stream = require("stream")
+const {scheduler} = require("timers/promises")
+
+const CHUNKED = true
+const REPEAT_COUNT = 10
+
+const b = new Uint8Array(1024).fill(0x65)
+
+async function* createData() {
+	for (let i = 0; i < REPEAT_COUNT; i++) {
+		console.log(`generated chunk ${i}`)
+		if (CHUNKED) {
+			yield b.length.toString(16)
+			yield "\r\n"
+		}
+		yield b
+		if (CHUNKED) {
+			yield "\r\n"
+		}
+		await scheduler.wait(800)
+	}
+	if (CHUNKED) {
+		yield "0\r\n\r\n"
+	}
+}
+
+const server = net.createServer(socket => {
+	socket.write(`HTTP/1.1 200 OK\r\n`)
+	socket.write("Connection: keep-alive\r\n")
+	if (CHUNKED) {
+		socket.write("Transfer-Encoding: chunked\r\n")
+	} else {
+		socket.write(`Content-Length: ${b.length*REPEAT_COUNT}\r\n`)
+	}
+	socket.write(`\r\n`)
+	const dataStream = stream.Readable.from(createData())
+	dataStream.pipe(socket)
+
+	// In case of problem with dataStream
+	dataStream.on("error", () => {
+		console.error("aw dang it")
+		socket.end()
+	})
+
+	// In case of problem with recipient
+	socket.on("close", () => {
+		console.log("socket close")
+		dataStream.destroy()
+	})
+
+	socket.on("error", () => {
+		console.log("socket error")
+	})
+
+	socket.on("end", () => {
+		console.log("socket end")
+	})
+})
+
+server.listen(6694)

src/d2m/actions/create-room.js

@@ -506,8 +506,17 @@ async function unbridgeChannel(channel, guildID, messageBeforeLeave = "This room
 	const powerLevelsUsers = (powerLevels.users ||= {})
 	for (const mxid of Object.keys(powerLevelsUsers)) {
 		if (powerLevelsUsers[mxid] >= (powerLevels.state_default ?? 50) && !allCreators.includes(mxid) && mUtils.eventSenderIsFromDiscord(mxid) && mxid !== mUtils.bot) {
-			delete powerLevelsUsers[mxid]
+			// can OOYE bot demote them?
-			await api.sendState(roomID, "m.room.power_levels", "", powerLevels, mxid) // done individually because each user must demote themselves
+			if (allCreators.includes(mUtils.bot) || powerLevelsUsers[mxid] < powerLevelsUsers[mUtils.bot]) {
+				delete powerLevelsUsers[mxid]
+				await api.sendState(roomID, "m.room.power_levels", "", powerLevels) // bot can demote them
+			} else {
+				// check user is still in room and can send events
+				if (select("sim_member", "mxid", {mxid, room_id: roomID}).get()) {
+					delete powerLevelsUsers[mxid]
+					await api.sendState(roomID, "m.room.power_levels", "", powerLevels, mxid) // done individually because each user must demote themselves
+				}
+			}
 		}
 	}
 

src/d2m/converters/message-to-event.js

@@ -79,8 +79,13 @@ function getDiscordParseCallbacks(message, guild, useHTML, spoilers = []) {
 			const text = discordTime.convertDiscordTime(node.timestamp, node.style)
 			if (useHTML) {
 				let url = `${reg.ooye.bridge_origin}/time/${node.timestamp}`
-				if (node.style) url += `/${node.style}`
+				let dataStyle = ""
-				return tag`<a href="${url}">${text}</a>`
+				if (node.style) {
+					url += `/${node.style}`
+					dataStyle = tag`data-style="${node.style}" `
+				}
+				const iso = new Date(node.timestamp * 1000).toISOString()
+				return tag`<a href="${url}"><time $${dataStyle}datetime="${iso}">${text}</time></a>`
 			} else {
 				return text
 			}

src/d2m/converters/message-to-event.test.components.js

@@ -96,7 +96,7 @@ test("message2event components: pk question mark output", async t => {
 			+ "<hr>"
 			+ "<p>Same hat</p>"
 			+ `🖼️ Image: <a href="https://bridge.example.org/download/discordcdn/934955898965729280/1466556006527012987/image.png">image.png</a></blockquote>`
-			+ "<p><sub>Original Message ID: 1466556003645657118 · <a href=\"https://bridge.example.org/time/1769724599/f\">29 January 2026 at 10:09 pm (UTC)</a></sub></p>",
+			+ "<p><sub>Original Message ID: 1466556003645657118 · <a href=\"https://bridge.example.org/time/1769724599/f\"><time data-style=\"f\" datetime=\"2026-01-29T22:09:59.000Z\">29 January 2026 at 10:09 pm (UTC)</time></a></sub></p>",
 		"m.mentions": {},
 		msgtype: "m.text",
 	}])

src/d2m/converters/message-to-event.test.js

@@ -214,12 +214,12 @@ test("message2event: timestamps", async t => {
 			+ "\n3 October 2026 at 9:30 am (UTC)"
 			+ "\n24 December 2026 at 11:00 pm (UTC)",
 		format: "org.matrix.custom.html",
-		formatted_body: `<a href="https://bridge.example.org/time/1791106413/S">4 Oct 2026, 9:33:33 am (UTC)</a>`
+		formatted_body: `<a href="https://bridge.example.org/time/1791106413/S"><time data-style="S" datetime="2026-10-04T09:33:33.000Z">4 Oct 2026, 9:33:33 am (UTC)</time></a>`
-			+ `<br><a href="https://bridge.example.org/time/1791106436/f">4 October 2026 at 9:33 am (UTC)</a>`
+			+ `<br><a href="https://bridge.example.org/time/1791106436/f"><time data-style="f" datetime="2026-10-04T09:33:56.000Z">4 October 2026 at 9:33 am (UTC)</time></a>`
-			+ `<br><a href="https://bridge.example.org/time/1791106440/F">Sunday, 4 October 2026 at 9:34 am (UTC)</a>`
+			+ `<br><a href="https://bridge.example.org/time/1791106440/F"><time data-style="F" datetime="2026-10-04T09:34:00.000Z">Sunday, 4 October 2026 at 9:34 am (UTC)</time></a>`
-			+ `<br><a href="https://bridge.example.org/time/1791106448/T">9:34:08 am (UTC)</a>`
+			+ `<br><a href="https://bridge.example.org/time/1791106448/T"><time data-style="T" datetime="2026-10-04T09:34:08.000Z">9:34:08 am (UTC)</time></a>`
-			+ `<br><a href="https://bridge.example.org/time/1791019800/R">3 October 2026 at 9:30 am (UTC)</a>`
+			+ `<br><a href="https://bridge.example.org/time/1791019800/R"><time data-style="R" datetime="2026-10-03T09:30:00.000Z">3 October 2026 at 9:30 am (UTC)</time></a>`
-			+ `<br><a href="https://bridge.example.org/time/1798153200/R">24 December 2026 at 11:00 pm (UTC)</a>`,
+			+ `<br><a href="https://bridge.example.org/time/1798153200/R"><time data-style="R" datetime="2026-12-24T23:00:00.000Z">24 December 2026 at 11:00 pm (UTC)</time></a>`,
 		"m.mentions": {}
 	}])
 })

src/d2m/discord-packets.js

@@ -221,7 +221,7 @@ async function onPacket(client, message, listen) {
 			// when not real time, skip optional events
 		} else {
 			// queue required events
-			homeserverStatus.homeserverStatus.queuePacket(message)
+			homeserverStatus.homeserverStatus.queuePacket(["Discord", message])
 		}
 	}
 }

src/m2d/actions/channel-webhook.js

@@ -57,7 +57,7 @@ async function withWebhook(channelID, callback) {
 
 /**
  * @param {string} channelID
- * @param {DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | stream.Readable}[]}} data
+ * @param {DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | Blob}[]}} data
  * @param {string} [threadID]
  */
 async function sendMessageWithWebhook(channelID, data, threadID) {
@@ -70,7 +70,7 @@ async function sendMessageWithWebhook(channelID, data, threadID) {
 /**
  * @param {string} channelID
  * @param {string} messageID
- * @param {DiscordTypes.RESTPatchAPIWebhookWithTokenMessageJSONBody & {files?: {name: string, file: Buffer | stream.Readable}[]}} data
+ * @param {DiscordTypes.RESTPatchAPIWebhookWithTokenMessageJSONBody & {files?: {name: string, file: Buffer | Blob}[]}} data
  * @param {string} [threadID]
  */
 async function editMessageWithWebhook(channelID, messageID, data, threadID) {

src/m2d/actions/send-event.js

@@ -28,8 +28,8 @@ const emojiSheet = sync.require("../actions/emoji-sheet")
 const latestSenderPerRoom = sync.remember(() => new Map())
 
 /**
- * @param {DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | stream.Readable}[], pendingFiles?: ({name: string, mxc: string} | {name: string, mxc: string, key: string, iv: string} | {name: string, buffer: Buffer | stream.Readable})[]}} message
+ * @param {DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | Blob}[], pendingFiles?: ({name: string, mxc: string} | {name: string, mxc: string, key: string, iv: string} | {name: string, buffer: Buffer | Blob})[]}} message
- * @returns {Promise<DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | stream.Readable}[]}>}
+ * @returns {Promise<DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | Blob}[]}>}
  */
 async function resolvePendingFiles(message) {
 	if (!message.pendingFiles) return message
@@ -42,24 +42,22 @@ async function resolvePendingFiles(message) {
 		}
 		if ("key" in p) {
 			// Encrypted file
+			const res = await api.getMedia(p.mxc)
 			const d = crypto.createDecipheriv("aes-256-ctr", Buffer.from(p.key, "base64url"), Buffer.from(p.iv, "base64url"))
-			await api.getMedia(p.mxc).then(res => stream.Readable.fromWeb(
+			/** @type {ReadableStream} */ // @ts-ignore - close enough
-				// @ts-ignore
+			const piped = res.body.pipeThrough(stream.Transform.toWeb(d))
-				res.body
+			const blob = await new Response(piped).blob()
-			).pipe(d))
 			return {
 				name: p.name,
-				file: d
+				file: blob
 			}
 		} else {
 			// Unencrypted file
-			const body = await api.getMedia(p.mxc).then(res => stream.Readable.fromWeb(
+			const res = await api.getMedia(p.mxc)
-				// @ts-ignore
+			const blob = await res.blob()
-				res.body
-			))
 			return {
 				name: p.name,
-				file: body
+				file: blob
 			}
 		}
 	}))

src/m2d/actions/sticker.js

@@ -1,6 +1,6 @@
 // @ts-check
 
-const {Readable} = require("stream")
+const {Transform} = require("stream")
 const {ReadableStream} = require("stream/web")
 
 const {sync} = require("../../passthrough")
@@ -9,7 +9,6 @@ const sharp = require("sharp")
 const api = sync.require("../../matrix/api")
 /** @type {import("../../matrix/mreq")} */
 const mreq = sync.require("../../matrix/mreq")
-const {streamType} = require("@cloudrac3r/stream-type")
 
 const WIDTH = 160
 const HEIGHT = 160
@@ -20,20 +19,14 @@ const HEIGHT = 160
  */
 async function getAndResizeSticker(mxc) {
 	const res = await api.getMedia(mxc)
-	if (res.status !== 200) {
-		const root = await res.json()
-		throw new mreq.MatrixServerError(root, res.status, {mxc})
-	}
 
-	const streamIn = Readable.fromWeb(res.body)
+	return res.body.pipeThrough(
-	const {streamThrough, type} = await streamType(streamIn)
+		Transform.toWeb(
-	const animated = ["image/gif", "image/webp"].includes(type)
+			sharp({animated: true})
-
+				.resize(WIDTH, HEIGHT, {fit: "inside", background: {r: 0, g: 0, b: 0, alpha: 0}})
-	const transformer = sharp({animated: animated})
+				.webp()
-		.resize(WIDTH, HEIGHT, {fit: "inside", background: {r: 0, g: 0, b: 0, alpha: 0}})
+		)
-		.webp()
+	)
-	streamThrough.pipe(transformer)
-	return Readable.toWeb(transformer)
 }
 
 

src/m2d/converters/event-to-message.js

@@ -1081,7 +1081,7 @@ async function eventToMessage(event, guild, channel, di) {
 	}
 
 	// Assemble chunks into Discord messages content
-	/** @type {(DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | stream.Readable}[]})[]} */
+	/** @type {(DiscordTypes.RESTPostAPIWebhookWithTokenJSONBody & {files?: {name: string, file: Buffer | Blob}[]})[]} */
 	const messages = chunks.map(content => ({
 		content,
 		allowed_mentions,

src/m2d/event-dispatcher.js

@@ -129,7 +129,7 @@ async function sendError(roomID, source, type, e, payload) {
 			|| (e instanceof mreq.MatrixServerError && [502, 503].includes(e.httpStatus) && e.errcode === "CX_SERVER_ERROR")
 		) {
 			// Matrix homeserver is down (reverse proxy indicated failure)
-			homeserverStatus.homeserverStatus.setErrorWithPacket(payload)
+			homeserverStatus.homeserverStatus.setErrorWithPacket([source, payload])
 		}
 	}
 }
@@ -296,7 +296,7 @@ async event => {
 		// Try to bridge a failed event again?
 		await onRetryReactionAdd(event)
 	} else {
-		matrixCommandHandler.onReactionAdd(event)
+		if (matrixCommandHandler.onReactionAdd(event)) return
 		await addReaction.addReaction(event)
 	}
 }))

src/matrix/homeserver-status.js

@@ -54,6 +54,7 @@ const homeserverStatus = new class HomeserverStatus {
 
 			.defineState("offline", {
 				onEnter: [() => {
+					console.log("[homeserver status] Offline. The failed event and further events will be queued.")
 					this.pingInterval = setInterval(async () => {
 						const pingResult = await api.ping().catch(e => ({ok: false, status: "net", root: e.message}))
 						this.lastCheck = Date.now()
@@ -70,13 +71,22 @@ const homeserverStatus = new class HomeserverStatus {
 
 			.defineState("recovering", {
 				onEnter: [async () => { // Drain queue.
-					while (!this.queue.isEmpty()) {
+					if (!this.queue.isEmpty()) {
-						const packet = this.queue.peekFront() // same position as .shift()
+						console.log(`[homeserver status] Online, playing ${this.queue.length} events from queue...`)
-						debugHomeserverStatus(`homeserver status: ${new Date().toISOString()} dq packet ${packet.t} ${packet.d?.content}`)
+						while (!this.queue.isEmpty()) {
-						await discordPackets.dispatchPacketToBridge(passthrough.discord, packet)
+							const [source, packet] = this.queue.peekFront() // same position as .shift()
-						if (this.sm.currentStateName !== "recovering") return // got kicked out due to another error
+							debugHomeserverStatus(`homeserver status: ${new Date().toISOString()} dq sourceAndPacket ${packet.t} ${packet.d?.content}`)
-						this.queue.shift()
+							if (source === "Discord") {
+								await discordPackets.dispatchPacketToBridge(passthrough.discord, packet)
+							} else {
+								// Unfortunately cannot await this, just have to trust it
+								passthrough.as.emit(`type:${packet.type}`, packet)
+							}
+							if (this.sm.currentStateName !== "recovering") return // got kicked out due to another error
+							this.queue.shift()
+						}
 					}
+					console.log("[homeserver status] Back to real-time operation.")
 					this.sm.doTransition("recovered")
 				}],
 				onLeave: [],
@@ -88,7 +98,7 @@ const homeserverStatus = new class HomeserverStatus {
 			.defineTransition("recovering", "recovered", "online")
 			.defineTransition("online", "check assume online", "checking")
 			.defineTransition("online", "check assume offline", "checking")
-			.defineTransition("checking", "check ok", "recovering")
+			.defineTransition("checking", "check ok", "online")
 			.defineTransition("checking", "check fail", "offline")
 
 		this.sm.on("enter", st => debugHomeserverStatus(`homeserver status: ${st}`))
@@ -129,16 +139,16 @@ const homeserverStatus = new class HomeserverStatus {
 	/**
 	 * When offline or recovering, call this for incoming packets to queue them to be sent in order later.
 	 */
-	queuePacket(packet) {
+	queuePacket(sourceAndPacket) {
 		assert(["offline", "recovering"].includes(this.sm.currentStateName))
-		this.queue.push(packet)
+		this.queue.push(sourceAndPacket)
 	}
 
-	setErrorWithPacket(packet) {
+	setErrorWithPacket(sourceAndPacket) {
 		const wasRecovering = this.sm.currentStateName === "recovering"
 		this.sm.doTransition("error")
-		if (!wasRecovering) { // if was recovering then packet is already in the right place in queue
+		if (!wasRecovering) { // if was recovering then sourceAndPacket is already in the right place in queue
-			this.queuePacket(packet)
+			this.queuePacket(sourceAndPacket)
 		}
 	}
 }

src/matrix/matrix-command-handler.js

@@ -222,7 +222,7 @@ const commands = [{
 						b.add("", `<img data-mx-emoticon height="48" src="${e.url}" title=":${e.name}:" alt=":${e.name}:">`)
 					}
 					await api.sendState(event.room_id, type, key, pack)
-					api.sendEvent(event.room_id, "m.room.message", {
+					await api.sendEvent(event.room_id, "m.room.message", {
 						...ctx,
 						...b.get()
 					})
@@ -238,7 +238,7 @@ const commands = [{
 						console.log(`uploading emoji ${resizeOutput.data.length} bytes to :${e.name}:`)
 						await discord.snow.assets.createGuildEmoji(guildID, {name: e.name, image: "data:image/png;base64," + resizeOutput.data.toString("base64")})
 					}
-					api.sendEvent(event.room_id, "m.room.message", {
+					await api.sendEvent(event.room_id, "m.room.message", {
 						...ctx,
 						msgtype: "m.text",
 						body: `Created ${toUpload.length} emojis`

src/matrix/mreq.js

@@ -77,6 +77,10 @@ async function mreq(method, url, bodyIn, extra = {}) {
 	}, extra)
 
 	const res = await fetch(baseUrl + url, opts)
+	if (!res.ok) {
+		throw await makeMatrixServerError(res, {baseUrl, url, ...opts})
+	}
+
 	const text = await res.text()
 	try {
 		/** @type {any} */

src/web/pug/time.pug

@@ -18,10 +18,10 @@ html(lang="en")
         max-width: !{WIDTH}px;
       }
       .only-script {
-        display: none;
+        visibility: hidden;
       }
       .yes-script .only-script {
-        display: revert;
+        visibility: visible;
       }
       .local {
         line-height: 1;

src/web/routes/download-matrix.js

@@ -4,7 +4,7 @@ const assert = require("assert/strict")
 const {defineEventHandler, getValidatedRouterParams, setResponseStatus, setResponseHeader, createError, H3Event, getValidatedQuery} = require("h3")
 const {z} = require("zod")
 const {ReadableStream} = require("stream/web")
-const {Readable} = require("stream")
+const {Transform} = require("stream")
 const sharp = require("sharp")
 
 /** @type {import("xxhash-wasm").XXHashAPI} */ // @ts-ignore
@@ -26,8 +26,8 @@ const sticker = sync.require("../../m2d/actions/sticker")
 /** @type {{[presetKey: string]: (body: ReadableStream) => ReadableStream}} */
 const MEDIA_THUMBNAIL_PRESETS = {
 	avatar: body =>
-		Readable.toWeb(
+		body.pipeThrough(
-			Readable.fromWeb(body).pipe(
+			Transform.toWeb(
 				sharp()
 				.resize({height: 210, width: 210, fit: "cover"}) // the largest display of the webhook pfp on Discord Android in screen pixels
 				.jpeg({force: false, quality: 90}) // File size works out to up to ~110k for a PNG, less for a JPEG