Centralize periodic checks for mutes, tempbans, vcalerts, reminders, and scheduled posts

This should result in a significant performance improvement.
The new method is also more precise than the old one, allowing
the aforementioned checks to be performed with second-precision.

backend/src/data/GuildEvents.ts

@@ -1,11 +1,15 @@
 import { Mute } from "./entities/Mute";
 import { ScheduledPost } from "./entities/ScheduledPost";
 import { Reminder } from "./entities/Reminder";
+import { Tempban } from "./entities/Tempban";
+import { VCAlert } from "./entities/VCAlert";
 
 interface GuildEventArgs extends Record<string, unknown[]> {
-  expiredMutes: [Mute[]];
-  scheduledPosts: [ScheduledPost[]];
-  reminders: [Reminder[]];
+  expiredMute: [Mute];
+  scheduledPost: [ScheduledPost];
+  reminder: [Reminder];
+  expiredTempban: [Tempban];
+  expiredVCAlert: [VCAlert];
 }
 
 type GuildEvent = keyof GuildEventArgs;
@@ -52,3 +56,14 @@ export function emitGuildEvent<K extends GuildEvent>(guildId: string, eventName:
     listener(...args);
   }
 }
+
+export function hasGuildEventListener<K extends GuildEvent>(guildId: string, eventName: K): boolean {
+  if (!guildListeners.has(guildId)) {
+    return false;
+  }
+  const listenerMap = guildListeners.get(guildId)!;
+  if (listenerMap[eventName] == null || listenerMap[eventName]!.length === 0) {
+    return false;
+  }
+  return true;
+}

backend/src/data/GuildMutes.ts

@@ -47,11 +47,11 @@ export class GuildMutes extends BaseGuildRepository {
     return (await this.mutes.findOne({ where: result.identifiers[0] }))!;
   }
 
