feat: division

packages/core/package.json

@@ -36,10 +36,12 @@
     "url": "https://github.com/Androz2091/discord-player/issues"
   },
   "devDependencies": {
-    "@discord-player/tsconfig": "workspace:^"
+    "@discord-player/tsconfig": "workspace:^",
+    "tsup": "^8.1.0"
   },
   "dependencies": {
     "@discord-player/utils": "workspace:^",
+    "discord-api-types": "^0.37.92",
     "discord-voip": "workspace:^"
   }
 }

packages/core/src/common/constants.ts

@@ -1,9 +1,14 @@
 export enum WorkerOp {
     OP_PING,
     OP_JOIN_VOICE_CHANNEL,
+    OP_VOICE_SERVER_UPDATE,
+    OP_VOICE_STATE_UPDATE,
 }
 
 export enum WorkerAckOp {
+    OP_EVT_DEBUG,
     OP_ACK_PING,
     OP_ACK_JOIN_VOICE_CHANNEL,
+    OP_EVT_GATEWAY_DISPATCH,
+    OP_EVT_STATS,
 }

packages/core/src/common/types.ts

@@ -0,0 +1,37 @@
+import { WorkerAckOp, WorkerOp } from './constants';
+import type { VoiceConnectionCreateOptions } from '../worker/node/AudioNode';
+import type { GatewayVoiceStateUpdateDispatchData, GatewayVoiceServerUpdateDispatchData } from 'discord-api-types/v10';
+
+export interface WorkerMessage<T> {
+    op: WorkerOp;
+    d: T;
+}
+
+export interface WorkerMessageAck<T> {
+    t: WorkerAckOp;
+    d: T;
+}
+
+export interface DebugEventDispatch {
+    data: string;
+}
+
+export interface WorkerStatistics {
+    memoryUsed: number;
+    subscriptions: number;
+}
+
+export interface WorkerMessagePayloadAck {
+    [WorkerAckOp.OP_EVT_DEBUG]: DebugEventDispatch;
+    [WorkerAckOp.OP_ACK_PING]: null;
+    [WorkerAckOp.OP_ACK_JOIN_VOICE_CHANNEL]: null;
+    [WorkerAckOp.OP_EVT_GATEWAY_DISPATCH]: null;
+    [WorkerAckOp.OP_EVT_STATS]: WorkerStatistics;
+}
+
+export interface WorkerMessagePayload {
+    [WorkerOp.OP_PING]: null;
+    [WorkerOp.OP_JOIN_VOICE_CHANNEL]: VoiceConnectionCreateOptions;
+    [WorkerOp.OP_VOICE_SERVER_UPDATE]: GatewayVoiceServerUpdateDispatchData;
+    [WorkerOp.OP_VOICE_STATE_UPDATE]: GatewayVoiceStateUpdateDispatchData;
+}

packages/core/src/index.ts

