add webrtc support

package.json

@@ -67,6 +67,7 @@
 		"husky": "^9.1.7",
 		"prettier": "^3.5.3",
 		"pretty-quick": "^4.1.1",
+		"spacebar-webrtc-types": "github:dank074/spacebar-webrtc-types",
 		"typescript": "^5.8.3"
 	},
 	"dependencies": {
@@ -118,7 +119,8 @@
 		"@spacebar/api": "dist/api",
 		"@spacebar/cdn": "dist/cdn",
 		"@spacebar/gateway": "dist/gateway",
-		"@spacebar/util": "dist/util"
+		"@spacebar/util": "dist/util",
+		"@spacebar/webrtc": "dist/webrtc"
 	},
 	"optionalDependencies": {
 		"@yukikaze-bot/erlpack": "^1.0.1",

src/api/routes/guilds/#guild_id/voice-states/#user_id/index.ts

@@ -93,7 +93,7 @@ router.patch(
 			voice_state.save(),
 			emitEvent({
 				event: "VOICE_STATE_UPDATE",
-				data: voice_state,
+				data: voice_state.toPublicVoiceState(),
 				guild_id,
 			} as VoiceStateUpdateEvent),
 		]);

src/bundle/Server.ts

@@ -22,6 +22,7 @@ process.on("uncaughtException", console.error);
 import http from "http";
 import * as Api from "@spacebar/api";
 import * as Gateway from "@spacebar/gateway";
+import * as Webrtc from "@spacebar/webrtc";
 import { CDNServer } from "@spacebar/cdn";
 import express from "express";
 import { green, bold } from "picocolors";
@@ -30,18 +31,25 @@ import { Config, initDatabase, Sentry } from "@spacebar/util";
 const app = express();
 const server = http.createServer();
 const port = Number(process.env.PORT) || 3001;
+const wrtcWsPort = Number(process.env.WRTC_WS_PORT) || 3004;
 const production = process.env.NODE_ENV == "development" ? false : true;
 server.on("request", app);
 
 const api = new Api.SpacebarServer({ server, port, production, app });
 const cdn = new CDNServer({ server, port, production, app });
 const gateway = new Gateway.Server({ server, port, production });
+const webrtc = new Webrtc.Server({
+	server: undefined,
+	port: wrtcWsPort,
+	production,
+});
 
 process.on("SIGTERM", async () => {
 	console.log("Shutting down due to SIGTERM");
 	await gateway.stop();
 	await cdn.stop();
 	await api.stop();
+	await webrtc.stop();
 	server.close();
 	Sentry.close();
 });
