Skip to content

Push Matrix->Slack messages through the same queue as Slack->Matrix #790

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Push Matrix->Slack messages through the same queue as Slack->Matrix #790

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 24 additions & 16 deletions src/BridgedRoom.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,11 +159,17 @@ export class BridgedRoom {
public MatrixRoomActive: boolean;
private recentSlackMessages: string[] = [];

private slackSendLock: Promise<unknown> = Promise.resolve();
private bridgingQueue: Promise<unknown> = Promise.resolve();

private waitingForJoin?: Promise<void>;
private waitingForJoinResolve?: () => void;

private async pushToBridgingQueue<T>(fn: () => Promise<T>): Promise<T> {
return new Promise((resolve) => {
this.bridgingQueue = this.bridgingQueue.finally(() => fn().then(resolve));
});
}

/**
* True if this instance has changed from the version last read/written to the RoomStore.
*/
Expand Down Expand Up @@ -528,7 +534,6 @@ export class BridgedRoom {
body.as_user = true;
delete body.username;
}
let res: ChatPostMessageResponse;
const chatPostMessageArgs = {
...body,
// Ensure that text is defined, even for attachments.
Expand All @@ -537,21 +542,25 @@ export class BridgedRoom {
unfurl_links: true,
};

try {
res = await slackClient.chat.postMessage(chatPostMessageArgs) as ChatPostMessageResponse;
} catch (ex) {
const platformError = ex as WebAPIPlatformError;
if (platformError.data?.error === "not_in_channel") {
await slackClient.conversations.join({
channel: chatPostMessageArgs.channel,
});
const res = await this.pushToBridgingQueue(async () => {
let res: ChatPostMessageResponse;
try {
res = await slackClient.chat.postMessage(chatPostMessageArgs) as ChatPostMessageResponse;
} else {
throw ex;
} catch (ex) {
const platformError = ex as WebAPIPlatformError;
if (platformError.data?.error === "not_in_channel") {
await slackClient.conversations.join({
channel: chatPostMessageArgs.channel,
});
res = await slackClient.chat.postMessage(chatPostMessageArgs) as ChatPostMessageResponse;
} else {
throw ex;
}
}
}

this.addRecentSlackMessage(res.ts);
this.addRecentSlackMessage(res.ts);
return res;
});

this.main.incCounter(METRIC_SENT_MESSAGES, {side: "remote"});
// Log activity, but don't await the answer or throw errors
Expand Down Expand Up @@ -710,7 +719,7 @@ export class BridgedRoom {
if (ghostChanged) {
await this.main.fixDMMetadata(this, ghost);
}
this.slackSendLock = this.slackSendLock.then(() => {
await this.pushToBridgingQueue(async () => {
// Check again
if (this.recentSlackMessages.includes(message.ts)) {
// We sent this, ignore
Expand All @@ -720,7 +729,6 @@ export class BridgedRoom {
log.warn(`Failed to handle slack message ${message.ts} for ${this.MatrixRoomId} ${this.slackChannelId}`, ex);
});
});
await this.slackSendLock;
} catch (err) {
log.error("Failed to process event");
log.error(err);
Expand Down