perf(backend): use websockets/ws instead of theturtle32/WebSocket-Node (#10884)

* perf(backend): use websockets/ws instead of theturtle32/WebSocket-Node

Resolve #10883

* refactor

* Update StreamingApiServerService.ts

* Update StreamingApiServerService.ts

* ✌️

* Update StreamingApiServerService.ts

* fix main stream init

* fix timing 2

* setIntervalの重複を避ける（気休め）

* add comment

* ✌️



---------

Co-authored-by: tamaina <tamaina@hotmail.co.jp>

packages/backend/package.json

@@ -35,6 +35,7 @@
 		"@swc/core-win32-x64-msvc": "1.3.56",
 		"@tensorflow/tfjs": "4.4.0",
 		"@tensorflow/tfjs-node": "4.4.0",
+		"bufferutil": "^4.0.7",
 		"slacc-android-arm-eabi": "0.0.9",
 		"slacc-android-arm64": "0.0.9",
 		"slacc-darwin-arm64": "0.0.9",
@@ -46,7 +47,8 @@
 		"slacc-linux-arm64-musl": "0.0.9",
 		"slacc-linux-x64-gnu": "0.0.9",
 		"slacc-win32-arm64-msvc": "0.0.9",
-		"slacc-win32-x64-msvc": "0.0.9"
+		"slacc-win32-x64-msvc": "0.0.9",
+		"utf-8-validate": "^6.0.3"
 	},
 	"dependencies": {
 		"@aws-sdk/client-s3": "3.321.1",
@@ -157,7 +159,6 @@
 		"uuid": "9.0.0",
 		"vary": "1.1.2",
 		"web-push": "3.6.1",
-		"websocket": "1.0.34",
 		"ws": "8.13.0",
 		"xev": "3.0.2"
 	},

packages/backend/src/server/ServerService.ts

