Wait for ratelimits when sending invites (and emit a count of the number of sent invites)

src/commands/InviteCommand.ts

@@ -24,13 +24,14 @@ import { logMessage } from "../LogProxy";
 import { IPerson, Role } from "../models/schedule";
 import { ConferenceMatrixClient } from "../ConferenceMatrixClient";
 import { IConfig } from "../config";
+import { sleep } from "../utils";
 
 export class InviteCommand implements ICommand {
     public readonly prefixes = ["invite", "inv"];
 
     constructor(private readonly client: ConferenceMatrixClient, private readonly conference: Conference, private readonly config: IConfig) {}
 
-    private async createInvites(people: IPerson[], alias: string) {
+    private async createInvites(people: IPerson[], alias: string): Promise<number> {
         const resolved = await resolveIdentifiers(this.client, people);
 
         let targetRoomId;
@@ -40,7 +41,7 @@ export class InviteCommand implements ICommand {
         catch (error) {
             throw Error(`Error resolving room id for ${alias}`, {cause: error})
         }
-        await this.ensureInvited(targetRoomId, resolved);
+        return await this.ensureInvited(targetRoomId, resolved);
     }
 
     public async run(managementRoomId: string, event: any, args: string[]) {
@@ -51,6 +52,8 @@ export class InviteCommand implements ICommand {
         // in it. We don't remove anyone and don't care about extras - we just want to make sure
         // that a subset of people are joined.
 
+        let invitesSent = 0;
+
         if (args[0] && args[0] === "speakers-support") {
             let people: IPerson[] = [];
             for (const aud of this.conference.storedAuditoriumBackstages) {
@@ -63,7 +66,7 @@ export class InviteCommand implements ICommand {
                     newPeople.push(p);
                 }
             });
-            await this.createInvites(newPeople, this.config.conference.supportRooms.speakers);
+            invitesSent += await this.createInvites(newPeople, this.config.conference.supportRooms.speakers);
         } else if (args[0] && args[0] === "coordinators-support") {
             let people: IPerson[] = [];
             for (const aud of this.conference.storedAuditoriums) {
@@ -84,24 +87,27 @@ export class InviteCommand implements ICommand {
                     newPeople.push(p);
                 }
             });
-            await this.createInvites(newPeople, this.config.conference.supportRooms.coordinators);
+            invitesSent += await this.createInvites(newPeople, this.config.conference.supportRooms.coordinators);
         } else if (args[0] && args[0] === "si-support") {
             const people: IPerson[] = [];
             for (const sir of this.conference.storedInterestRooms) {
                 people.push(...await this.conference.getInviteTargetsForInterest(sir));
             }
-            await this.createInvites(people, this.config.conference.supportRooms.specialInterest);
+            invitesSent += await this.createInvites(people, this.config.conference.supportRooms.specialInterest);
         } else {
-            await runRoleCommand((_client, room, people) => this.ensureInvited(room, people), this.conference, this.client, managementRoomId, event, args);
+            await runRoleCommand(async (_client, room, people) => {
+                invitesSent += await this.ensureInvited(room, people);
+            }, this.conference, this.client, managementRoomId, event, args);
         }
 
-        await this.client.sendNotice(managementRoomId, "Invites sent!");
+        await this.client.sendNotice(managementRoomId, `${invitesSent} invites sent!`);
     }
 
-    public async ensureInvited(roomId: string, people: ResolvedPersonIdentifier[]) {
+    public async ensureInvited(roomId: string, people: ResolvedPersonIdentifier[]): Promise<number> {
         // We don't want to invite anyone we have already invited or that has joined though, so
         // avoid those people. We do this by querying the room state and filtering.
-        let state;
+        let invitesSent = 0;
+        let state: any[];
         try {
             state = await this.client.getRoomState(roomId);
         }
@@ -114,12 +120,28 @@ export class InviteCommand implements ICommand {
         for (const target of people) {
             if (target.mxid && effectiveJoinedUserIds.includes(target.mxid)) continue;
             if (emailInvitePersonIds.includes(target.person.id)) continue;
-            try {
-                await invitePersonToRoom(this.client, target, roomId, this.config);
-            } catch (e) {
-                LogService.error("InviteCommand", e);
-                await logMessage(LogLevel.ERROR, "InviteCommand", `Error inviting ${target.mxid}/${target.emails} / ${target.person.id} to ${roomId} - ignoring: ${e.message ?? e.statusMessage ?? '(see logs)'}`, this.client);
+            for (let attempt = 0; attempt < 3; ++attempt) {
+                try {
+                    await invitePersonToRoom(this.client, target, roomId, this.config);
+                    ++invitesSent;
+                } catch (e) {
+                    if (e.statusCode === 429) {
+                        // Retry after ratelimits
+                        // Add 1 second to the ratelimit just to ensure we don't retry too quickly
+                        // due to clock drift or a very small requested wait.
+                        // If no retry time set, use 5 minutes.
+                        let delay = (e.retryAfterMs ?? 300_000) + 1_000;
+
+                        await sleep(delay);
+                        continue;
+                    }
+                    LogService.error("InviteCommand", e);
+                    await logMessage(LogLevel.ERROR, "InviteCommand", `Error inviting ${target.mxid}/${target.emails} / ${target.person.id} to ${roomId} - ignoring: ${e.message ?? e.statusMessage ?? '(see logs)'}`, this.client);
+                }
+                break;
             }
         }
+
+        return invitesSent;
     }
 }

src/invites.ts

@@ -79,6 +79,11 @@ export async function resolveIdentifiers(client: ConferenceMatrixClient, people:
     return resolved;
 }
 
+/**
+ * Invites a person to a room idempotently.
+ *
+ * @throws an exception when we don't have information to invite the user, or there is some Matrix or network error preventing us from doing so.
+ */
 export async function invitePersonToRoom(client: ConferenceMatrixClient, resolvedPerson: ResolvedPersonIdentifier, roomId: string, config: IConfig): Promise<void> {
     if (resolvedPerson.mxid) {
         if (config.dry_run_enabled) {

src/utils.ts

@@ -245,3 +245,7 @@ export function jsonReplacerMapToObject(_key: any, input: any): any {
     }
     return input;
 }
+
+export function sleep(millis: number): Promise<void> {
+    return new Promise(resolve => setTimeout(resolve, millis));
+}