@@ -54,7 +62,12 @@ async function main() {
 	await new Promise((resolve) =>
 		server.listen({ port }, () => resolve(undefined)),
 	);
-	await Promise.all([api.start(), cdn.start(), gateway.start()]);
+	await Promise.all([
+		api.start(),
+		cdn.start(),
+		gateway.start(),
+		webrtc.start(),
+	]);
 
 	Sentry.errorHandler(app);
 

src/gateway/events/Close.ts

@@ -24,6 +24,8 @@ import {
 	Session,
 	SessionsReplace,
 	User,
+	VoiceState,
+	VoiceStateUpdateEvent,
 } from "@spacebar/util";
 
 export async function Close(this: WebSocket, code: number, reason: Buffer) {
@@ -36,6 +38,39 @@ export async function Close(this: WebSocket, code: number, reason: Buffer) {
 
 	if (this.session_id) {
 		await Session.delete({ session_id: this.session_id });
+
+		const voiceState = await VoiceState.findOne({
+			where: { user_id: this.user_id },
+		});
+
+		// clear the voice state for this session if user was in voice channel
+		if (
+			voiceState &&
+			voiceState.session_id === this.session_id &&
+			voiceState.channel_id
+		) {
+			const prevGuildId = voiceState.guild_id;
+			const prevChannelId = voiceState.channel_id;
+
+			// @ts-expect-error channel_id is nullable
+			voiceState.channel_id = null;
+			// @ts-expect-error guild_id is nullable
+			voiceState.guild_id = null;
+			voiceState.self_stream = false;
+			voiceState.self_video = false;
+			await voiceState.save();
+
+			// let the users in previous guild/channel know that user disconnected
+			await emitEvent({
+				event: "VOICE_STATE_UPDATE",
+				data: {
+					...voiceState.toPublicVoiceState(),
+					guild_id: prevGuildId, // have to send the previous guild_id because that's what client expects for disconnect messages
+				},
+				guild_id: prevGuildId,
+				channel_id: prevChannelId,
+			} as VoiceStateUpdateEvent);
+		}
 	}
 
 	if (this.user_id) {

src/gateway/opcodes/Identify.ts

@@ -183,6 +183,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
 					"guild.emojis",
 					"guild.roles",
 					"guild.stickers",
+					"guild.voice_states",
 					"roles",
 
 					// For these entities, `user` is always just the logged in user we fetched above
@@ -485,6 +486,18 @@ export async function onIdentify(this: WebSocket, data: Payload) {
 		}),
 	);
 
+	const readySupplementalGuilds = (
+		guilds.filter((guild) => !guild.unavailable) as Guild[]
+	).map((guild) => {
+		return {
+			voice_states: guild.voice_states.map((state) =>
+				state.toPublicVoiceState(),
+			),
+			id: guild.id,
+			embedded_activities: [],
+		};
+	});
+
 	// TODO: ready supplemental
 	await Send(this, {
 		op: OPCodes.DISPATCH,
@@ -498,7 +511,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
 			// these merged members seem to be all users currently in vc in your guilds
 			merged_members: [],
 			lazy_private_channels: [],
-			guilds: [], // { voice_states: [], id: string, embedded_activities: [] }
+			guilds: readySupplementalGuilds, // { voice_states: [], id: string, embedded_activities: [] }
 			// embedded_activities are users currently in an activity?
 			disclose: [], // Config.get().general.uniqueUsernames ? ["pomelo"] : []
 		},

src/gateway/opcodes/StreamCreate.ts

@@ -0,0 +1,131 @@
+import {
+	genVoiceToken,
+	Payload,
+	WebSocket,
+	generateStreamKey,
+} from "@spacebar/gateway";
+import {
+	Channel,
+	Config,
+	emitEvent,
+	Member,
+	Region,
+	Snowflake,
+	Stream,
+	StreamCreateEvent,
+	StreamCreateSchema,
+	StreamServerUpdateEvent,
+	StreamSession,
+	VoiceState,
+	VoiceStateUpdateEvent,
+} from "@spacebar/util";
+import { check } from "./instanceOf";
+
+export async function onStreamCreate(this: WebSocket, data: Payload) {
+	check.call(this, StreamCreateSchema, data.d);
+	const body = data.d as StreamCreateSchema;
+
+	if (body.channel_id.trim().length === 0) return;
+
+	// first check if we are in a voice channel already. cannot create a stream if there's no existing voice connection
+	const voiceState = await VoiceState.findOne({
+		where: { user_id: this.user_id },
+	});
+
+	if (!voiceState || !voiceState.channel_id) return;
+
+	if (body.guild_id) {
+		voiceState.member = await Member.findOneOrFail({
+			where: { id: voiceState.user_id, guild_id: voiceState.guild_id },
+			relations: ["user", "roles"],
+		});
+	}
+
+	// TODO: permissions check - if it's a guild, check if user is allowed to create stream in this guild
+
+	const channel = await Channel.findOne({
+		where: { id: body.channel_id },
+	});
+
+	if (
+		!channel ||
+		(body.type === "guild" && channel.guild_id != body.guild_id)
+	)
+		return this.close(4000, "invalid channel");
+
+	// TODO: actually apply preferred_region from the event payload
+	const regions = Config.get().regions;
+	const guildRegion = regions.available.filter(
+		(r) => r.id === regions.default,
+	)[0];
+
+	// first make sure theres no other streams for this user that somehow didnt get cleared
+	await Stream.delete({
+		owner_id: this.user_id,
+	});
+
+	// create a new entry in db containing the token for authenticating user in stream gateway IDENTIFY
+	const stream = Stream.create({
+		id: Snowflake.generate(),
+		owner_id: this.user_id,
+		channel_id: body.channel_id,
+		endpoint: guildRegion.endpoint,
+	});
+
+	await stream.save();
+
+	const token = genVoiceToken();
+
+	const streamSession = StreamSession.create({
+		stream_id: stream.id,
+		user_id: this.user_id,
+		session_id: this.session_id,
+		token,
+	});
+
+	await streamSession.save();
+
+	const streamKey = generateStreamKey(
+		body.type,
+		body.guild_id,
+		body.channel_id,
+		this.user_id,
+	);
+
+	await emitEvent({
+		event: "STREAM_CREATE",
+		data: {
+			stream_key: streamKey,
+			rtc_server_id: stream.id, // for voice connections in guilds it is guild_id, for dm voice calls it seems to be DM channel id, for GoLive streams a generated number
+			viewer_ids: [],
+			region: guildRegion.name,
+			paused: false,
+		},
+		user_id: this.user_id,
+	} as StreamCreateEvent);
+
+	await emitEvent({
+		event: "STREAM_SERVER_UPDATE",
+		data: {
+			token: streamSession.token,
+			stream_key: streamKey,
+			guild_id: null, // not sure why its always null
+			endpoint: stream.endpoint,
+		},
+		user_id: this.user_id,
+	} as StreamServerUpdateEvent);
+
+	voiceState.self_stream = true;
+	await voiceState.save();
+
+	await emitEvent({
+		event: "VOICE_STATE_UPDATE",
+		data: voiceState.toPublicVoiceState(),
+		guild_id: voiceState.guild_id,
+		channel_id: voiceState.channel_id,
+	} as VoiceStateUpdateEvent);
+}
+
+//stream key:
+// guild:${guild_id}:${channel_id}:${user_id}
+// call:${channel_id}:${user_id}

src/gateway/opcodes/StreamDelete.ts

@@ -0,0 +1,76 @@
+import { parseStreamKey, Payload, WebSocket } from "@spacebar/gateway";
+import {
+	emitEvent,
+	Stream,
+	StreamDeleteEvent,
+	StreamDeleteSchema,
+	VoiceState,
+	VoiceStateUpdateEvent,
+} from "@spacebar/util";
+import { check } from "./instanceOf";
+
+export async function onStreamDelete(this: WebSocket, data: Payload) {
+	check.call(this, StreamDeleteSchema, data.d);
+	const body = data.d as StreamDeleteSchema;
+
+	let parsedKey: {
+		type: "guild" | "call";
+		channelId: string;
+		guildId?: string;
+		userId: string;
+	};
+
+	try {
+		parsedKey = parseStreamKey(body.stream_key);
+	} catch (e) {
+		return this.close(4000, "Invalid stream key");
+	}
+
+	const { userId, channelId, guildId, type } = parsedKey;
+
+	// when a user selects to stop watching another user stream, this event gets triggered
+	// just disconnect user without actually deleting stream
+	if (this.user_id !== userId) {
+		await emitEvent({
+			event: "STREAM_DELETE",
+			data: {
+				stream_key: body.stream_key,
+			},
+			user_id: this.user_id,
+		} as StreamDeleteEvent);
+		return;
+	}
+
+	const stream = await Stream.findOne({
+		where: { channel_id: channelId, owner_id: userId },
+	});
+
+	if (!stream) return;
+
+	await stream.remove();
+
+	const voiceState = await VoiceState.findOne({
+		where: { user_id: this.user_id },
+	});
+
+	if (voiceState) {
+		voiceState.self_stream = false;
+		await voiceState.save();
+
+		await emitEvent({
+			event: "VOICE_STATE_UPDATE",
+			data: voiceState.toPublicVoiceState(),
+			guild_id: guildId,
+			channel_id: channelId,
+		} as VoiceStateUpdateEvent);
+	}
+
+	await emitEvent({
+		event: "STREAM_DELETE",
+		data: {
+			stream_key: body.stream_key,
+		},
+		guild_id: guildId,
+		channel_id: channelId,
+	} as StreamDeleteEvent);
+}