daenamnu/discordeno
1
0
Fork 0
mirror of https://github.com/discordeno/discordeno.git synced 2026-09-17 16:57:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #161 from Skillz4Killz/large-bot-optimizations

Browse Source 
Big Brain Bots (Massive Bot optimizations)
This commit is contained in:
Skillz4Killz
2020-11-18 00:45:03 -05:00
committed by GitHub
parent 7c8b85cde9 9399f30523
commit 213b8dcf7d
9 changed files with 450 additions and 680 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mod.ts
-4
View File
@@ -1,5 +1,3 @@
import createClient from "./src/module/client.ts";
export * from "./src/controllers/bans.ts";
export * from "./src/controllers/cache.ts";
export * from "./src/controllers/channels.ts";
@@ -42,5 +40,3 @@ export * from "./src/utils/cdn.ts";
export * from "./src/utils/collection.ts";
export * from "./src/utils/permissions.ts";
export * from "./src/utils/utils.ts";
export default createClient;
src/module/basicShard.ts
-404
View File
@@ -1,404 +0,0 @@
import {
connectWebSocket,
delay,
inflate,
isWebSocketCloseEvent,
isWebSocketPingEvent,
isWebSocketPongEvent,
WebSocket,
} from "../../deps.ts";
import {
DiscordBotGatewayData,
DiscordHeartbeatPayload,
GatewayOpcode,
ReadyPayload,
} from "../types/discord.ts";
import { FetchMembersOptions } from "../types/guild.ts";
import { BotStatusRequest } from "../utils/utils.ts";
import { botGatewayData, eventHandlers, IdentifyPayload } from "./client.ts";
import { handleDiscordPayload } from "./shardingManager.ts";
const basicShards = new Map<number, BasicShard>();
const heartbeating = new Map<number, boolean>();
const utf8decoder = new TextDecoder();
const RequestMembersQueue: RequestMemberQueuedRequest[] = [];
let processQueue = false;
export interface BasicShard {
id: number;
socket: WebSocket;
resumeInterval: number;
sessionID: string;
previousSequenceNumber: number | null;
needToResume: boolean;
}
interface RequestMemberQueuedRequest {
guildID: string;
shardID: number;
nonce: string;
options?: FetchMembersOptions;
}
export async function createBasicShard(
data: DiscordBotGatewayData,
identifyPayload: IdentifyPayload,
resuming = false,
shardID = 0,
) {
const oldShard = basicShards.get(shardID);
const basicShard: BasicShard = {
id: shardID,
socket: await connectWebSocket(`${data.url}?v=8&encoding=json`),
resumeInterval: 0,
sessionID: oldShard?.sessionID || "",
previousSequenceNumber: oldShard?.previousSequenceNumber || 0,
needToResume: false,
};
basicShards.set(basicShard.id, basicShard);
if (!resuming) {
// Intial identify with the gateway
await identify(basicShard, identifyPayload);
} else {
await resume(basicShard, identifyPayload);
}
for await (let message of basicShard.socket) {
if (isWebSocketCloseEvent(message)) {
eventHandlers.debug?.(
{ type: "websocketClose", data: { shardID: basicShard.id, message } },
);
// These error codes should just crash the projects
if ([4004, 4005, 4012, 4013, 4014].includes(message.code)) {
console.error(`Close :( ${JSON.stringify(message)}`);
eventHandlers.debug?.(
{
type: "websocketErrored",
data: { shardID: basicShard.id, message },
},
);
throw new Error(
"Shard.ts: Error occurred that is not resumeable or able to be reconnected.",
);
}
// These error codes can not be resumed but need to reconnect from start
if ([4003, 4007, 4008, 4009].includes(message.code)) {
eventHandlers.debug?.(
{
type: "websocketReconnecting",
data: { shardID: basicShard.id, message },
},
);
createBasicShard(botGatewayData, identifyPayload, false, shardID);
} else {
basicShard.needToResume = true;
resumeConnection(botGatewayData, identifyPayload, basicShard.id);
}
continue;
} else if (isWebSocketPingEvent(message) || isWebSocketPongEvent(message)) {
continue;
}
if (message instanceof Uint8Array) {
message = inflate(
message,
0,
(slice: Uint8Array) => utf8decoder.decode(slice),
);
}
if (typeof message === "string") {
const data = JSON.parse(message);
if (!data.t) eventHandlers.rawGateway?.(data);
switch (data.op) {
case GatewayOpcode.Hello:
if (!heartbeating.has(basicShard.id)) {
heartbeat(
basicShard,
(data.d as DiscordHeartbeatPayload).heartbeat_interval,
identifyPayload,
);
}
break;
case GatewayOpcode.HeartbeatACK:
heartbeating.set(shardID, true);
break;
case GatewayOpcode.Reconnect:
eventHandlers.debug?.(
{ type: "reconnect", data: { shardID: basicShard.id } },
);
basicShard.needToResume = true;
resumeConnection(botGatewayData, identifyPayload, basicShard.id);
break;
case GatewayOpcode.InvalidSession:
eventHandlers.debug?.(
{ type: "invalidSession", data: { shardID: basicShard.id, data } },
);
// When d is false we need to reidentify
if (!data.d) {
createBasicShard(botGatewayData, identifyPayload, false, shardID);
break;
}
basicShard.needToResume = true;
resumeConnection(botGatewayData, identifyPayload, basicShard.id);
break;
default:
if (data.t === "RESUMED") {
eventHandlers.debug?.(
{ type: "resumed", data: { shardID: basicShard.id } },
);
basicShard.needToResume = false;
break;
}
// Important for RESUME
if (data.t === "READY") {
basicShard.sessionID = (data.d as ReadyPayload).session_id;
}
// Update the sequence number if it is present
if (data.s) basicShard.previousSequenceNumber = data.s;
handleDiscordPayload(data, basicShard.id);
break;
}
}
}
}
function identify(shard: BasicShard, payload: IdentifyPayload) {
eventHandlers.debug?.(
{
type: "identifying",
data: {
shardID: shard.id,
},
},
);
return shard.socket.send(
JSON.stringify(
{
op: GatewayOpcode.Identify,
d: { ...payload, shard: [shard.id, payload.shard[1]] },
},
),
);
}
function resume(shard: BasicShard, payload: IdentifyPayload) {
return shard.socket.send(JSON.stringify({
op: GatewayOpcode.Resume,
d: {
token: payload.token,
session_id: shard.sessionID,
seq: shard.previousSequenceNumber,
},
}));
}
async function heartbeat(
shard: BasicShard,
interval: number,
payload: IdentifyPayload,
) {
// We lost socket connection between heartbeats, resume connection
if (shard.socket.isClosed) {
shard.needToResume = true;
resumeConnection(botGatewayData, payload, shard.id);
heartbeating.delete(shard.id);
return;
}
if (heartbeating.has(shard.id)) {
const receivedACK = heartbeating.get(shard.id);
// If a ACK response was not received since last heartbeat, issue invalid session close
if (!receivedACK) {
eventHandlers.debug?.(
{
type: "heartbeatStopped",
data: {
interval,
previousSequenceNumber: shard.previousSequenceNumber,
shardID: shard.id,
},
},
);
return shard.socket.send(JSON.stringify({ op: 4009 }));
}
}
// Set it to false as we are issuing a new heartbeat
heartbeating.set(shard.id, false);
shard.socket.send(
JSON.stringify(
{ op: GatewayOpcode.Heartbeat, d: shard.previousSequenceNumber },
),
);
eventHandlers.debug?.(
{
type: "heartbeat",
data: {
interval,
previousSequenceNumber: shard.previousSequenceNumber,
shardID: shard.id,
},
},
);
await delay(interval);
heartbeat(shard, interval, payload);
}
async function resumeConnection(
botGatewayData: DiscordBotGatewayData,
payload: IdentifyPayload,
shardID: number,
) {
const shard = basicShards.get(shardID);
if (!shard) {
eventHandlers.debug?.(
{ type: "missingShard", data: { shardID: shardID } },
);
return;
}
if (!shard.needToResume) return;
eventHandlers.debug?.({ type: "resuming", data: { shardID: shard.id } });
// Run it once
createBasicShard(botGatewayData, payload, true, shard.id);
// Then retry every 15 seconds
await delay(1000 * 15);
if (shard.needToResume) resumeConnection(botGatewayData, payload, shardID);
}
export function requestGuildMembers(
guildID: string,
shardID: number,
nonce: string,
options?: FetchMembersOptions,
queuedRequest = false,
) {
const shard = basicShards.get(shardID);
// This request was not from this queue so we add it to queue first
if (!queuedRequest) {
RequestMembersQueue.push({
guildID,
shardID,
nonce,
options,
});
if (!processQueue) {
processQueue = true;
processGatewayQueue();
}
return;
}
// If its closed add back to queue to redo on resume
if (shard?.socket.isClosed) {
requestGuildMembers(guildID, shardID, nonce, options);
return;
}
shard?.socket.send(JSON.stringify({
op: GatewayOpcode.RequestGuildMembers,
d: {
guild_id: guildID,
query: options?.query || "",
limit: options?.limit || 0,
presences: options?.presences || false,
user_ids: options?.userIDs,
nonce,
},
}));
}
async function processGatewayQueue() {
if (!RequestMembersQueue.length) {
processQueue = false;
return;
}
basicShards.forEach((shard) => {
const index = RequestMembersQueue.findIndex((q) => q.shardID === shard.id);
// 2 events per second is the rate limit.
const request = RequestMembersQueue[index];
if (request) {
eventHandlers.debug?.(
{
type: "requestMembersProcessing",
data: {
remaining: RequestMembersQueue.length,
request,
},
},
);
requestGuildMembers(
request.guildID,
request.shardID,
request.nonce,
request.options,
true,
);
// Remove item from queue
RequestMembersQueue.splice(index, 1);
const secondIndex = RequestMembersQueue.findIndex((q) =>
q.shardID === shard.id
);
const secondRequest = RequestMembersQueue[secondIndex];
if (secondRequest) {
eventHandlers.debug?.(
{
type: "requestMembersProcessing",
data: {
remaining: RequestMembersQueue.length,
request,
},
},
);
requestGuildMembers(
secondRequest.guildID,
secondRequest.shardID,
secondRequest.nonce,
secondRequest.options,
true,
);
// Remove item from queue
RequestMembersQueue.splice(secondIndex, 1);
}
}
});
await delay(1500);
processGatewayQueue();
}
export function botGatewayStatusRequest(payload: BotStatusRequest) {
basicShards.forEach((shard) => {
shard.socket.send(JSON.stringify({
op: GatewayOpcode.StatusUpdate,
d: {
since: null,
game: payload.game.name
? {
name: payload.game.name,
type: payload.game.type,
}
: null,
status: payload.status,
afk: false,
},
}));
});
}
src/module/client.ts
+58 -4
View File
@@ -1,6 +1,6 @@
import { DiscordBotGatewayData } from "../types/discord.ts";
import { ClientOptions, EventHandlers } from "../types/options.ts";
import { endpoints } from "../utils/constants.ts";
import { baseEndpoints, endpoints } from "../utils/constants.ts";
import { RequestManager } from "./requestManager.ts";
import { spawnShards } from "./shardingManager.ts";
@@ -10,6 +10,7 @@ export let botID = "";
export let eventHandlers: EventHandlers = {};
export let botGatewayData: DiscordBotGatewayData;
export let proxyWSURL = "";
export const identifyPayload: IdentifyPayload = {
token: "",
@@ -35,7 +36,7 @@ export interface IdentifyPayload {
shard: [number, number];
}
export const createClient = async (data: ClientOptions) => {
export async function createClient(data: ClientOptions) {
if (data.eventHandlers) eventHandlers = data.eventHandlers;
authorization = `Bot ${data.token}`;
@@ -51,8 +52,8 @@ export const createClient = async (data: ClientOptions) => {
);
identifyPayload.shard = [0, botGatewayData.shards];
spawnShards(botGatewayData, identifyPayload);
};
spawnShards(botGatewayData, identifyPayload, 0, botGatewayData.shards);
}
export default createClient;
@@ -63,3 +64,56 @@ export function updateEventHandlers(newEventHandlers: EventHandlers) {
export function setBotID(id: string) {
if (botID !== id) botID = id;
}
// BIG BRAIN BOT STUFF ONLY BELOW THIS
/**
* This function should be used only by bot developers whose bots are in over 25,000 servers.
* Please be aware if you are a beginner developer using this, things will not work as per the guides. This is for advanced developers only!
*
* 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: BigBrainBotOptions) {
authorization = `Bot ${data.token}`;
identifyPayload.token = `Bot ${data.token}`;
if (data.restURL) baseEndpoints.BASE_URL = data.restURL;
if (data.cdnURL) baseEndpoints.CDN_URL = data.cdnURL;
if (data.wsURL) proxyWSURL = data.wsURL;
if (data.eventHandlers) eventHandlers = data.eventHandlers;
if (data.compress) {
identifyPayload.compress = data.compress;
}
identifyPayload.intents = data.intents.reduce(
(bits, next) => (bits |= next),
0,
);
// Initial API connection to get info about bots connection
botGatewayData = await RequestManager.get(
endpoints.GATEWAY_BOT,
) as DiscordBotGatewayData;
spawnShards(
botGatewayData,
identifyPayload,
data.firstShardID,
data.lastShardID || botGatewayData.shards >= 25
? (data.firstShardID + 25)
: botGatewayData.shards,
);
}
export interface BigBrainBotOptions extends ClientOptions {
/** The first shard to start at for this worker. Use this to control which shards to run in each worker. */
firstShardID: number;
/** The last shard to start for this worker. By default it will be 25 + the firstShardID. */
lastShardID?: number;
/** This can be used to forward the ws handling to a proxy. */
wsURL?: string;
/** This can be used to forward the REST handling to a proxy. */
restURL?: string;
/** This can be used to forward the CDN handling to a proxy. */
cdnURL?: string;
}
src/module/requestManager.ts
+16 -2
View File
@@ -2,7 +2,7 @@ import { delay } from "../../deps.ts";
import { HttpResponseCode } from "../types/discord.ts";
import { Errors } from "../types/errors.ts";
import { RequestMethods } from "../types/fetch.ts";
import { baseEndpoints } from "../utils/constants.ts";
import { baseEndpoints, discordAPIURLS } from "../utils/constants.ts";
import { authorization, eventHandlers } from "./client.ts";
const pathQueues: { [key: string]: QueuedRequest[] } = {};
@@ -144,7 +144,7 @@ function createRequestBody(body: any, method: RequestMethods) {
const headers: { [key: string]: string } = {
Authorization: authorization,
"User-Agent":
`DiscordBot (https://github.com/skillz4killz/discordeno, 6.0.0)`,
`DiscordBot (https://github.com/skillz4killz/discordeno, v10)`,
};
if (method === "get") body = undefined;
@@ -203,6 +203,20 @@ async function runMethod(
const errorStack = new Error("Location:");
Error.captureStackTrace(errorStack);
// For proxies we don't need to do any of the legwork so we just forward the request
if (
!url.startsWith(discordAPIURLS.BASE_URL) &&
!url.startsWith(discordAPIURLS.CDN_URL)
) {
return fetch(url, { method, body: body ? JSON.stringify(body) : undefined })
.then((res) => res.json())
.catch((error) => {
console.error(error);
throw errorStack;
});
}
// No proxy so we need to handl all rate limiting and such
return new Promise((resolve, reject) => {
const callback = async () => {
try {
src/module/shard.ts
+314 -181
View File
@@ -1,9 +1,13 @@
import {
connectWebSocket,
delay,
inflate,
isWebSocketCloseEvent,
isWebSocketPingEvent,
isWebSocketPongEvent,
WebSocket,
} from "../../deps.ts";
import { eventHandlers } from "../../mod.ts";
import {
DiscordBotGatewayData,
DiscordHeartbeatPayload,
@@ -11,167 +15,75 @@ import {
ReadyPayload,
} from "../types/discord.ts";
import { FetchMembersOptions } from "../types/guild.ts";
import { DebugArg } from "../types/options.ts";
let shardSocket: WebSocket;
/** The session id is needed for RESUME functionality when discord disconnects randomly. */
let sessionID = "";
// Discord requests null if no number has yet been sent by discord
let previousSequenceNumber: number | null = null;
let needToResume = false;
let shardID = 0;
import { BotStatusRequest } from "../utils/utils.ts";
import { IdentifyPayload, proxyWSURL } from "./client.ts";
import { handleDiscordPayload } from "./shardingManager.ts";
const basicShards = new Map<number, BasicShard>();
const heartbeating = new Map<number, boolean>();
const utf8decoder = new TextDecoder();
const RequestMembersQueue: RequestMemberQueuedRequest[] = [];
let processQueue = false;
export interface BasicShard {
id: number;
socket: WebSocket;
resumeInterval: number;
sessionID: string;
previousSequenceNumber: number | null;
needToResume: boolean;
}
interface RequestMemberQueuedRequest {
guildID: string;
shardID: number;
nonce: string;
options?: FetchMembersOptions;
}
async function processRequestMembersQueue() {
if (!RequestMembersQueue.length) {
processQueue = false;
return;
}
// 2 events per second is the rate limit.
const request = RequestMembersQueue.shift();
if (request) {
requestGuildMembers(request.guildID, request.nonce, request.options, true);
const secondRequest = RequestMembersQueue.shift();
if (secondRequest) {
requestGuildMembers(
secondRequest.guildID,
secondRequest.nonce,
secondRequest.options,
true,
);
}
}
await delay(1500);
postDebug(
{
type: "requestMembersProcessing",
data: { shardID, remaining: RequestMembersQueue.length },
},
);
processRequestMembersQueue();
}
// TODO: If a client does not receive a heartbeat ack between its attempts at sending heartbeats, it should immediately terminate the connection with a non-1000 close code, reconnect, and attempt to resume.
async function sendConstantHeartbeats(
interval: number,
) {
await delay(interval);
shardSocket.send(
JSON.stringify({ op: GatewayOpcode.Heartbeat, d: previousSequenceNumber }),
);
postDebug(
{ type: "heartbeat", data: { interval, previousSequenceNumber, shardID } },
);
sendConstantHeartbeats(interval);
}
async function resumeConnection(
botGatewayData: DiscordBotGatewayData,
identifyPayload: object,
) {
postDebug({ type: "resuming", data: { shardID } });
// Run it once
createShard(botGatewayData, identifyPayload, true);
// Then retry every 15 seconds
await delay(1000 * 15);
if (needToResume) resumeConnection(botGatewayData, identifyPayload);
}
const createShard = async (
botGatewayData: DiscordBotGatewayData,
identifyPayload: object,
export async function createShard(
data: DiscordBotGatewayData,
identifyPayload: IdentifyPayload,
resuming = false,
) => {
postDebug({ type: "createShard", data: { shardID } });
shardID = 0,
) {
const oldShard = basicShards.get(shardID);
shardSocket = await connectWebSocket(botGatewayData.url);
let resumeInterval = 0;
const basicShard: BasicShard = {
id: shardID,
socket: await connectWebSocket(
proxyWSURL || `${data.url}?v=8&encoding=json`,
),
resumeInterval: 0,
sessionID: oldShard?.sessionID || "",
previousSequenceNumber: oldShard?.previousSequenceNumber || 0,
needToResume: false,
};
basicShards.set(basicShard.id, basicShard);
if (!resuming) {
// Intial identify with the gateway
await shardSocket.send(
JSON.stringify({ op: GatewayOpcode.Identify, d: identifyPayload }),
);
await identify(basicShard, identifyPayload);
} else {
await shardSocket.send(JSON.stringify({
op: GatewayOpcode.Resume,
d: {
...identifyPayload,
session_id: sessionID,
seq: previousSequenceNumber,
},
}));
await resume(basicShard, identifyPayload);
}
for await (const message of shardSocket) {
if (typeof message === "string") {
const data = JSON.parse(message);
switch (data.op) {
case GatewayOpcode.Hello:
sendConstantHeartbeats(
(data.d as DiscordHeartbeatPayload).heartbeat_interval,
);
break;
case GatewayOpcode.Reconnect:
case GatewayOpcode.InvalidSession:
// When d is false we need to reidentify
if (!data.d) {
postDebug({ type: "invalidSession", data: { shardID } });
createShard(botGatewayData, identifyPayload);
break;
}
needToResume = true;
resumeConnection(botGatewayData, identifyPayload);
break;
default:
if (data.t === "RESUMED") {
postDebug({ type: "resumed", data: { shardID } });
needToResume = false;
break;
}
// Important for RESUME
if (data.t === "READY") {
sessionID = (data.d as ReadyPayload).session_id;
}
// Update the sequence number if it is present
if (data.s) previousSequenceNumber = data.s;
// @ts-ignore
postMessage(
{
type: "HANDLE_DISCORD_PAYLOAD",
payload: message,
resumeInterval,
shardID,
},
);
break;
}
} else if (isWebSocketCloseEvent(message)) {
postDebug({ type: "websocketClose", data: { shardID, message } });
for await (let message of basicShard.socket) {
if (isWebSocketCloseEvent(message)) {
eventHandlers.debug?.(
{ type: "websocketClose", data: { shardID: basicShard.id, message } },
);
// These error codes should just crash the projects
if ([4004, 4005, 4012, 4013, 4014].includes(message.code)) {
console.error(`Close :( ${JSON.stringify(message)}`);
postDebug({ type: "websocketErrored", data: { shardID, message } });
eventHandlers.debug?.(
{
type: "websocketErrored",
data: { shardID: basicShard.id, message },
},
);
throw new Error(
"Shard.ts: Error occurred that is not resumeable or able to be reconnected.",
@@ -179,51 +91,235 @@ const createShard = async (
}
// These error codes can not be resumed but need to reconnect from start
if ([4003, 4007, 4008, 4009].includes(message.code)) {
postDebug(
{ type: "websocketReconnecting", data: { shardID, message } },
eventHandlers.debug?.(
{
type: "websocketReconnecting",
data: { shardID: basicShard.id, message },
},
);
createShard(botGatewayData, identifyPayload);
createShard(data, identifyPayload, false, shardID);
} else {
needToResume = true;
resumeConnection(botGatewayData, identifyPayload);
basicShard.needToResume = true;
resumeConnection(data, identifyPayload, basicShard.id);
}
continue;
} else if (isWebSocketPingEvent(message) || isWebSocketPongEvent(message)) {
continue;
}
if (message instanceof Uint8Array) {
message = inflate(
message,
0,
(slice: Uint8Array) => utf8decoder.decode(slice),
);
}
if (typeof message === "string") {
const data = JSON.parse(message);
if (!data.t) eventHandlers.rawGateway?.(data);
switch (data.op) {
case GatewayOpcode.Hello:
if (!heartbeating.has(basicShard.id)) {
heartbeat(
basicShard,
(data.d as DiscordHeartbeatPayload).heartbeat_interval,
identifyPayload,
data,
);
}
break;
case GatewayOpcode.HeartbeatACK:
heartbeating.set(shardID, true);
break;
case GatewayOpcode.Reconnect:
eventHandlers.debug?.(
{ type: "reconnect", data: { shardID: basicShard.id } },
);
basicShard.needToResume = true;
resumeConnection(data, identifyPayload, basicShard.id);
break;
case GatewayOpcode.InvalidSession:
eventHandlers.debug?.(
{ type: "invalidSession", data: { shardID: basicShard.id, data } },
);
// When d is false we need to reidentify
if (!data.d) {
createShard(data, identifyPayload, false, shardID);
break;
}
basicShard.needToResume = true;
resumeConnection(data, identifyPayload, basicShard.id);
break;
default:
if (data.t === "RESUMED") {
eventHandlers.debug?.(
{ type: "resumed", data: { shardID: basicShard.id } },
);
basicShard.needToResume = false;
break;
}
// Important for RESUME
if (data.t === "READY") {
basicShard.sessionID = (data.d as ReadyPayload).session_id;
}
// Update the sequence number if it is present
if (data.s) basicShard.previousSequenceNumber = data.s;
handleDiscordPayload(data, basicShard.id);
break;
}
}
}
};
}
function requestGuildMembers(
function identify(shard: BasicShard, payload: IdentifyPayload) {
eventHandlers.debug?.(
{
type: "identifying",
data: {
shardID: shard.id,
},
},
);
return shard.socket.send(
JSON.stringify(
{
op: GatewayOpcode.Identify,
d: { ...payload, shard: [shard.id, payload.shard[1]] },
},
),
);
}
function resume(shard: BasicShard, payload: IdentifyPayload) {
return shard.socket.send(JSON.stringify({
op: GatewayOpcode.Resume,
d: {
token: payload.token,
session_id: shard.sessionID,
seq: shard.previousSequenceNumber,
},
}));
}
async function heartbeat(
shard: BasicShard,
interval: number,
payload: IdentifyPayload,
data: DiscordBotGatewayData,
) {
// We lost socket connection between heartbeats, resume connection
if (shard.socket.isClosed) {
shard.needToResume = true;
resumeConnection(data, payload, shard.id);
heartbeating.delete(shard.id);
return;
}
if (heartbeating.has(shard.id)) {
const receivedACK = heartbeating.get(shard.id);
// If a ACK response was not received since last heartbeat, issue invalid session close
if (!receivedACK) {
eventHandlers.debug?.(
{
type: "heartbeatStopped",
data: {
interval,
previousSequenceNumber: shard.previousSequenceNumber,
shardID: shard.id,
},
},
);
return shard.socket.send(JSON.stringify({ op: 4009 }));
}
}
// Set it to false as we are issuing a new heartbeat
heartbeating.set(shard.id, false);
shard.socket.send(
JSON.stringify(
{ op: GatewayOpcode.Heartbeat, d: shard.previousSequenceNumber },
),
);
eventHandlers.debug?.(
{
type: "heartbeat",
data: {
interval,
previousSequenceNumber: shard.previousSequenceNumber,
shardID: shard.id,
},
},
);
await delay(interval);
heartbeat(shard, interval, payload, data);
}
async function resumeConnection(
data: DiscordBotGatewayData,
payload: IdentifyPayload,
shardID: number,
) {
const shard = basicShards.get(shardID);
if (!shard) {
eventHandlers.debug?.(
{ type: "missingShard", data: { shardID: shardID } },
);
return;
}
if (!shard.needToResume) return;
eventHandlers.debug?.({ type: "resuming", data: { shardID: shard.id } });
// Run it once
createShard(data, payload, true, shard.id);
// Then retry every 15 seconds
await delay(1000 * 15);
if (shard.needToResume) resumeConnection(data, payload, shardID);
}
export function requestGuildMembers(
guildID: string,
shardID: number,
nonce: string,
options?: FetchMembersOptions,
queuedRequest = false,
) {
const shard = basicShards.get(shardID);
// This request was not from this queue so we add it to queue first
if (!queuedRequest) {
RequestMembersQueue.push({
guildID,
shardID,
nonce,
options,
});
if (!processQueue) {
processQueue = true;
processRequestMembersQueue();
processGatewayQueue();
}
return;
}
// If its closed add back to queue to redo on resume
if (shardSocket.isClosed) {
requestGuildMembers(guildID, nonce, options);
if (shard?.socket.isClosed) {
requestGuildMembers(guildID, shardID, nonce, options);
return;
}
shardSocket.send(JSON.stringify({
shard?.socket.send(JSON.stringify({
op: GatewayOpcode.RequestGuildMembers,
d: {
guild_id: guildID,
query: options?.query || "",
limit: options?.query || 0,
limit: options?.limit || 0,
presences: options?.presences || false,
user_ids: options?.userIDs,
nonce,
@@ -231,46 +327,83 @@ function requestGuildMembers(
}));
}
// TODO: Errors need to be fixed by VSC plugin
// @ts-ignore
postMessage({ type: "REQUEST_CLIENT_OPTIONS" });
// @ts-ignore
onmessage = (message: MessageEvent) => {
if (message.data.type === "CREATE_SHARD") {
createShard(
message.data.botGatewayData,
message.data.identifyPayload,
);
shardID = message.data.shardID;
async function processGatewayQueue() {
if (!RequestMembersQueue.length) {
processQueue = false;
return;
}
if (message.data.type === "FETCH_MEMBERS") {
requestGuildMembers(
message.data.guildID,
message.data.nonce,
message.data.options,
);
}
basicShards.forEach((shard) => {
const index = RequestMembersQueue.findIndex((q) => q.shardID === shard.id);
// 2 events per second is the rate limit.
const request = RequestMembersQueue[index];
if (request) {
eventHandlers.debug?.(
{
type: "requestMembersProcessing",
data: {
remaining: RequestMembersQueue.length,
request,
},
},
);
requestGuildMembers(
request.guildID,
request.shardID,
request.nonce,
request.options,
true,
);
// Remove item from queue
RequestMembersQueue.splice(index, 1);
if (message.data.type === "EDIT_BOTS_STATUS") {
shardSocket.send(JSON.stringify({
const secondIndex = RequestMembersQueue.findIndex((q) =>
q.shardID === shard.id
);
const secondRequest = RequestMembersQueue[secondIndex];
if (secondRequest) {
eventHandlers.debug?.(
{
type: "requestMembersProcessing",
data: {
remaining: RequestMembersQueue.length,
request,
},
},
);
requestGuildMembers(
secondRequest.guildID,
secondRequest.shardID,
secondRequest.nonce,
secondRequest.options,
true,
);
// Remove item from queue
RequestMembersQueue.splice(secondIndex, 1);
}
}
});
await delay(1500);
processGatewayQueue();
}
export function botGatewayStatusRequest(payload: BotStatusRequest) {
basicShards.forEach((shard) => {
shard.socket.send(JSON.stringify({
op: GatewayOpcode.StatusUpdate,
d: {
since: null,
game: message.data.game.name
game: payload.game.name
? {
name: message.data.game.name,
type: message.data.game.type,
name: payload.game.name,
type: payload.game.type,
}
: null,
status: message.data.status,
status: payload.status,
afk: false,
},
}));
}
};
function postDebug(details: DebugArg) {
// TODO: Errors need to be fixed by VSC plugin
postMessage({ type: "DEBUG_LOG", details });
});
}
src/module/shardingManager.ts
+47 -83
View File
@@ -11,20 +11,11 @@ import { cache } from "../utils/cache.ts";
import { BotStatusRequest } from "../utils/utils.ts";
import {
botGatewayStatusRequest,
createBasicShard,
createShard,
requestGuildMembers,
} from "./basicShard.ts";
import {
botGatewayData,
eventHandlers,
IdentifyPayload,
identifyPayload,
} from "./client.ts";
} from "./shard.ts";
import { eventHandlers, IdentifyPayload } from "./client.ts";
let shardCounter = 0;
let basicSharding = false;
const shards: Worker[] = [];
let createNextShard = true;
/** This function is meant to be used on the ready event to alert the library to start the next shard. */
@@ -32,58 +23,48 @@ export function allowNextShard(enabled = true) {
createNextShard = enabled;
}
export function createShardWorker(shardID?: number) {
const path = new URL("./shard.ts", import.meta.url).toString();
const shard = new Worker(path, { type: "module", deno: true });
shard.onmessage = (message) => {
if (message.data.type === "REQUEST_CLIENT_OPTIONS") {
identifyPayload.shard = [
shardID || shardCounter,
botGatewayData.shards,
];
shard.postMessage(
{
type: "CREATE_SHARD",
botGatewayData,
identifyPayload,
shardID: shardCounter,
},
);
// Update the shard counter
shardCounter++;
} else if (message.data.type === "HANDLE_DISCORD_PAYLOAD") {
handleDiscordPayload(
JSON.parse(message.data.payload),
message.data.shardID,
);
} else if (message.data.type === "DEBUG_LOG") {
eventHandlers.debug?.(message.data.details);
}
};
shards.push(shard);
}
export const spawnShards = async (
export async function spawnShards(
data: DiscordBotGatewayData,
payload: IdentifyPayload,
id = 1,
) => {
if ((data.shards === 1 && id === 1) || id <= data.shards) {
if (createNextShard) {
createNextShard = false;
if (data.shards >= 25) createShardWorker();
else {
basicSharding = true;
createBasicShard(data, payload, false, id - 1);
}
spawnShards(data, payload, id + 1);
} else {
await delay(1000);
spawnShards(data, payload, id);
}
shardID: number,
lastShardID: number,
skipChecks?: number,
) {
// All shards on this worker have started! Cancel out.
if (shardID >= lastShardID) return;
if (skipChecks) {
payload.shard = [shardID, data.shards];
// Start The shard
createShard(data, payload, false, shardID);
// Spawn next shard
spawnShards(
data,
payload,
shardID + 1,
lastShardID,
skipChecks - 1,
);
return;
}
};
// Make sure we can create a shard or we are waiting for shards to connect still.
if (createNextShard) {
createNextShard = false;
// Start the next few shards based on max concurrency
spawnShards(
data,
payload,
shardID,
lastShardID,
data.session_start_limit.max_concurrency,
);
return;
}
await delay(1000);
spawnShards(data, payload, shardID, lastShardID, skipChecks);
}
export async function handleDiscordPayload(
data: DiscordPayload,
@@ -113,30 +94,13 @@ export async function requestAllMembers(
const nonce = `${guild.id}-${Math.random().toString()}`;
cache.fetchAllMembersProcessingRequests.set(nonce, resolve);
if (basicSharding) {
return requestGuildMembers(guild.id, guild.shardID, nonce, options);
}
shards[guild.shardID].postMessage({
type: "FETCH_MEMBERS",
guildID: guild.id,
nonce,
options,
});
return requestGuildMembers(guild.id, guild.shardID, nonce, options);
}
export function sendGatewayCommand(type: "EDIT_BOTS_STATUS", payload: object) {
if (basicSharding) {
if (type === "EDIT_BOTS_STATUS") {
botGatewayStatusRequest(payload as BotStatusRequest);
}
return;
if (type === "EDIT_BOTS_STATUS") {
botGatewayStatusRequest(payload as BotStatusRequest);
}
shards.forEach((shard) => {
shard.postMessage({
type,
...payload,
});
});
return;
}
src/types/discord.ts
+5
View File
@@ -57,6 +57,11 @@ export interface DiscordBotGatewayData {
remaining: number;
/** Milliseconds left until limit is reset. */
reset_after: number;
/** The number of identify requests allowed per 5 seconds.
* So, if you had a max concurrency of 16, and 16 shards for example, you could start them all up at the same time.
* Whereas if you had 32 shards, if you tried to start up shard 0 and 16 at the same time for example, it would not work. You can start shards 0-15 concurrently, then 16-31...
* */
max_concurrency: number;
};
}
src/types/options.ts
+1
View File
@@ -31,6 +31,7 @@ export interface Fulfilled_Client_Options {
export interface ClientOptions {
token: string;
/** @deprecated Will be removed in next major version! */
properties?: Properties;
compress?: boolean;
intents: Intents[];
src/utils/constants.ts
+9 -2
View File
@@ -1,8 +1,15 @@
export const baseEndpoints = {
BASE_URL: "https://discord.com/api/v8",
// These will never be modified and remain constants
export const discordAPIURLS = {
BASE_URL: `https://discord.com/api/v8`,
CDN_URL: "https://cdn.discordapp.com",
};
// This can be modified by big brain bots and use a proxy
export const baseEndpoints = {
BASE_URL: discordAPIURLS.BASE_URL,
CDN_URL: discordAPIURLS.CDN_URL,
};
const GUILDS_BASE = (id: string) => `${baseEndpoints.BASE_URL}/guilds/${id}`;
export const endpoints = {