@@ -194,7 +194,7 @@ export class ServerService implements OnApplicationShutdown {
 
 		fastify.register(this.clientServerService.createServer);
 
-		this.streamingApiServerService.attachStreamingApi(fastify.server);
+		this.streamingApiServerService.attach(fastify.server);
 
 		fastify.server.on('error', err => {
 			switch ((err as any).code) {
@@ -224,6 +224,7 @@ export class ServerService implements OnApplicationShutdown {
 
 	@bindThis
 	public async dispose(): Promise<void> {
+    await this.streamingApiServerService.detach();
 		await this.#fastify.close();
 	}
 

packages/backend/src/server/api/AuthenticateService.ts

@@ -36,7 +36,7 @@ export class AuthenticateService {
 	}
 
 	@bindThis
-	public async authenticate(token: string | null | undefined): Promise<[LocalUser | null | undefined, AccessToken | null | undefined]> {
+	public async authenticate(token: string | null | undefined): Promise<[LocalUser | null, AccessToken | null]> {
 		if (token == null) {
 			return [null, null];
 		}

packages/backend/src/server/api/StreamingApiServerService.ts

 import { EventEmitter } from 'events';
 import { Inject, Injectable } from '@nestjs/common';
 import * as Redis from 'ioredis';
-import * as websocket from 'websocket';
+import * as WebSocket from 'ws';
 import { DI } from '@/di-symbols.js';
-import type { UsersRepository, BlockingsRepository, ChannelFollowingsRepository, FollowingsRepository, MutingsRepository, UserProfilesRepository, RenoteMutingsRepository } from '@/models/index.js';
+import type { UsersRepository, AccessToken } from '@/models/index.js';
 import type { Config } from '@/config.js';
 import { NoteReadService } from '@/core/NoteReadService.js';
 import { GlobalEventService } from '@/core/GlobalEventService.js';
 import { NotificationService } from '@/core/NotificationService.js';
 import { bindThis } from '@/decorators.js';
 import { CacheService } from '@/core/CacheService.js';
-import { AuthenticateService } from './AuthenticateService.js';
+import { LocalUser } from '@/models/entities/User';
+import { AuthenticateService, AuthenticationError } from './AuthenticateService.js';
 import MainStreamConnection from './stream/index.js';
 import { ChannelsService } from './stream/ChannelsService.js';
-import type { ParsedUrlQuery } from 'querystring';
 import type * as http from 'node:http';
 
 @Injectable()
 export class StreamingApiServerService {
+	#wss: WebSocket.WebSocketServer;
+
 	constructor(
 		@Inject(DI.config)
 		private config: Config,
@@ -28,24 +30,6 @@ export class StreamingApiServerService {
 		@Inject(DI.usersRepository)
 		private usersRepository: UsersRepository,
 
-		@Inject(DI.followingsRepository)
-		private followingsRepository: FollowingsRepository,
-
-		@Inject(DI.mutingsRepository)
-		private mutingsRepository: MutingsRepository,
-
-		@Inject(DI.renoteMutingsRepository)
-		private renoteMutingsRepository: RenoteMutingsRepository,
-
-		@Inject(DI.blockingsRepository)
-		private blockingsRepository: BlockingsRepository,
-
-		@Inject(DI.channelFollowingsRepository)
-		private channelFollowingsRepository: ChannelFollowingsRepository,
-
-		@Inject(DI.userProfilesRepository)
-		private userProfilesRepository: UserProfilesRepository,
-	
 		private cacheService: CacheService,
 		private noteReadService: NoteReadService,
 		private authenticateService: AuthenticateService,
@@ -55,47 +39,75 @@ export class StreamingApiServerService {
 	}
 
 	@bindThis
-	public attachStreamingApi(server: http.Server) {
-		// Init websocket server
-		const ws = new websocket.server({
-			httpServer: server,
+	public attach(server: http.Server): void {
+		this.#wss = new WebSocket.WebSocketServer({
+			noServer: true,
 		});
 
-		ws.on('request', async (request) => {
-			const q = request.resourceURL.query as ParsedUrlQuery;
-
-			// TODO: トークンが間違ってるなどしてauthenticateに失敗したら
-			// コネクション切断するなりエラーメッセージ返すなりする
-			// (現状はエラーがキャッチされておらずサーバーのログに流れて邪魔なので)
-			const [user, miapp] = await this.authenticateService.authenticate(q.i as string);
-
-			if (user?.isSuspended) {
-				request.reject(400);
+		server.on('upgrade', async (request, socket, head) => {
+			if (request.url == null) {
+				socket.write('HTTP/1.1 400 Bad Request\r\n\r\n');
+				socket.destroy();
 				return;
 			}
 
-			const ev = new EventEmitter();
+			const q = new URL(request.url, `http://${request.headers.host}`).searchParams;
 
-			async function onRedisMessage(_: string, data: string): Promise<void> {
-				const parsed = JSON.parse(data);
-				ev.emit(parsed.channel, parsed.message);
+			let user: LocalUser | null = null;
+			let app: AccessToken | null = null;
+
+			try {
+				[user, app] = await this.authenticateService.authenticate(q.get('i'));
+			} catch (e) {
+				if (e instanceof AuthenticationError) {
+					socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
+				} else {
+					socket.write('HTTP/1.1 500 Internal Server Error\r\n\r\n');
+				}
+				socket.destroy();
+				return;
 			}
 
-			this.redisForSub.on('message', onRedisMessage);
+			if (user?.isSuspended) {
+				socket.write('HTTP/1.1 403 Forbidden\r\n\r\n');
+				socket.destroy();
+				return;
+			}
 
-			const main = new MainStreamConnection(
+			const stream = new MainStreamConnection(
 				this.channelsService,
 				this.noteReadService,
 				this.notificationService,
 				this.cacheService,
-				ev, user, miapp,
+				user, app,
 			);
 
-			await main.init();
+			await stream.init();
+
+			this.#wss.handleUpgrade(request, socket, head, (ws) => {
+				this.#wss.emit('connection', ws, request, {
+					stream, user, app,
+				});
+			});
+		});
+
+		this.#wss.on('connection', async (connection: WebSocket.WebSocket, request: http.IncomingMessage, ctx: {
+			stream: MainStreamConnection,
+			user: LocalUser | null;
+			app: AccessToken | null
+		}) => {
+			const { stream, user, app } = ctx;
+
+			const ev = new EventEmitter();
+
+			async function onRedisMessage(_: string, data: string): Promise<void> {
+				const parsed = JSON.parse(data);
+				ev.emit(parsed.channel, parsed.message);
+			}
 
-			const connection = request.accept();
+			this.redisForSub.on('message', onRedisMessage);
 
-			main.init2(connection);
+			await stream.listen(ev, connection);
 
 			const intervalId = user ? setInterval(() => {
 				this.usersRepository.update(user.id, {
@@ -110,16 +122,23 @@ export class StreamingApiServerService {
 
 			connection.once('close', () => {
 				ev.removeAllListeners();
-				main.dispose();
+				stream.dispose();
 				this.redisForSub.off('message', onRedisMessage);
 				if (intervalId) clearInterval(intervalId);
 			});
 
 			connection.on('message', async (data) => {
-				if (data.type === 'utf8' && data.utf8Data === 'ping') {
+				if (data.toString() === 'ping') {
 					connection.send('pong');
 				}
 			});
 		});
 	}
+
+	@bindThis
+	public detach(): Promise<void> {
+		return new Promise((resolve) => {
+			this.#wss.close(() => resolve());
+		});
+	}
 }

packages/backend/src/server/api/stream/index.ts

+import * as WebSocket from 'ws';
 import type { User } from '@/models/entities/User.js';
 import type { AccessToken } from '@/models/entities/AccessToken.js';
 import type { Packed } from '@/misc/json-schema.js';
@@ -7,7 +8,6 @@ import { bindThis } from '@/decorators.js';
 import { CacheService } from '@/core/CacheService.js';
 import { UserProfile } from '@/models/index.js';
 import type { ChannelsService } from './ChannelsService.js';
-import type * as websocket from 'websocket';
 import type { EventEmitter } from 'events';
 import type Channel from './channel.js';
 import type { StreamEventEmitter, StreamMessages } from './types.js';
@@ -18,7 +18,7 @@ import type { StreamEventEmitter, StreamMessages } from './types.js';
 export default class Connection {
 	public user?: User;
 	public token?: AccessToken;
-	private wsConnection: websocket.connection;
+	private wsConnection: WebSocket.WebSocket;
 	public subscriber: StreamEventEmitter;
 	private channels: Channel[] = [];
 	private subscribingNotes: any = {};
@@ -37,11 +37,9 @@ export default class Connection {
 		private notificationService: NotificationService,
 		private cacheService: CacheService,
 
-		subscriber: EventEmitter,
 		user: User | null | undefined,
 		token: AccessToken | null | undefined,
 	) {
-		this.subscriber = subscriber;
 		if (user) this.user = user;
 		if (token) this.token = token;
 	}
@@ -70,12 +68,16 @@ export default class Connection {
 		if (this.user != null) {
 			await this.fetch();
 
-			this.fetchIntervalId = setInterval(this.fetch, 1000 * 10);
+			if (!this.fetchIntervalId) {
+				this.fetchIntervalId = setInterval(this.fetch, 1000 * 10);
+			}
 		}
 	}
 
 	@bindThis
-	public async init2(wsConnection: websocket.connection) {
+	public async listen(subscriber: EventEmitter, wsConnection: WebSocket.WebSocket) {
+		this.subscriber = subscriber;
+
 		this.wsConnection = wsConnection;
 		this.wsConnection.on('message', this.onWsConnectionMessage);
 
@@ -88,14 +90,11 @@ export default class Connection {
 	 * クライアントからメッセージ受信時
 	 */
 	@bindThis
-	private async onWsConnectionMessage(data: websocket.Message) {
-		if (data.type !== 'utf8') return;
-		if (data.utf8Data == null) return;
-
+	private async onWsConnectionMessage(data: WebSocket.RawData) {
 		let obj: Record<string, any>;
 
 		try {
-			obj = JSON.parse(data.utf8Data);
+			obj = JSON.parse(data.toString());
 		} catch (e) {
 			return;
 		}

pnpm-lock.yaml

@@ -96,7 +96,7 @@ importers:
         version: 8.2.1
       '@fastify/http-proxy':
         specifier: 9.1.0
-        version: 9.1.0
+        version: 9.1.0(bufferutil@4.0.7)(utf-8-validate@6.0.3)
       '@fastify/multipart':
         specifier: 7.6.0
         version: 7.6.0
@@ -219,7 +219,7 @@ importers:
         version: 4.1.0
       jsdom:
         specifier: 21.1.1
-        version: 21.1.1
+        version: 21.1.1(bufferutil@4.0.7)(utf-8-validate@6.0.3)
       json5:
         specifier: 2.2.3
         version: 2.2.3
@@ -388,12 +388,9 @@ importers:
       web-push:
         specifier: 3.6.1
         version: 3.6.1
-      websocket:
-        specifier: 1.0.34
-        version: 1.0.34
       ws:
         specifier: 8.13.0
-        version: 8.13.0
+        version: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3)
       xev:
         specifier: 3.0.2
         version: 3.0.2
@@ -437,6 +434,9 @@ importers:
       '@tensorflow/tfjs-node':
         specifier: 4.4.0
         version: 4.4.0(seedrandom@3.0.5)
+      bufferutil:
+        specifier: ^4.0.7
+        version: 4.0.7
       slacc-android-arm-eabi:
         specifier: 0.0.9
         version: 0.0.9
@@ -473,6 +473,9 @@ importers:
       slacc-win32-x64-msvc:
         specifier: 0.0.9
         version: 0.0.9
+      utf-8-validate:
+        specifier: ^6.0.3
+        version: 6.0.3
     devDependencies:
       '@jest/globals':
         specifier: 29.5.0
@@ -3852,12 +3855,12 @@ packages:
       fast-json-stringify: 5.7.0
     dev: false
-  /@fastify/http-proxy@9.1.0:
+  /@fastify/http-proxy@9.1.0(bufferutil@4.0.7)(utf-8-validate@6.0.3):
     resolution: {integrity: sha512-vgHCTDKOqLB437zQJiLWFFnsrYfFZ6Lfwu/xXQoKqRUKIPDt+xG6LBRtf8s5MNqfFVoTE7kw1U/0qdRGDsMp4Q==}
     dependencies:
       '@fastify/reply-from': 9.0.1
       fastify-plugin: 4.5.0
-      ws: 8.13.0
+      ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3)
     transitivePeerDependencies:
       - bufferutil
       - utf-8-validate
@@ -5532,7 +5535,7 @@ packages:
       ts-dedent: 2.2.0
       util-deprecate: 1.0.2
       watchpack: 2.4.0
-      ws: 8.13.0
+      ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3)
     transitivePeerDependencies:
       - bufferutil
       - encoding
@@ -8702,7 +8705,6 @@ packages:
     requiresBuild: true
     dependencies:
       node-gyp-build: 4.6.0
-    dev: false
   /bullmq@3.14.1:
     resolution: {integrity: sha512-Fom78UKljYsnJmwbROVPx3eFLuVfQjQbw9KCnVupLzT31RQHhFHV2xd/4J4oWl4u34bZ1JmEUfNnqNBz+IOJuA==}
@@ -13872,7 +13874,7 @@ packages:
       - supports-color
     dev: true
-  /jsdom@21.1.1:
+  /jsdom@21.1.1(bufferutil@4.0.7)(utf-8-validate@6.0.3):
     resolution: {integrity: sha512-Jjgdmw48RKcdAIQyUD1UdBh2ecH7VqwaXPN3ehoZN6MqgVbMn+lRm1aAT1AsdJRAJpwfa4IpwgzySn61h2qu3w==}
     engines: {node: '>=14'}
     peerDependencies:
@@ -13905,7 +13907,7 @@ packages:
       whatwg-encoding: 2.0.0
       whatwg-mimetype: 3.0.0
       whatwg-url: 12.0.1
-      ws: 8.13.0
+      ws: 8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3)
       xml-name-validator: 4.0.0
     transitivePeerDependencies:
       - bufferutil
@@ -15231,7 +15233,7 @@ packages:
   /node-gyp-build@4.6.0:
     resolution: {integrity: sha512-NTZVKn9IylLwUzaKjkas1e4u2DLNcV4rdYagA4PWdPwW87Bi7z+BznyKSRwS/761tV/lzCGXplWsiaMjLqP2zQ==}
-    dev: false
+    hasBin: true
   /node-gyp@9.3.1:
     resolution: {integrity: sha512-4Q16ZCqq3g8awk6UplT7AuxQ35XN4R/yf/+wSAwcBUAjg7l58RTactWaP8fIDTi0FzI7YcVLujwExakZlfWkXg==}
@@ -19276,12 +19278,6 @@ packages:
     resolution: {integrity: sha512-dzlvlNlt6AXU7EBSfpAscydQ7gXB+pPGsPnfJnZpiNJBDj7IaJzQlBZYGdEi4R9HmPdBv2XmWJ6YUtoTa7lmCw==}
     dev: false
-  /typedarray-to-buffer@3.1.5:
-    resolution: {integrity: sha512-zdu8XMNEDepKKR+XYOXAVPtWui0ly0NtohUscw+UmaHiAWT8hrV1rr//H6V+0DvJ3OQ19S979M0laLfX8rm82Q==}
-    dependencies:
-      is-typedarray: 1.0.0
-    dev: false
-
   /typedarray@0.0.6:
     resolution: {integrity: sha512-/aCDEGatGvZ2BIk+HmLf4ifCJFwvKFNb9/JeZPMulfgFracn9QFcAf5GO8B/mweUjSoblS5In0cWhqpfs/5PQA==}
@@ -19656,13 +19652,12 @@ packages:
     resolution: {integrity: sha512-cwESVXlO3url9YWlFW/TA9cshCEhtu7IKJ/p5soJ/gGpj7vbvFrAY/eIioQ6Dw23KjZhYgiIo8HOs1nQ2vr/oQ==}
     engines: {node: '>=0.10.0'}
-  /utf-8-validate@5.0.10:
-    resolution: {integrity: sha512-Z6czzLq4u8fPOyx7TU6X3dvUZVvoJmxSQ+IcrlmagKhilxlhZgxPK6C5Jqbkw1IDUmFTM+cz9QDnnLTwDz/2gQ==}
+  /utf-8-validate@6.0.3:
+    resolution: {integrity: sha512-uIuGf9TWQ/y+0Lp+KGZCMuJWc3N9BHA+l/UmHd/oUHwJJDeysyTRxNQVkbzsIWfGFbRe3OcgML/i0mvVRPOyDA==}
     engines: {node: '>=6.14.2'}
     requiresBuild: true
     dependencies:
       node-gyp-build: 4.6.0
-    dev: false
   /util-deprecate@1.0.2:
     resolution: {integrity: sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==}
@@ -20132,20 +20127,6 @@ packages:
     resolution: {integrity: sha512-kyDivFZ7ZM0BVOUteVbDFhlRt7Ah/CSPwJdi8hBpkK7QLumUqdLtVfm/PX/hkcnrvr0i77fO5+TjZ94Pe+C9iw==}
     dev: false
-  /websocket@1.0.34:
-    resolution: {integrity: sha512-PRDso2sGwF6kM75QykIesBijKSVceR6jL2G8NGYyq2XrItNC2P5/qL5XeR056GhA+Ly7JMFvJb9I312mJfmqnQ==}
-    engines: {node: '>=4.0.0'}
-    dependencies:
-      bufferutil: 4.0.7
-      debug: 2.6.9
-      es5-ext: 0.10.62
-      typedarray-to-buffer: 3.1.5
-      utf-8-validate: 5.0.10
-      yaeti: 0.0.6
-    transitivePeerDependencies:
-      - supports-color
-    dev: false
-
   /well-known-symbols@2.0.0:
     resolution: {integrity: sha512-ZMjC3ho+KXo0BfJb7JgtQ5IBuvnShdlACNkKkdsqBmYw3bPAaJfPeYUo6tLUaT5tG/Gkh7xkpBhKRQ9e7pyg9Q==}
     engines: {node: '>=6'}
@@ -20336,7 +20317,7 @@ packages:
       async-limiter: 1.0.1
     dev: true
-  /ws@8.13.0:
+  /ws@8.13.0(bufferutil@4.0.7)(utf-8-validate@6.0.3):
     resolution: {integrity: sha512-x9vcZYTrFPC7aSIbj7sRCYo7L/Xb8Iy+pW0ng0wt2vCJv7M9HOMy0UoN3rr+IFC7hb7vXoqS+P9ktyLLLhO+LA==}
     engines: {node: '>=10.0.0'}
     peerDependencies:
@@ -20347,6 +20328,9 @@ packages:
         optional: true
       utf-8-validate:
         optional: true
+    dependencies:
+      bufferutil: 4.0.7
+      utf-8-validate: 6.0.3
   /xev@3.0.2:
     resolution: {integrity: sha512-8kxuH95iMXzHZj+fwqfA4UrPcYOy6bGIgfWzo9Ji23JoEc30ge/Z++Ubkiuy8c0+M64nXmmxrmJ7C8wnuBhluw==}
@@ -20394,11 +20378,6 @@ packages:
     resolution: {integrity: sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==}
     engines: {node: '>=10'}
-  /yaeti@0.0.6:
-    resolution: {integrity: sha512-MvQa//+KcZCUkBTIC9blM+CU9J2GzuTytsOUwf2lidtvkx/6gnEp1QvJv34t9vdjhFmha/mUiNDbN0D0mJWdug==}
-    engines: {node: '>=0.10.32'}
-    dev: false
-
   /yallist@2.1.2:
     resolution: {integrity: sha512-ncTzHV7NvsQZkYe1DW7cbDLm0YpzHmZF5r/iyP3ZnQtMiJ+pjzisCiMNI+Sj+xQF5pXhSHxSB3uDbsBTzY/c2A==}