Merge pull request #866 from discordeno/fix-sbbb

Fix sbbb

src/bot.ts

@@ -14,18 +14,6 @@ export let eventHandlers: EventHandlers = {};
 
 export let proxyWSURL = `wss://gateway.discord.gg`;
 
-export const identifyPayload = {
-  token: "",
-  compress: true,
-  properties: {
-    $os: "linux",
-    $browser: "Discordeno",
-    $device: "Discordeno",
-  },
-  intents: 0,
-  shard: [0, 0],
-};
-
 export async function startBot(config: BotConfig) {
   if (config.eventHandlers) eventHandlers = config.eventHandlers;
   authorization = `Bot ${config.token}`;
@@ -49,17 +37,8 @@ export async function startBot(config: BotConfig) {
   ws.botGatewayData.url += `?v=${GATEWAY_VERSION}&encoding=json`;
 
   proxyWSURL = ws.botGatewayData.url;
-  identifyPayload.token = config.token;
-  identifyPayload.intents = config.intents.reduce(
-    (
-      bits,
-      next,
-    ) => (bits |= typeof next === "string"
-      ? DiscordGatewayIntents[next]
-      : next),
-    0,
-  );
-  identifyPayload.shard = [0, ws.botGatewayData.shards];
+
+  // ws.lastShardId = ws.maxShards;
 
   ws.spawnShards();
 }
@@ -90,34 +69,43 @@ export function setApplicationId(id: string) {
  *
  * Advanced Devs: This function will allow you to have an insane amount of customization potential as when you get to large bots you need to be able to optimize every tiny detail to make you bot work the way you need.
  */
-export async function startBigBrainBot(data: BigBrainBotConfig) {
-  authorization = `Bot ${data.token}`;
-  identifyPayload.token = `Bot ${data.token}`;
+export async function startBigBrainBot(options: BigBrainBotConfig) {
+  authorization = `Bot ${options.token}`;
+  rest.token = `Bot ${options.token}`;
 
-  if (data.secretKey) secretKey = data.secretKey;
-  if (data.restURL) baseEndpoints.BASE_URL = data.restURL;
-  if (data.cdnURL) baseEndpoints.CDN_URL = data.cdnURL;
-  if (data.eventHandlers) eventHandlers = data.eventHandlers;
-  if (data.compress) {
-    identifyPayload.compress = data.compress;
-  }
-
-  identifyPayload.intents = data.intents.reduce(
-    (
-      bits,
-      next,
-    ) => (bits |= typeof next === "string"
-      ? DiscordGatewayIntents[next]
-      : next),
-    0,
-  );
+  if (options.secretKey) secretKey = options.secretKey;
+  if (options.restURL) baseEndpoints.BASE_URL = options.restURL;
+  if (options.cdnURL) baseEndpoints.CDN_URL = options.cdnURL;
+  if (options.eventHandlers) eventHandlers = options.eventHandlers;
 
   // PROXY DOESNT NEED US SPAWNING SHARDS
-  if (!data.wsPort) {
+  if (!options.wsPort) {
+    ws.identifyPayload.token = `Bot ${options.token}`;
+
+    if (options.compress) {
+      ws.identifyPayload.compress = options.compress;
+    }
+
+    ws.identifyPayload.intents = options.intents.reduce(
+      (
+        bits,
+        next,
+      ) => (bits |= typeof next === "string"
+        ? DiscordGatewayIntents[next]
+        : next),
+      0,
+    );
+
     // Initial API connection to get info about bots connection
     ws.botGatewayData = await getGatewayBot();
+    ws.maxShards = ws.maxShards ||
+      ws.botGatewayData.shards;
+    ws.lastShardId = options.lastShardId || ws.botGatewayData.shards;
+    // Explicitly append gateway version and encoding
+    ws.botGatewayData.url += `?v=${GATEWAY_VERSION}&encoding=json`;
     proxyWSURL = ws.botGatewayData.url;
-    ws.spawnShards(data.firstShardId);
+
+    ws.spawnShards(options.firstShardId);
   }
 }
 
@@ -133,6 +121,8 @@ export interface BigBrainBotConfig extends BotConfig {
   firstShardId: number;
   /** The last shard to start for this worker. By default it will be 25 + the firstShardId. */
   lastShardId?: number;
+  /** The maximum shard Id number. Useful for zero-downtime updates or resharding. */
+  maxShards?: number;
   /** This can be used to forward the ws handling to a proxy. It will disable the sharding done by the bot side. */
   wsPort?: number;
   /** This can be used to forward the REST handling to a proxy. */

src/helpers/members/fetch_members.ts

@@ -1,9 +1,12 @@
-import { identifyPayload } from "../../bot.ts";
+import { cache } from "../../cache.ts";
 import { DiscordenoMember } from "../../structures/member.ts";
+import { DiscordGatewayOpcodes } from "../../types/codes/gateway_opcodes.ts";
 import { DiscordGatewayIntents } from "../../types/gateway/gateway_intents.ts";
-import { RequestGuildMembers } from "../../types/guilds/request_guild_members.ts";
+import type { RequestGuildMembers } from "../../types/guilds/request_guild_members.ts";
 import { Errors } from "../../types/misc/errors.ts";
 import { Collection } from "../../util/collection.ts";
+import { sendShardMessage } from "../../ws/send_shard_message.ts";
+import { ws } from "../../ws/ws.ts";
 
 /**
  * ⚠️ BEGINNER DEVS!! YOU SHOULD ALMOST NEVER NEED THIS AND YOU CAN GET FROM cache.members.get()
@@ -16,12 +19,12 @@ import { Collection } from "../../util/collection.ts";
 export function fetchMembers(
   guildId: string,
   shardId: number,
-  options?: RequestGuildMembers,
+  options?: Omit<RequestGuildMembers, "guildId">,
 ) {
   // You can request 1 member without the intent
   if (
     (!options?.limit || options.limit > 1) &&
-    !(identifyPayload.intents && DiscordGatewayIntents.GUILD_MEMBERS)
+    !(ws.identifyPayload.intents & DiscordGatewayIntents.GUILD_MEMBERS)
   ) {
     throw new Error(Errors.MISSING_INTENT_GUILD_MEMBERS);
   }
@@ -31,21 +34,20 @@ export function fetchMembers(
   }
 
   return new Promise((resolve) => {
-    return requestAllMembers(guildId, shardId, resolve, options);
+    const nonce = `${guildId}-${Date.now()}`;
+    cache.fetchAllMembersProcessingRequests.set(nonce, resolve);
+
+    sendShardMessage(shardId, {
+      op: DiscordGatewayOpcodes.RequestGuildMembers,
+      d: {
+        guild_id: guildId,
+        // If a query is provided use it, OR if a limit is NOT provided use ""
+        query: options?.query || (options?.limit ? undefined : ""),
+        limit: options?.limit || 0,
+        presences: options?.presences || false,
+        user_ids: options?.userIds,
+        nonce,
+      },
+    });
   }) as Promise<Collection<string, DiscordenoMember>>;
 }
-
-// TODO: finish implementing this
-function requestAllMembers(
-  _guildId: string,
-  _shardId: number,
-  _resolve: (
-    value:
-      | Collection<string, DiscordenoMember>
-      | PromiseLike<Collection<string, DiscordenoMember>>,
-  ) => void,
-  // deno-lint-ignore no-explicit-any
-  _options: any,
-): void {
-  throw new Error("Function not implemented.");
-}

src/helpers/members/get_members.ts

@@ -1,4 +1,4 @@
-import { eventHandlers, identifyPayload } from "../../bot.ts";
+import { eventHandlers } from "../../bot.ts";
 import { cacheHandlers } from "../../cache.ts";
 import { rest } from "../../rest/rest.ts";
 import { DiscordenoMember } from "../../structures/member.ts";
@@ -9,6 +9,7 @@ import { ListGuildMembers } from "../../types/guilds/list_guild_members.ts";
 import { Errors } from "../../types/misc/errors.ts";
 import { Collection } from "../../util/collection.ts";
 import { endpoints } from "../../util/constants.ts";
+import { ws } from "../../ws/ws.ts";
 
 /**
  * ⚠️ BEGINNER DEVS!! YOU SHOULD ALMOST NEVER NEED THIS AND YOU CAN GET FROM cache.members.get()
@@ -19,7 +20,7 @@ import { endpoints } from "../../util/constants.ts";
  * GW(fetchMembers): 120/m(PER shard) rate limit. Meaning if you have 8 shards your limit is 960/m.
  */
 export async function getMembers(guildId: string, options?: ListGuildMembers) {
-  if (!(identifyPayload.intents && DiscordGatewayIntents.GUILD_MEMBERS)) {
+  if (!(ws.identifyPayload.intents && DiscordGatewayIntents.GUILD_MEMBERS)) {
     throw new Error(Errors.MISSING_INTENT_GUILD_MEMBERS);
   }
 

src/helpers/members/get_members_by_query.ts

@@ -1,37 +0,0 @@
-import { cacheHandlers } from "../../cache.ts";
-import { DiscordenoMember } from "../../structures/member.ts";
-import { Collection } from "../../util/collection.ts";
-
-/** Returns guild member objects for the specified user by their nickname/username.
- *
- * ⚠️ **ADVANCED USE ONLY: Your members will be cached in your guild most likely. Only use this when you are absolutely sure the member is not cached.**
- */
-export async function getMembersByQuery(
-  guildId: string,
-  name: string,
-  limit = 1,
-) {
-  const guild = await cacheHandlers.get("guilds", guildId);
-  if (!guild) return;
-
-  return new Promise((resolve) => {
-    return requestAllMembers(guild.id, guild.shardId, resolve, {
-      query: name,
-      limit,
-    });
-  }) as Promise<Collection<string, DiscordenoMember>>;
-}
-
-// TODO: implement this
-function requestAllMembers(
-  _id: string,
-  _shardId: number,
-  _resolve: (
-    value:
-      | Collection<string, DiscordenoMember>
-      | PromiseLike<Collection<string, DiscordenoMember>>,
-  ) => void,
-  _arg3: { query: string; limit: number },
-): void {
-  throw new Error("Function not implemented.");
-}

src/helpers/misc/edit_bot_status.ts

@@ -1,6 +1,7 @@
 import { eventHandlers } from "../../bot.ts";
 import { DiscordGatewayOpcodes } from "../../types/codes/gateway_opcodes.ts";
 import type { StatusUpdate } from "../../types/gateway/status_update.ts";
+import { sendShardMessage } from "../../ws/send_shard_message.ts";
 import { ws } from "../../ws/ws.ts";
 
 export function editBotStatus(data: Omit<StatusUpdate, "afk" | "since">) {
@@ -10,7 +11,7 @@ export function editBotStatus(data: Omit<StatusUpdate, "afk" | "since">) {
       `Running forEach loop in editBotStatus function.`,
     );
 
-    shard.queue.push({
+    sendShardMessage(shard, {
       op: DiscordGatewayOpcodes.StatusUpdate,
       d: {
         since: null,
@@ -18,6 +19,5 @@ export function editBotStatus(data: Omit<StatusUpdate, "afk" | "since">) {
         ...data,
       },
     });
-    ws.processQueue(shard.id);
   });
 }

src/helpers/mod.ts

@@ -75,7 +75,6 @@ import { editMember } from "./members/edit_member.ts";
 import { fetchMembers } from "./members/fetch_members.ts";
 import { getMember } from "./members/get_member.ts";
 import { getMembers } from "./members/get_members.ts";
-import { getMembersByQuery } from "./members/get_members_by_query.ts";
 import { kick, kickMember } from "./members/kick_member.ts";
 import { moveMember } from "./members/move_member.ts";
 import { pruneMembers } from "./members/prune_members.ts";
@@ -203,7 +202,6 @@ export {
   getInvites,
   getMember,
   getMembers,
-  getMembersByQuery,
   getMessage,
   getMessages,
   getPins,
@@ -343,7 +341,6 @@ export let helpers = {
   editMember,
   fetchMembers,
   getMember,
-  getMembersByQuery,
   getMembers,
   kickMember,
   moveMember,

src/ws/handle_on_message.ts

@@ -8,6 +8,7 @@ import { delay, snakeKeysToCamelCase } from "../util/utils.ts";
 import { decompressWith } from "./deps.ts";
 import { identify } from "./identify.ts";
 import { resume } from "./resume.ts";
+import { sendShardMessage } from "./send_shard_message.ts";
 import { ws } from "./ws.ts";
 
 /** Handler for handling every message event from websocket. */
@@ -38,11 +39,10 @@ export async function handleOnMessage(message: any, shardId: number) {
 
       shard.heartbeat.lastSentAt = Date.now();
       // Discord randomly sends this requiring an immediate heartbeat back
-      shard.queue.unshift({
+      sendShardMessage(shard, {
         op: DiscordGatewayOpcodes.Heartbeat,
         d: shard?.previousSequenceNumber,
-      });
-      ws.processQueue(shard.id);
+      }, true);
       break;
     case DiscordGatewayOpcodes.Hello:
       ws.heartbeat(

src/ws/heartbeat.ts

@@ -12,20 +12,21 @@ export async function heartbeat(shardId: number, interval: number) {
 
   ws.log("HEARTBEATING_DETAILS", { shardId, interval, shard });
 
+  // The first heartbeat is special so we send it without setInterval: https://discord.com/developers/docs/topics/gateway#heartbeating
+  await delay(Math.floor(shard.heartbeat.interval * Math.random()));
+
+  if (shard.ws.readyState !== WebSocket.OPEN) return;
+
+  shard.ws.send(JSON.stringify({
+    op: DiscordGatewayOpcodes.Heartbeat,
+    d: shard.previousSequenceNumber,
+  }));
+
   shard.heartbeat.keepAlive = true;
   shard.heartbeat.acknowledged = false;
   shard.heartbeat.lastSentAt = Date.now();
   shard.heartbeat.interval = interval;
 
-  // The first heartbeat is special so we send it without setInterval: https://discord.com/developers/docs/topics/gateway#heartbeating
-  await delay(Math.floor(shard.heartbeat.interval * Math.random()));
-
-  shard.queue.unshift({
-    op: DiscordGatewayOpcodes.Heartbeat,
-    d: shard.previousSequenceNumber,
-  });
-  ws.processQueue(shard.id);
-
   shard.heartbeat.intervalId = setInterval(() => {
     ws.log("DEBUG", `Running setInterval in heartbeat file.`);
     const currentShard = ws.shards.get(shardId);
@@ -50,11 +51,11 @@ export async function heartbeat(shardId: number, interval: number) {
 
     if (currentShard.ws.readyState !== WebSocket.OPEN) return;
 
+    currentShard.heartbeat.acknowledged = false;
+
     currentShard.ws.send(JSON.stringify({
       op: DiscordGatewayOpcodes.Heartbeat,
       d: currentShard.previousSequenceNumber,
     }));
-
-    currentShard.heartbeat.acknowledged = false;
   }, shard.heartbeat.interval);
 }

src/ws/identify.ts

@@ -1,5 +1,6 @@
 import { DiscordGatewayOpcodes } from "../types/codes/gateway_opcodes.ts";
 import { closeWS } from "./close_ws.ts";
+import { sendShardMessage } from "./send_shard_message.ts";
 import { ws } from "./ws.ts";
 
 export async function identify(shardId: number, maxShards: number) {
@@ -40,11 +41,10 @@ export async function identify(shardId: number, maxShards: number) {
   });
 
   socket.onopen = () => {
-    ws.shards.get(shardId)?.queue.unshift({
+    sendShardMessage(shardId, {
       op: DiscordGatewayOpcodes.Identify,
       d: { ...ws.identifyPayload, shard: [shardId, maxShards] },
-    });
-    ws.processQueue(shardId);
+    }, true);
   };
 
   return new Promise((resolve, reject) => {

src/ws/resume.ts

@@ -1,13 +1,11 @@
 import { DiscordGatewayOpcodes } from "../types/codes/gateway_opcodes.ts";
 import { closeWS } from "./close_ws.ts";
+import { sendShardMessage } from "./send_shard_message.ts";
 import { ws } from "./ws.ts";
 
 export async function resume(shardId: number) {
   ws.log("RESUMING", { shardId });
 
-  // CREATE A SHARD
-  const socket = await ws.createShard(shardId);
-
   // NOW WE HANDLE RESUMING THIS SHARD
   // Get the old data for this shard necessary for resuming
   const oldShard = ws.shards.get(shardId);
@@ -19,6 +17,9 @@ export async function resume(shardId: number) {
     clearInterval(oldShard.heartbeat.intervalId);
   }
 
+  // CREATE A SHARD
+  const socket = await ws.createShard(shardId);
+
   const sessionId = oldShard?.sessionId || "";
   const previousSequenceNumber = oldShard?.previousSequenceNumber || 0;
 
@@ -47,15 +48,13 @@ export async function resume(shardId: number) {
 
   // Resume on open
   socket.onopen = () => {
-    ws.shards.get(shardId)?.queue.unshift({
+    sendShardMessage(shardId, {
       op: DiscordGatewayOpcodes.Resume,
       d: {
         token: ws.identifyPayload.token,
         session_id: sessionId,
         seq: previousSequenceNumber,
       },
-    });
-
-    ws.processQueue(shardId);
+    }, true);
   };
 }

src/ws/send_shard_message.ts

@@ -0,0 +1,18 @@
+import { DiscordenoShard, WebSocketRequest, ws } from "./ws.ts";
+
+export function sendShardMessage(
+  shard: number | DiscordenoShard,
+  message: WebSocketRequest,
+  highPriority = false,
+) {
+  if (typeof shard === "number") shard = ws.shards.get(shard)!;
+  if (!shard) return;
+
+  if (!highPriority) {
+    shard.queue.push(message);
+  } else {
+    shard.queue.unshift(message);
+  }
+
+  ws.processQueue(shard.id);
+}

src/ws/tell_cluster_to_identify.ts

@@ -1,4 +1,3 @@
-import { closeWS } from "./close_ws.ts";
 import { ws } from "./ws.ts";
 
 /** Allows users to hook in and change to communicate to different clusters across different servers or anything they like. For example using redis pubsub to talk to other servers. */
@@ -7,11 +6,5 @@ export async function tellClusterToIdentify(
   shardId: number,
   _bucketId: number,
 ) {
-  // When resharding this may exist already
-  const oldShard = ws.shards.get(shardId);
   await ws.identify(shardId, ws.maxShards);
-
-  if (oldShard) {
-    closeWS(oldShard.ws, 3063, "Resharded!");
-  }
 }