@@ -1,5 +1,8 @@
-export async function createServer() {
-  // TODO
+import { type WorkerConfig, WorkerManager } from './workers';
+
+export async function createWorker(config: WorkerConfig) {
+    const manager = new WorkerManager(config);
+    return manager;
 }
 
 // eslint-disable-next-line @typescript-eslint/no-inferrable-types

packages/core/src/worker/entrypoint.ts

@@ -1,25 +1,13 @@
-import { isMainThread, parentPort } from 'node:worker_threads';
+import { isMainThread, parentPort, workerData } from 'node:worker_threads';
 import { WorkerAckOp, WorkerOp } from '../common/constants';
 import { AudioNodeManager } from './node/AudioNodeManager';
-import { VoiceConnectionCreateOptions } from './node/AudioNode';
+import { WorkerMessage, WorkerMessageAck, WorkerMessagePayload } from '../common/types';
 
-interface WorkerMessage<T> {
-    op: WorkerOp;
-    d: T;
-}
-
-interface WorkerMessageAck<T> {
-    op: WorkerAckOp;
-    d: T;
-}
-
-interface WorkerMessagePayload {
-    [WorkerOp.OP_PING]: null;
-    [WorkerOp.OP_JOIN_VOICE_CHANNEL]: VoiceConnectionCreateOptions;
-}
+const { STATS_DISPATCH_INTERVAL, KEEP_ALIVE } = workerData;
 
 export class WorkerListener {
     private readonly audioNodes: AudioNodeManager;
+    private statsClock: NodeJS.Timeout | null = null;
 
     public constructor() {
         if (!parentPort) throw new Error('This script must be run as a worker');
@@ -34,6 +22,29 @@ export class WorkerListener {
                 //
             }
         });
+
+        this.createStatsClock();
+    }
+
+    private createStatsClock() {
+        // ignore invalid cases
+        if (typeof STATS_DISPATCH_INTERVAL !== 'number' || Number.isNaN(STATS_DISPATCH_INTERVAL)) return;
+
+        if (this.statsClock) clearInterval(this.statsClock);
+
+        this.statsClock = setInterval(() => {
+            this.send({
+                t: WorkerAckOp.OP_EVT_STATS,
+                d: {
+                    memoryUsed: process.memoryUsage().heapUsed,
+                    subscriptions: this.audioNodes.nodes.size,
+                },
+            });
+        }, STATS_DISPATCH_INTERVAL);
+
+        if (KEEP_ALIVE !== true) {
+            this.statsClock.unref();
+        }
     }
 
     private validate(message: WorkerMessage<unknown>) {
@@ -50,9 +61,21 @@ export class WorkerListener {
     private async handleMessage(message: WorkerMessage<unknown>) {
         switch (true) {
             case assertPayload(message, WorkerOp.OP_PING):
-                return this.send({ op: WorkerAckOp.OP_ACK_PING, d: null });
+                return this.send({ t: WorkerAckOp.OP_ACK_PING, d: null });
             case assertPayload(message, WorkerOp.OP_JOIN_VOICE_CHANNEL):
                 return this.audioNodes.connect(message.d);
+            case assertPayload(message, WorkerOp.OP_VOICE_SERVER_UPDATE): {
+                const adapter = this.audioNodes.transmitters.get(message.d.guild_id);
+                if (!adapter) return;
+                adapter.onVoiceServerUpdate(message.d);
+                return;
+            }
+            case assertPayload(message, WorkerOp.OP_VOICE_STATE_UPDATE): {
+                const adapter = this.audioNodes.transmitters.get(message.d.guild_id as string);
+                if (!adapter) return;
+                adapter.onVoiceStateUpdate(message.d);
+                return;
+            }
         }
     }
 }

packages/core/src/workers.ts

@@ -1,15 +1,37 @@
 import { Worker } from 'node:worker_threads';
 import { Collection, EventEmitter } from '@discord-player/utils';
+import { WorkerMessage, WorkerMessageAck, WorkerMessagePayload, WorkerMessagePayloadAck } from './common/types';
+import { WorkerAckOp, WorkerOp } from './common/constants';
+
+export interface IWorkerMessage<T> {
+    metadata: {
+        clientId: string;
+        channelId: string;
+        guildId: string;
+    };
+    payload: T;
+}
 
-interface IWorkerStats {
+export interface IWorkerStats {
     memoryUsed: number;
     subscriptions: number;
 }
 
-interface WorkerInfo {
+export interface IClientInformation {
+    clientId: string;
+    guildId: string;
+    channelId: string;
+}
+
+/**
+ * A collection of clients. The root key is the client id, the second key is the channel id and the value is the client information.
+ */
+export type IClientCollection = Collection<string, Collection<string, IClientInformation>>;
+
+export interface WorkerInfo {
     worker: Worker;
     lastAccess: number;
-    estimatedClients: number;
+    clients: IClientCollection;
     stats: IWorkerStats;
 }
 
@@ -22,58 +44,104 @@ export enum WorkerDistributionMode {
 export interface WorkerConfig {
     maxWorkers: number;
     distributionMode: WorkerDistributionMode;
+    statsDispatchInterval?: number;
 }
 
 export interface WorkerEvents {
     error: (error: Error) => void;
-    exit: () => void;
-    ready: () => void;
+    payload: (payload: WorkerMessageAck<WorkerMessagePayloadAck[WorkerAckOp.OP_EVT_GATEWAY_DISPATCH]>) => void;
+    debug: (data: string) => void;
 }
 
+const WorkerCreatable = new Set([WorkerOp.OP_JOIN_VOICE_CHANNEL]);
+const STATS_DISPATCH_INTERVAL = 5000;
+
+const assertPayload = <T extends WorkerAckOp>(
+    message: WorkerMessageAck<unknown>,
+    t: T,
+): message is WorkerMessageAck<WorkerMessagePayloadAck[T]> => {
+    return message.t === t;
+};
+
 export class WorkerManager extends EventEmitter<WorkerEvents> {
+    /**
+     * The workers collection.
+     */
     private workers = new Collection<number, WorkerInfo>();
 
+    /**
+     * Creates a new worker manager.
+     * @param config The worker configuration.
+     */
     public constructor(public readonly config: WorkerConfig) {
         super();
     }
 
+    /**
+     * Write a debug message.
+     * @param data The data to write.
+     */
+    public debug(data: string) {
+        this.emit('debug', data);
+    }
+
+    /**
+     * Returns The optimal worker.
+     */
     public getOptimalWorker(): WorkerInfo | undefined {
         if (this.workers.size === 0) return undefined;
 
         switch (this.config.distributionMode) {
             case WorkerDistributionMode.LeastLoad:
                 return this.workers.sort((a, b) => a.stats.memoryUsed - b.stats.memoryUsed).first();
             case WorkerDistributionMode.Balanced:
-                return this.workers.sort((a, b) => a.estimatedClients - b.estimatedClients).first();
+                return this.workers.sort((a, b) => a.clients.size - b.clients.size).first();
             case WorkerDistributionMode.Random:
                 return this.workers.random();
         }
     }
 
+    /**
+     * Whether a new worker can be created.
+     */
     public canCreateWorker(): boolean {
         return this.workers.size < this.config.maxWorkers;
     }
 
-    public create() {
+    /**
+     * Create a new worker if needed.
+     * @param [op] The operation to create the worker for.
+     * @returns The worker info.
+     */
+    public create(op?: WorkerOp) {
+        if (op != null && !WorkerCreatable.has(op)) return;
         if (!this.canCreateWorker()) return this.getOptimalWorker()!;
 
-        const thread = new Worker('./worker/entrypoint.js');
+        const thread = new Worker('./worker/entrypoint.js', {
+            workerData: {
+                STATS_DISPATCH_INTERVAL: this.config.statsDispatchInterval ?? STATS_DISPATCH_INTERVAL,
+                KEEP_ALIVE: true,
+            },
+        });
+
+        thread.on('message', this.handleMessage.bind(this));
 
         thread.on('online', () => {
-            this.emit('ready');
+            this.debug(`Worker ${thread.threadId} is online`);
         });
 
         thread.on('error', (error) => {
-            void error;
+            this.debug(`Worker ${thread.threadId} encountered an error: ${error.message}`);
         });
 
         thread.once('exit', () => {
+            this.debug(`Worker ${thread.threadId} exited`);
             this.workers.delete(thread.threadId);
         });
 
         const workerInfo: WorkerInfo = {
             worker: thread,
-            estimatedClients: 0,
+            clients: new Collection(),
             stats: {
                 memoryUsed: 0,
                 subscriptions: 0,
@@ -85,4 +153,54 @@ export class WorkerManager extends EventEmitter<WorkerEvents> {
 
         return workerInfo;
     }
+
+    /**
+     * Handle a message from the worker.
+     * @param message The message to handle.
+     */
+    public handleMessage(message: WorkerMessageAck<WorkerMessagePayloadAck>) {
+        switch (true) {
+            case assertPayload(message, WorkerAckOp.OP_EVT_GATEWAY_DISPATCH):
+                return this.emit('payload', message);
+            case assertPayload(message, WorkerAckOp.OP_EVT_DEBUG):
+                // eslint-disable-next-line no-console
+                return console.log(message.d.data);
+        }
+    }
+
+    /**
+     * Find the worker associated with the client and channel.
+     * @param clientId The client id.
+     * @param channelId The channel id.
+     * @returns The worker info.
+     */
+    public findAssociatedWorker(clientId: string, channelId: string): WorkerInfo | undefined {
+        return this.workers.find((worker) => !!worker.clients.get(clientId)?.has(channelId));
+    }
+
+    /**
+     * Send a message to the worker.
+     * @param message The message to send.
+     */
+    public send<Op extends WorkerOp>(message: IWorkerMessage<WorkerMessage<WorkerMessagePayload[Op]>>) {
+        const { metadata, payload } = message;
+
+        const worker = this.findAssociatedWorker(metadata.clientId, metadata.channelId) || this.create(payload.op);
+
+        if (!worker) return;
+
+        worker.worker.postMessage(payload);
+    }
+
+    /**
+     * Send a message to all workers.
+     * @param message The message to send.
+     */
+    public broadcast<Op extends WorkerOp>(message: IWorkerMessage<WorkerMessage<WorkerMessagePayload[Op]>>) {
+        const { payload } = message;
+
+        for (const worker of this.workers.values()) {
+            worker.worker.postMessage(payload);
+        }
+    }
 }

packages/server/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 Androz2091
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.