Merge pull request #1000 from discordeno/implement-sweeper

add: sweepers

src/bot.ts

@@ -5,13 +5,14 @@ import { DiscordGatewayIntents } from "./types/gateway/gateway_intents.ts";
 import { snowflakeToBigint } from "./util/bigint.ts";
 import { GATEWAY_VERSION } from "./util/constants.ts";
 import { ws } from "./ws/ws.ts";
+import { dispatchRequirements } from "./util/dispatch_requirements.ts";
 
 // deno-lint-ignore prefer-const
 export let secretKey = "";
 export let botId = 0n;
 export let applicationId = 0n;
 
-export let eventHandlers: EventHandlers = {};
+export let eventHandlers: EventHandlers = { dispatchRequirements };
 
 export let proxyWSURL = `wss://gateway.discord.gg`;
 

src/cache.ts

@@ -1,4 +1,5 @@
 // deno-lint-ignore-file require-await no-explicit-any prefer-const
+import { botId } from "./bot.ts";
 import type { DiscordenoChannel } from "./structures/channel.ts";
 import type { DiscordenoGuild } from "./structures/guild.ts";
 import type { DiscordenoMember } from "./structures/member.ts";
@@ -10,17 +11,17 @@ import { Collection } from "./util/collection.ts";
 export const cache = {
   isReady: false,
   /** All of the guild objects the bot has access to, mapped by their Ids */
-  guilds: new Collection<bigint, DiscordenoGuild>(),
+  guilds: new Collection<bigint, DiscordenoGuild>([], { sweeper: { filter: guildSweeper, interval: 3600000 } }),
   /** All of the channel objects the bot has access to, mapped by their Ids */
   channels: new Collection<bigint, DiscordenoChannel>(),
   /** All of the message objects the bot has cached since the bot acquired `READY` state, mapped by their Ids */
-  messages: new Collection<bigint, DiscordenoMessage>(),
+  messages: new Collection<bigint, DiscordenoMessage>([], { sweeper: { filter: messageSweeper, interval: 300000 } }),
   /** All of the member objects that have been cached since the bot acquired `READY` state, mapped by their Ids */
-  members: new Collection<bigint, DiscordenoMember>(),
+  members: new Collection<bigint, DiscordenoMember>([], { sweeper: { filter: memberSweeper, interval: 300000 } }),
   /** All of the unavailable guilds, mapped by their Ids (id, timestamp) */
   unavailableGuilds: new Collection<bigint, number>(),
   /** All of the presence update objects received in PRESENCE_UPDATE gateway event, mapped by their user Id */
-  presences: new Collection<bigint, PresenceUpdate>(),
+  presences: new Collection<bigint, PresenceUpdate>([], { sweeper: { filter: () => true, interval: 300000 } }),
   fetchAllMembersProcessingRequests: new Collection<
     string,
     (value: Collection<bigint, DiscordenoMember> | PromiseLike<Collection<bigint, DiscordenoMember>>) => void
@@ -31,8 +32,43 @@ export const cache = {
       this.guilds.reduce((a, b) => [...a, ...b.emojis.map((e) => [e.id, e])], [] as any[])
     );
   },
+  activeGuildIds: new Set<bigint>(),
+  dispatchedGuildIds: new Set<bigint>(),
+  dispatchedChannelIds: new Set<bigint>(),
 };
 