-  async updateExpiryTime(userId, newExpiryTime, rolesToRestore?: string[]) {
+  async updateExpiryTime(userId, newExpiryTime, rolesToRestore?: string[]): Promise<void> {
     const expiresAt = newExpiryTime ? moment.utc().add(newExpiryTime, "ms").format("YYYY-MM-DD HH:mm:ss") : null;
 
     if (rolesToRestore && rolesToRestore.length) {
-      return this.mutes.update(
+      await this.mutes.update(
         {
           guild_id: this.guildId,
           user_id: userId,
@@ -62,7 +62,7 @@ export class GuildMutes extends BaseGuildRepository {
         },
       );
     } else {
-      return this.mutes.update(
+      await this.mutes.update(
         {
           guild_id: this.guildId,
           user_id: userId,

backend/src/data/GuildReminders.ts

@@ -27,6 +27,10 @@ export class GuildReminders extends BaseGuildRepository {
     });
   }
 
+  find(id: number) {
+    return this.reminders.findOne({ id });
+  }
+
   async delete(id) {
     await this.reminders.delete({
       guild_id: this.guildId,
@@ -35,7 +39,7 @@ export class GuildReminders extends BaseGuildRepository {
   }
 
   async add(userId: string, channelId: string, remindAt: string, body: string, created_at: string) {
-    await this.reminders.insert({
+    const result = await this.reminders.insert({
       guild_id: this.guildId,
       user_id: userId,
       channel_id: channelId,
@@ -43,5 +47,7 @@ export class GuildReminders extends BaseGuildRepository {
       body,
       created_at,
     });
+
+    return (await this.find(result.identifiers[0].id))!;
   }
 }

backend/src/data/GuildScheduledPosts.ts

@@ -22,6 +22,10 @@ export class GuildScheduledPosts extends BaseGuildRepository {
       .getMany();
   }
 
+  find(id: number) {
+    return this.scheduledPosts.findOne({ id });
+  }
+
   async delete(id) {
     await this.scheduledPosts.delete({
       guild_id: this.guildId,
@@ -30,10 +34,12 @@ export class GuildScheduledPosts extends BaseGuildRepository {
   }
 
   async create(data: Partial<ScheduledPost>) {
-    await this.scheduledPosts.insert({
+    const result = await this.scheduledPosts.insert({
       ...data,
       guild_id: this.guildId,
     });
+
+    return (await this.find(result.identifiers[0].id))!;
   }
 
   async update(id: number, data: Partial<ScheduledPost>) {

backend/src/data/GuildVCAlerts.ts

@@ -40,6 +40,10 @@ export class GuildVCAlerts extends BaseGuildRepository {
     });
   }
 
+  find(id: number) {
+    return this.allAlerts.findOne({ id });
+  }
+
   async delete(id) {
     await this.allAlerts.delete({
       guild_id: this.guildId,
@@ -48,7 +52,7 @@ export class GuildVCAlerts extends BaseGuildRepository {
   }
 
   async add(requestorId: string, userId: string, channelId: string, expiresAt: string, body: string, active: boolean) {
-    await this.allAlerts.insert({
+    const result = await this.allAlerts.insert({
       guild_id: this.guildId,
       requestor_id: requestorId,
       user_id: userId,
@@ -57,5 +61,7 @@ export class GuildVCAlerts extends BaseGuildRepository {
       body,
       active,
     });
+
+    return (await this.find(result.identifiers[0].id))!;
   }
 }

backend/src/data/Mutes.ts

@@ -0,0 +1,23 @@
+import moment from "moment-timezone";
+import { Brackets, getRepository, Repository } from "typeorm";
+import { Mute } from "./entities/Mute";
+import { DBDateFormat } from "../utils";
+import { BaseRepository } from "./BaseRepository";
+
+export class Mutes extends BaseRepository {
+  private mutes: Repository<Mute>;
+
+  constructor() {
+    super();
+    this.mutes = getRepository(Mute);
+  }
+
+  async getSoonExpiringMutes(threshold: number): Promise<Mute[]> {
+    const thresholdDateStr = moment.utc().add(threshold, "ms").format(DBDateFormat);
+    return this.mutes
+      .createQueryBuilder("mutes")
+      .andWhere("expires_at IS NOT NULL")
+      .andWhere("expires_at <= :date", { date: thresholdDateStr })
+      .getMany();
+  }
+}

backend/src/data/Reminders.ts

@@ -0,0 +1,19 @@
+import { getRepository, Repository } from "typeorm";
+import { Reminder } from "./entities/Reminder";
+import { BaseRepository } from "./BaseRepository";
+import moment from "moment-timezone";
+import { DBDateFormat } from "../utils";
+
+export class Reminders extends BaseRepository {
+  private reminders: Repository<Reminder>;
+
+  constructor() {
+    super();
+    this.reminders = getRepository(Reminder);
+  }
+
+  async getRemindersDueSoon(threshold: number): Promise<Reminder[]> {
+    const thresholdDateStr = moment.utc().add(threshold, "ms").format(DBDateFormat);
+    return this.reminders.createQueryBuilder().andWhere("remind_at <= :date", { date: thresholdDateStr }).getMany();
+  }
+}

backend/src/data/ScheduledPosts.ts

@@ -0,0 +1,19 @@
+import { getRepository, Repository } from "typeorm";
+import { ScheduledPost } from "./entities/ScheduledPost";
+import { BaseRepository } from "./BaseRepository";
+import moment from "moment-timezone";
+import { DBDateFormat } from "../utils";
+
+export class ScheduledPosts extends BaseRepository {
+  private scheduledPosts: Repository<ScheduledPost>;
+
+  constructor() {
+    super();
+    this.scheduledPosts = getRepository(ScheduledPost);
+  }
+
+  getScheduledPostsDueSoon(threshold: number): Promise<ScheduledPost[]> {
+    const thresholdDateStr = moment.utc().add(threshold, "ms").format(DBDateFormat);
+    return this.scheduledPosts.createQueryBuilder().andWhere("post_at <= :date", { date: thresholdDateStr }).getMany();
+  }
+}

backend/src/data/Tempbans.ts

@@ -0,0 +1,19 @@
+import moment from "moment-timezone";
+import { getRepository, Repository } from "typeorm";
+import { Tempban } from "./entities/Tempban";
+import { BaseRepository } from "./BaseRepository";
+import { DBDateFormat } from "../utils";
+
+export class Tempbans extends BaseRepository {
+  private tempbans: Repository<Tempban>;
+
+  constructor() {
+    super();
+    this.tempbans = getRepository(Tempban);
+  }
+
+  getSoonExpiringTempbans(threshold: number): Promise<Tempban[]> {
+    const thresholdDateStr = moment.utc().add(threshold, "ms").format(DBDateFormat);
+    return this.tempbans.createQueryBuilder().where("expires_at <= :date", { date: thresholdDateStr }).getMany();
+  }
+}

backend/src/data/VCAlerts.ts

@@ -0,0 +1,19 @@
+import { getRepository, Repository } from "typeorm";
+import { VCAlert } from "./entities/VCAlert";
+import { BaseRepository } from "./BaseRepository";
+import moment from "moment-timezone";
+import { DBDateFormat } from "../utils";
+
+export class VCAlerts extends BaseRepository {
+  private allAlerts: Repository<VCAlert>;
+
+  constructor() {
+    super();
+    this.allAlerts = getRepository(VCAlert);
+  }
+
+  async getSoonExpiringAlerts(threshold: number): Promise<VCAlert[]> {
+    const thresholdDateStr = moment.utc().add(threshold, "ms").format(DBDateFormat);
+    return this.allAlerts.createQueryBuilder().andWhere("expires_at <= :date", { date: thresholdDateStr }).getMany();
+  }
+}

backend/src/data/entities/Reminder.ts

@@ -1,9 +1,8 @@
-import { Column, Entity, PrimaryColumn } from "typeorm";
+import { Column, Entity, PrimaryColumn, PrimaryGeneratedColumn } from "typeorm";
 
 @Entity("reminders")
 export class Reminder {
-  @Column()
-  @PrimaryColumn()
+  @PrimaryGeneratedColumn()
   id: number;
 
   @Column() guild_id: string;

backend/src/data/entities/ScheduledPost.ts

@@ -1,11 +1,10 @@
 import { MessageAttachment } from "discord.js";
-import { Column, Entity, PrimaryColumn } from "typeorm";
+import { Column, Entity, PrimaryColumn, PrimaryGeneratedColumn } from "typeorm";
 import { StrictMessageContent } from "../../utils";
 
 @Entity("scheduled_posts")
 export class ScheduledPost {
-  @Column()
-  @PrimaryColumn()
+  @PrimaryGeneratedColumn()
   id: number;
 
   @Column() guild_id: string;

backend/src/data/entities/VCAlert.ts

@@ -1,9 +1,8 @@
-import { Column, Entity, PrimaryColumn } from "typeorm";
+import { Column, Entity, PrimaryColumn, PrimaryGeneratedColumn } from "typeorm";
 
 @Entity("vc_alerts")
 export class VCAlert {
-  @Column()
-  @PrimaryColumn()
+  @PrimaryGeneratedColumn()
   id: number;
 
   @Column() guild_id: string;

backend/src/data/loops/expiringMutesLoop.ts

@@ -0,0 +1,72 @@
+import { lazyMemoize, memoize, MINUTES } from "../../utils";
+import { Mutes } from "../Mutes";
+import Timeout = NodeJS.Timeout;
+import moment from "moment-timezone";
+import { Mute } from "../entities/Mute";
+import { emitGuildEvent, hasGuildEventListener } from "../GuildEvents";
+
+const LOOP_INTERVAL = 15 * MINUTES;
+const MAX_TRIES_PER_SERVER = 3;
+const getMutesRepository = lazyMemoize(() => new Mutes());
+const timeouts = new Map<string, Timeout>();
+
+function muteToKey(mute: Mute) {
+  return `${mute.guild_id}/${mute.user_id}`;
+}
+
+function broadcastExpiredMute(mute: Mute, tries = 0) {
+  console.log(`[EXPIRING MUTES LOOP] Broadcasting expired mute: ${mute.guild_id}/${mute.user_id}`);
+  if (!hasGuildEventListener(mute.guild_id, "expiredMute")) {
+    // If there are no listeners registered for the server yet, try again in a bit
+    if (tries < MAX_TRIES_PER_SERVER) {
+      timeouts.set(
+        muteToKey(mute),
+        setTimeout(() => broadcastExpiredMute(mute, tries + 1), 1 * MINUTES),
+      );
+    }
+    return;
+  }
+  emitGuildEvent(mute.guild_id, "expiredMute", [mute]);
+}
+
+export async function runExpiringMutesLoop() {
+  console.log("[EXPIRING MUTES LOOP] Clearing old timeouts");
+  for (const timeout of timeouts.values()) {
+    clearTimeout(timeout);
+  }
+
+  console.log("[EXPIRING MUTES LOOP] Setting timeouts for expiring mutes");
+  const expiringMutes = await getMutesRepository().getSoonExpiringMutes(LOOP_INTERVAL);
+  for (const mute of expiringMutes) {
+    const remaining = Math.max(0, moment.utc(mute.expires_at!).diff(moment.utc()));
+    timeouts.set(
+      muteToKey(mute),
+      setTimeout(() => broadcastExpiredMute(mute), remaining),
+    );
+  }
+
+  console.log("[EXPIRING MUTES LOOP] Scheduling next loop");
+  setTimeout(() => runExpiringMutesLoop(), LOOP_INTERVAL);
+}
+
+export function registerExpiringMute(mute: Mute) {
+  clearExpiringMute(mute);
+
+  console.log("[EXPIRING MUTES LOOP] Registering new expiring mute");
+  const remaining = Math.max(0, moment.utc(mute.expires_at!).diff(moment.utc()));
+  if (remaining > LOOP_INTERVAL) {
+    return;
+  }
+
+  timeouts.set(
+    muteToKey(mute),
+    setTimeout(() => broadcastExpiredMute(mute), remaining),
+  );
+}
+
+export function clearExpiringMute(mute: Mute) {
+  console.log("[EXPIRING MUTES LOOP] Clearing expiring mute");
+  if (timeouts.has(muteToKey(mute))) {
+    clearTimeout(timeouts.get(muteToKey(mute))!);
+  }
+}

backend/src/data/loops/expiringTempbansLoop.ts

@@ -0,0 +1,72 @@
+import { lazyMemoize, MINUTES } from "../../utils";
+import moment from "moment-timezone";
+import { emitGuildEvent, hasGuildEventListener } from "../GuildEvents";
+import { Tempbans } from "../Tempbans";
+import { Tempban } from "../entities/Tempban";
+import Timeout = NodeJS.Timeout;
+
+const LOOP_INTERVAL = 15 * MINUTES;
+const MAX_TRIES_PER_SERVER = 3;
+const getBansRepository = lazyMemoize(() => new Tempbans());
+const timeouts = new Map<string, Timeout>();
+
+function tempbanToKey(tempban: Tempban) {
+  return `${tempban.guild_id}/${tempban.user_id}`;
+}
+
+function broadcastExpiredTempban(tempban: Tempban, tries = 0) {
+  console.log(`[EXPIRING TEMPBANS LOOP] Broadcasting expired tempban: ${tempban.guild_id}/${tempban.user_id}`);
+  if (!hasGuildEventListener(tempban.guild_id, "expiredTempban")) {
+    // If there are no listeners registered for the server yet, try again in a bit
+    if (tries < MAX_TRIES_PER_SERVER) {
+      timeouts.set(
+        tempbanToKey(tempban),
+        setTimeout(() => broadcastExpiredTempban(tempban, tries + 1), 1 * MINUTES),
+      );
+    }
+    return;
+  }
+  emitGuildEvent(tempban.guild_id, "expiredTempban", [tempban]);
+}
+
+export async function runExpiringTempbansLoop() {
+  console.log("[EXPIRING TEMPBANS LOOP] Clearing old timeouts");
+  for (const timeout of timeouts.values()) {
+    clearTimeout(timeout);
+  }
+
+  console.log("[EXPIRING TEMPBANS LOOP] Setting timeouts for expiring tempbans");
+  const expiringTempbans = await getBansRepository().getSoonExpiringTempbans(LOOP_INTERVAL);
+  for (const tempban of expiringTempbans) {
+    const remaining = Math.max(0, moment.utc(tempban.expires_at!).diff(moment.utc()));
+    timeouts.set(
+      tempbanToKey(tempban),
+      setTimeout(() => broadcastExpiredTempban(tempban), remaining),
+    );
+  }
+
+  console.log("[EXPIRING TEMPBANS LOOP] Scheduling next loop");
+  setTimeout(() => runExpiringTempbansLoop(), LOOP_INTERVAL);
+}
+
+export function registerExpiringTempban(tempban: Tempban) {
+  clearExpiringTempban(tempban);
+
+  console.log("[EXPIRING TEMPBANS LOOP] Registering new expiring tempban");
+  const remaining = Math.max(0, moment.utc(tempban.expires_at!).diff(moment.utc()));
+  if (remaining > LOOP_INTERVAL) {
+    return;
+  }
+
+  timeouts.set(
+    tempbanToKey(tempban),
+    setTimeout(() => broadcastExpiredTempban(tempban), remaining),
+  );
+}
+
+export function clearExpiringTempban(tempban: Tempban) {
+  console.log("[EXPIRING TEMPBANS LOOP] Clearing expiring tempban");
+  if (timeouts.has(tempbanToKey(tempban))) {
+    clearTimeout(timeouts.get(tempbanToKey(tempban))!);
+  }
+}

backend/src/data/loops/expiringVCAlertsLoop.ts

@@ -0,0 +1,68 @@
+import { lazyMemoize, MINUTES } from "../../utils";
+import moment from "moment-timezone";
+import { emitGuildEvent, hasGuildEventListener } from "../GuildEvents";
+import Timeout = NodeJS.Timeout;
+import { VCAlerts } from "../VCAlerts";
+import { VCAlert } from "../entities/VCAlert";
+
+const LOOP_INTERVAL = 15 * MINUTES;
+const MAX_TRIES_PER_SERVER = 3;
+const getVCAlertsRepository = lazyMemoize(() => new VCAlerts());
+const timeouts = new Map<number, Timeout>();
+
+function broadcastExpiredVCAlert(alert: VCAlert, tries = 0) {
+  console.log(`[EXPIRING VCALERTS LOOP] Broadcasting expired vcalert: ${alert.guild_id}/${alert.user_id}`);
+  if (!hasGuildEventListener(alert.guild_id, "expiredVCAlert")) {
+    // If there are no listeners registered for the server yet, try again in a bit
+    if (tries < MAX_TRIES_PER_SERVER) {
+      timeouts.set(
+        alert.id,
+        setTimeout(() => broadcastExpiredVCAlert(alert, tries + 1), 1 * MINUTES),
+      );
+    }
+    return;
+  }
+  emitGuildEvent(alert.guild_id, "expiredVCAlert", [alert]);
+}
+
+export async function runExpiringVCAlertsLoop() {
+  console.log("[EXPIRING VCALERTS LOOP] Clearing old timeouts");
+  for (const timeout of timeouts.values()) {
+    clearTimeout(timeout);
+  }
+
+  console.log("[EXPIRING VCALERTS LOOP] Setting timeouts for expiring vcalerts");
+  const expiringVCAlerts = await getVCAlertsRepository().getSoonExpiringAlerts(LOOP_INTERVAL);
+  for (const alert of expiringVCAlerts) {
+    const remaining = Math.max(0, moment.utc(alert.expires_at!).diff(moment.utc()));
+    timeouts.set(
+      alert.id,
+      setTimeout(() => broadcastExpiredVCAlert(alert), remaining),
+    );
+  }
+
+  console.log("[EXPIRING VCALERTS LOOP] Scheduling next loop");
+  setTimeout(() => runExpiringVCAlertsLoop(), LOOP_INTERVAL);
+}
+
+export function registerExpiringVCAlert(alert: VCAlert) {
+  clearExpiringVCAlert(alert);
+
+  console.log("[EXPIRING VCALERTS LOOP] Registering new expiring vcalert");
+  const remaining = Math.max(0, moment.utc(alert.expires_at!).diff(moment.utc()));
+  if (remaining > LOOP_INTERVAL) {
+    return;
+  }
+
+  timeouts.set(
+    alert.id,
+    setTimeout(() => broadcastExpiredVCAlert(alert), remaining),
+  );
+}
+
+export function clearExpiringVCAlert(alert: VCAlert) {
+  console.log("[EXPIRING VCALERTS LOOP] Clearing expiring vcalert");
+  if (timeouts.has(alert.id)) {
+    clearTimeout(timeouts.get(alert.id)!);
+  }
+}

backend/src/data/loops/upcomingRemindersLoop.ts

@@ -0,0 +1,67 @@
+import { lazyMemoize, MINUTES } from "../../utils";
+import moment from "moment-timezone";
+import { emitGuildEvent, hasGuildEventListener } from "../GuildEvents";
+import { Reminder } from "../entities/Reminder";
+import { Reminders } from "../Reminders";
+import Timeout = NodeJS.Timeout;
+
+const LOOP_INTERVAL = 15 * MINUTES;
+const MAX_TRIES_PER_SERVER = 3;
+const getRemindersRepository = lazyMemoize(() => new Reminders());
+const timeouts = new Map<number, Timeout>();
+
+function broadcastReminder(reminder: Reminder, tries = 0) {
+  if (!hasGuildEventListener(reminder.guild_id, "reminder")) {
+    // If there are no listeners registered for the server yet, try again in a bit
+    if (tries < MAX_TRIES_PER_SERVER) {
+      timeouts.set(
+        reminder.id,
+        setTimeout(() => broadcastReminder(reminder, tries + 1), 1 * MINUTES),
+      );
+    }
+    return;
+  }
+  emitGuildEvent(reminder.guild_id, "reminder", [reminder]);
+}
+
+export async function runUpcomingRemindersLoop() {
+  console.log("[REMINDERS LOOP] Clearing old timeouts");
+  for (const timeout of timeouts.values()) {
+    clearTimeout(timeout);
+  }
+
+  console.log("[REMINDERS LOOP] Setting timeouts for upcoming reminders");
+  const remindersDueSoon = await getRemindersRepository().getRemindersDueSoon(LOOP_INTERVAL);
+  for (const reminder of remindersDueSoon) {
+    const remaining = Math.max(0, moment.utc(reminder.remind_at!).diff(moment.utc()));
+    timeouts.set(
+      reminder.id,
+      setTimeout(() => broadcastReminder(reminder), remaining),
+    );
+  }
+
+  console.log("[REMINDERS LOOP] Scheduling next loop");
+  setTimeout(() => runUpcomingRemindersLoop(), LOOP_INTERVAL);
+}
+
+export function registerUpcomingReminder(reminder: Reminder) {
+  clearUpcomingReminder(reminder);
+
+  console.log("[REMINDERS LOOP] Registering new upcoming reminder");
+  const remaining = Math.max(0, moment.utc(reminder.remind_at!).diff(moment.utc()));
+  if (remaining > LOOP_INTERVAL) {
+    return;
+  }
+
+  timeouts.set(
+    reminder.id,
+    setTimeout(() => broadcastReminder(reminder), remaining),
+  );
+}
+
+export function clearUpcomingReminder(reminder: Reminder) {
+  console.log("[REMINDERS LOOP] Clearing upcoming reminder");
+  if (timeouts.has(reminder.id)) {
+    clearTimeout(timeouts.get(reminder.id)!);
+  }
+}

backend/src/data/loops/upcomingScheduledPostsLoop.ts

@@ -0,0 +1,67 @@
+import { lazyMemoize, MINUTES } from "../../utils";
+import moment from "moment-timezone";
+import { emitGuildEvent, hasGuildEventListener } from "../GuildEvents";
+import { ScheduledPosts } from "../ScheduledPosts";
+import { ScheduledPost } from "../entities/ScheduledPost";
+import Timeout = NodeJS.Timeout;
+
+const LOOP_INTERVAL = 15 * MINUTES;
+const MAX_TRIES_PER_SERVER = 3;
+const getScheduledPostsRepository = lazyMemoize(() => new ScheduledPosts());
+const timeouts = new Map<number, Timeout>();
+
+function broadcastScheduledPost(post: ScheduledPost, tries = 0) {
+  if (!hasGuildEventListener(post.guild_id, "scheduledPost")) {
+    // If there are no listeners registered for the server yet, try again in a bit
+    if (tries < MAX_TRIES_PER_SERVER) {
+      timeouts.set(
+        post.id,
+        setTimeout(() => broadcastScheduledPost(post, tries + 1), 1 * MINUTES),
+      );
+    }
+    return;
+  }
+  emitGuildEvent(post.guild_id, "scheduledPost", [post]);
+}
+
+export async function runUpcomingScheduledPostsLoop() {
+  console.log("[SCHEDULED POSTS LOOP] Clearing old timeouts");
+  for (const timeout of timeouts.values()) {
+    clearTimeout(timeout);
+  }
+
+  console.log("[SCHEDULED POSTS LOOP] Setting timeouts for upcoming scheduled posts");
+  const postsDueSoon = await getScheduledPostsRepository().getScheduledPostsDueSoon(LOOP_INTERVAL);
+  for (const post of postsDueSoon) {
+    const remaining = Math.max(0, moment.utc(post.post_at!).diff(moment.utc()));
+    timeouts.set(
+      post.id,
+      setTimeout(() => broadcastScheduledPost(post), remaining),
+    );
+  }
+
+  console.log("[SCHEDULED POSTS LOOP] Scheduling next loop");
+  setTimeout(() => runUpcomingScheduledPostsLoop(), LOOP_INTERVAL);
+}
+
+export function registerUpcomingScheduledPost(post: ScheduledPost) {
+  clearUpcomingScheduledPost(post);
+
+  console.log("[SCHEDULED POSTS LOOP] Registering new upcoming scheduled post");
+  const remaining = Math.max(0, moment.utc(post.post_at!).diff(moment.utc()));
+  if (remaining > LOOP_INTERVAL) {
+    return;
+  }
+
+  timeouts.set(
+    post.id,
+    setTimeout(() => broadcastScheduledPost(post), remaining),
+  );
+}
+
+export function clearUpcomingScheduledPost(post: ScheduledPost) {
+  console.log("[SCHEDULED POSTS LOOP] Clearing upcoming scheduled post");
+  if (timeouts.has(post.id)) {
+    clearTimeout(timeouts.get(post.id)!);
+  }
+}