N0VA/out-of-your-element
0
1
Fork 0
Code Issues Pull Requests Releases Activity

Queue events for later if homeserver is down

Browse Source
This commit is contained in:
Cadence Ember
2026-07-12 02:12:53 +12:00
parent 8779b8b1b0
commit 414da4caf4
6 changed files with 290 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/developer-orientation.md
+1
View File
@@ -113,6 +113,7 @@ Total transitive production dependencies: 144
* (0) ansi-colors: Helps with interactive prompting for the initial setup, and it's already pulled in by enquirer.
* (1) chunk-text: It does what I want.
* (0) cloudstorm: Discord gateway library with bring-your-own-caching that I trust.
* (0) denque: Much faster than using arrays as a queue.
* (0) discord-api-types: Bitfields needed at runtime and types needed for development.
* (0) domino: DOM implementation that's already pulled in by turndown.
* (2) enquirer: Interactive prompting for the initial setup rather than forcing users to edit YAML non-interactively.
src/d2m/actions/create-space.js
+6
View File
@@ -8,6 +8,8 @@ const {reg} = require("../../matrix/read-registration")
const passthrough = require("../../passthrough")
const {discord, sync, db, select} = passthrough
/** @type {import("../../matrix/mreq")} */
const mreq = sync.require("../../matrix/mreq")
/** @type {import("../../matrix/api")} */
const api = sync.require("../../matrix/api")
/** @type {import("../../matrix/file")} */
@@ -237,8 +239,12 @@ async function syncSpaceExpressions(data, checkBeforeSync) {
try {
existing = await api.getStateEvent(spaceID, "im.ponies.room_emotes", eventKey)
} catch (e) {
if (e instanceof mreq.MatrixServerError && e.httpStatus < 400) {
// State event not found. This space doesn't have any existing emojis. We create a dummy empty event for comparison's sake.
existing = fn([], guild)
} else {
throw e
}
}
if (isDeepStrictEqual(existing, content)) return
}
src/d2m/discord-packets.js
+31 -10
View File
@@ -1,18 +1,18 @@
// @ts-check
// Discord library internals type beat
const DiscordTypes = require("discord-api-types/v10")
const assert = require("assert")
const passthrough = require("../passthrough")
const {sync, db} = passthrough
const {sync} = passthrough
const utils = {
/**
/** @type {import("../matrix/homeserver-status")} */
const homeserverStatus = sync.require("../matrix/homeserver-status")
/**
* @param {import("./discord-client")} client
* @param {import("cloudstorm").IGatewayMessage} message
* @param {string} listen "full", "half", "no" - whether to set up the event listeners for OOYE to operate
*/
async onPacket(client, message, listen) {
async function onPacket(client, message, listen) {
// requiring this later so that the client is already constructed by the time event-dispatcher is loaded
/** @type {typeof import("./event-dispatcher")} */
const eventDispatcher = sync.require("./event-dispatcher")
@@ -152,6 +152,28 @@ const utils = {
// Event dispatcher for OOYE bridge operations
if (listen === "full" && message.t) {
const alwaysRealTimeEvents = ["PRESENCE_UPDATE"]
if (alwaysRealTimeEvents.includes(message.t) || homeserverStatus.homeserverStatus.isRealTime()) {
dispatchPacketToBridge(client, message)
} else {
homeserverStatus.homeserverStatus.queuePacket(message)
}
}
}
/**
* @param {import("./discord-client")} client
* @param {import("cloudstorm").IGatewayMessage} message
*/
async function dispatchPacketToBridge(client, message) {
// requiring this later so that the client is already constructed by the time event-dispatcher is loaded
/** @type {typeof import("./event-dispatcher")} */
const eventDispatcher = sync.require("./event-dispatcher")
/** @type {import("../discord/register-interactions")} */
const interactions = sync.require("../discord/register-interactions")
assert(message.t) // checked above
try {
if (message.t === "MESSAGE_REACTION_REMOVE" || message.t === "MESSAGE_REACTION_REMOVE_EMOJI" || message.t === "MESSAGE_REACTION_REMOVE_ALL") {
await eventDispatcher.onSomeReactionsRemoved(client, message.d)
@@ -166,8 +188,7 @@ const utils = {
// Let OOYE try to handle errors too
await eventDispatcher.onError(client, e, message)
}
}
}
}
module.exports = utils
module.exports.onPacket = onPacket
module.exports.dispatchPacketToBridge = dispatchPacketToBridge
src/m2d/event-dispatcher.js
+10 -1
View File
@@ -24,6 +24,8 @@ const vote = sync.require("./actions/vote")
const matrixCommandHandler = sync.require("../matrix/matrix-command-handler")
/** @type {import("../matrix/utils")} */
const utils = sync.require("../matrix/utils")
/** @type {import("../matrix/mreq")}) */
const mreq = sync.require("../matrix/mreq")
/** @type {import("../matrix/api")}) */
const api = sync.require("../matrix/api")
/** @type {import("../d2m/actions/create-room")} */
@@ -32,6 +34,8 @@ const createRoom = sync.require("../d2m/actions/create-room")
const roomUpgrade = require("../matrix/room-upgrade")
/** @type {import("../d2m/actions/retrigger")} */
const retrigger = sync.require("../d2m/actions/retrigger")
/** @type {import("../matrix/homeserver-status")} */
const homeserverStatus = sync.require("../matrix/homeserver-status")
const {reg} = require("../matrix/read-registration")
let lastReportedEvent = 0
@@ -166,7 +170,12 @@ async function sendError(roomID, source, type, e, payload) {
key: "🔁"
}
})
} catch (e) {}
} catch (e) {
if (e instanceof mreq.MatrixServerError && [502, 503].includes(e.httpStatus)) {
// Matrix homeserver is down (reverse proxy indicated failure; synapse doesn't generate 502/503 when posting an event to a room)
homeserverStatus.homeserverStatus.setErrorWithPacket(payload)
}
}
}
function guard(type, fn) {
src/matrix/api.js
+2 -1
View File
@@ -452,7 +452,8 @@ async function ping() {
headers: {
Authorization: `Bearer ${reg.as_token}`
},
body: "{}"
body: "{}",
signal: AbortSignal.timeout(15e3)
})
const root = await res.json()
return {
src/matrix/homeserver-status.js
+94
View File
@@ -0,0 +1,94 @@
// @ts-check
const assert = require("assert")
const Denque = require("denque")
const StateMachine = require("snowtransfer").StateMachine
const passthrough = require("../passthrough")
const {sync} = passthrough
/** @type {import("../d2m/discord-packets")} */
const discordPackets = sync.require("../d2m/discord-packets")
/** @type {import("../matrix/api")} */
const api = sync.require("../matrix/api")
const DEBUG_HOMESERVER_STATUS = true
function debugHomeserverStatus(message) {
if (DEBUG_HOMESERVER_STATUS) {
console.log(message)
}
}
const homeserverStatus = new class HomeserverStatus {
constructor() {
/** @private */
this.queue = new Denque()
/** @private */
this.pingInterval = undefined
/** @private */
this.sm = new StateMachine("online")
.defineState("online")
.defineState("offline", {
onEnter: [() => {
this.pingInterval = setInterval(async () => {
const pingResult = await api.ping().catch(e => ({ok: false, status: "net", root: e.message}))
if (pingResult.ok) {
this.sm.doTransition("ping ok")
}
}, 15e3)
}],
onLeave: [() => {
clearInterval(this.pingInterval)
}],
transitions: new Map()
})
.defineState("recovering", {
onEnter: [async () => { // Drain queue.
while (!this.queue.isEmpty()) {
const packet = this.queue.peekFront() // same position as .shift()
debugHomeserverStatus(`homeserver status: ${new Date().toISOString()} dq packet ${packet.t} ${packet.d?.content}`)
await discordPackets.dispatchPacketToBridge(passthrough.discord, packet)
if (this.sm.currentStateName !== "recovering") return // got kicked out due to another error
this.queue.shift()
}
this.sm.doTransition("recovered")
}],
onLeave: [],
transitions: new Map()
})
.defineUniversalTransition("error", "offline")
.defineTransition("offline", "ping ok", "recovering")
.defineTransition("recovering", "recovered", "online")
this.sm.on("enter", st => debugHomeserverStatus(`homeserver status: ${st}`))
this.sm.freeze()
}
isRealTime() {
return this.sm.currentStateName === "online"
}
/**
* When offline or recovering, call this for incoming packets to queue them to be sent in order later.
*/
queuePacket(packet) {
assert(["offline", "recovering"].includes(this.sm.currentStateName))
this.queue.push(packet)
}
setErrorWithPacket(packet) {
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
this.queuePacket(packet)
}
}
}
module.exports.homeserverStatus = homeserverStatus