+function messageSweeper(message: DiscordenoMessage) {
+  // DM messages aren't needed
+  if (!message.guildId) return true;
+
+  // Only delete messages older than 10 minutes
+  return Date.now() - message.timestamp > 600000;
+}
+
+function memberSweeper(member: DiscordenoMember) {
+  // Don't sweep the bot else strange things will happen
+  if (member.id === botId) return false;
+
+  // Only sweep members who were not active the last 30 minutes
+  return member.cachedAt - Date.now() < 1800000;
+}
+
+function guildSweeper(guild: DiscordenoGuild) {
+  // Reset activity for next interval
+  if (!cache.activeGuildIds.delete(guild.id)) return false;
+
+  guild.channels.forEach((channel) => {
+    cache.channels.delete(channel.id);
+    cache.dispatchedChannelIds.add(channel.id);
+  });
+
+  // This is inactive guild. Not a single thing has happened for atleast 30 minutes.
+  // Not a reaction, not a message, not any event!
+  cache.dispatchedGuildIds.add(guild.id);
+
+  return true;
+}
+
 export let cacheHandlers = {
   /** Deletes all items from the cache */
   async clear(table: TableName) {

src/handlers/misc/READY.ts

@@ -38,7 +38,7 @@ function checkReady(payload: Ready, shard: DiscordenoShard) {
   // Check if all guilds were loaded
   if (!shard.unavailableGuildIds.size) return loaded(shard);
 
-  // If the last GUILD_CREATE has been received before 5 seconds if so most likely the remaining guilds are unavailable
+  // If the last GUILD_CREATE was received 5 seconds ago, the remaining guilds are most likely not available
   if (shard.lastAvailable + 5000 < Date.now()) {
     eventHandlers.shardFailedToLoad?.(shard.id, shard.unavailableGuildIds);
     // Force execute the loaded function to prevent infinite loop

src/structures/member.ts

@@ -169,6 +169,7 @@ export async function createDiscordenoMember(
     /** The guild related data mapped by guild id */
     guilds: createNewProp(new Collection<bigint, GuildMember>()),
     bitfield: createNewProp(bitfield),
+    cachedAt: createNewProp(Date.now()),
   });
 
   const cached = await cacheHandlers.get("members", snowflakeToBigint(user.id));
@@ -210,6 +211,8 @@ export interface DiscordenoMember extends Omit<User, "discriminator" | "id" | "a
   >;
   /** Holds all the boolean toggles. */
   bitfield: bigint;
+  /** When the member has been cached the last time. */
+  cachedAt: number;
 
   // GETTERS
   /** The avatar url using the default format and size. */
@@ -230,7 +233,9 @@ export interface DiscordenoMember extends Omit<User, "discriminator" | "id" | "a
   /** Get the nickname or the username if no nickname */
   name(guildId: bigint): string;
   /** Get the guild member object for the specified guild */
-  guildMember(guildId: bigint):
+  guildMember(
+    guildId: bigint
+  ):
     | (Omit<GuildMember, "joinedAt" | "premiumSince" | "roles"> & {
         joinedAt?: number;
         premiumSince?: number;

src/util/collection.ts

@@ -31,6 +31,18 @@ export class Collection<K, V> extends Map<K, V> {
     return clearInterval(this.sweeper?.intervalId);
   }
 
+  changeSweeperInterval(newInterval: number) {
+    if (!this.sweeper) return;
+
+    this.startSweeper({ filter: this.sweeper.filter, interval: newInterval });
+  }
+
+  changeSweeperFilter(newFilter: (value: V, key: K) => boolean | Promise<boolean>) {
+    if (!this.sweeper) return;
+
+    this.startSweeper({ filter: newFilter, interval: this.sweeper.interval });
+  }
+
   set(key: K, value: V) {
     // When this collection is maxSizeed make sure we can add first
     if ((this.maxSize || this.maxSize === 0) && this.size >= this.maxSize) {

src/util/dispatch_requirements.ts

@@ -0,0 +1,93 @@
+import { botId } from "../bot.ts";
+import { cache } from "../cache.ts";
+import { getChannels } from "../helpers/channels/get_channels.ts";
+import { getGuild } from "../helpers/guilds/get_guild.ts";
+import { getMember } from "../helpers/members/get_member.ts";
+import { structures } from "../structures/mod.ts";
+import type { DiscordGatewayPayload } from "../types/gateway/gateway_payload.ts";
+import type { Guild } from "../types/guilds/guild.ts";
+import { snowflakeToBigint } from "./bigint.ts";
+import { delay } from "./utils.ts";
+
+const processing = new Set<bigint>();
+
+export async function dispatchRequirements(data: DiscordGatewayPayload, shardId: number) {
+  if (!cache.isReady) return;
+
+  // DELETE MEANS WE DONT NEED TO FETCH. CREATE SHOULD HAVE DATA TO CACHE
+  if (data.t && ["GUILD_CREATE", "GUILD_DELETE"].includes(data.t)) return;
+
+  const id = snowflakeToBigint(
+    (data.t && ["GUILD_UPDATE"].includes(data.t)
+      ? // deno-lint-ignore no-explicit-any
+        (data.d as any)?.id
+      : // deno-lint-ignore no-explicit-any
+        (data.d as any)?.guild_id) ?? ""
+  );
+
+  if (!id || cache.activeGuildIds.has(id)) return;
+
+  // If this guild is in cache, it has not been swept and we can cancel
+  if (cache.guilds.has(id)) {
+    cache.activeGuildIds.add(id);
+    return;
+  }
+
+  if (processing.has(id)) {
+    console.info(`[DISPATCH] New Guild ID already being processed: ${id} in ${data.t} event`);
+
+    let runs = 0;
+    do {
+      await delay(500);
+      runs++;
+    } while (processing.has(id) && runs < 40);
+
+    if (!processing.has(id)) return;
+
+    return console.warn(`[DISPATCH] Already processed guild was not successfully fetched:  ${id} in ${data.t} event`);
+  }
+
+  processing.add(id);
+
+  // New guild id has appeared, fetch all relevant data
+  console.info(`[DISPATCH] New Guild ID has appeared: ${id} in ${data.t} event`);
+
+  const rawGuild = (await getGuild(id, {
+    counts: true,
+    addToCache: false,
+  }).catch(console.info)) as Guild | undefined;
+
+  if (!rawGuild) {
+    processing.delete(id);
+    return console.warn(`[DISPATCH] Guild ID ${id} failed to fetch.`);
+  }
+
+  console.info(`[DISPATCH] Guild ID ${id} has been found. ${rawGuild.name}`);
+
+  const [channels, botMember] = await Promise.all([
+    getChannels(id, false),
+    getMember(id, botId, { force: true }),
+  ]).catch((error) => {
+    console.warn(error);
+    return [];
+  });
+
+  if (!botMember || !channels) {
+    processing.delete(id);
+    return console.info(`[DISPATCH] Guild ID ${id} Name: ${rawGuild.name} failed. Unable to get botMember or channels`);
+  }
+
+  const guild = await structures.createDiscordenoGuild(rawGuild, shardId);
+
+  // Add to cache
+  cache.guilds.set(id, guild);
+  cache.dispatchedGuildIds.delete(id);
+  channels.forEach((channel) => {
+    cache.dispatchedChannelIds.delete(channel.id);
+    cache.channels.set(channel.id, channel);
+  });
+
+  processing.delete(id);
+
+  console.info(`[DISPATCH] Guild ID ${id} Name: ${guild.name} completely loaded.`);
+}

tests/local.ts

@@ -1,3 +1,22 @@
+// THE ORDER OF THE IMPORTS IN THIS FILE MATTER!
+// DO NOT MOVE THEM UNLESS YOU KNOW WHAT YOUR DOING!
+
 import "./util/utils.ts";
 import "./util/validate_length.ts";
 import "./util/loop_object.ts";
+
+// Final cleanup
+
+import { cache } from "../src/cache.ts";
+import { delay } from "../src/util/utils.ts";
+if (import.meta.main) {
+  // clear all the sweeper intervals
+  for (const c of Object.values(cache)) {
+    if (!(c instanceof Map)) continue;
+
+    c.stopSweeper();
+    console.log("Cleaned");
+  }
+
+  await delay(3000);
+}

tests/mod.ts

@@ -74,3 +74,17 @@ import "./discoveries/valid_discovery_term.ts";
 // Final cleanup
 import "./guilds/delete_guild.ts";
 import "./ws/ws_close.ts";
+
+import { cache } from "../src/cache.ts";
+import { delay } from "../src/util/utils.ts";
+if (import.meta.main) {
+  // clear all the sweeper intervals
+  for (const c of Object.values(cache)) {
+    if (!(c instanceof Map)) continue;
+
+    c.stopSweeper();
+    console.log("Cleaned");
+  }
+
+  await delay(3000);
+}