add schemas for stream gateway messages

src/gateway/opcodes/StreamCreate.ts

@@ -11,17 +11,13 @@ import {
 	Region,
 	Snowflake,
 	Stream,
+	StreamCreateSchema,
 	StreamSession,
 } from "@spacebar/util";
-
-interface StreamCreateSchema {
-	type: "guild" | "call";
-	channel_id: string;
-	guild_id?: string;
-	preferred_region?: string;
-}
+import { check } from "./instanceOf";
 
 export async function onStreamCreate(this: WebSocket, data: Payload) {
+	check.call(this, StreamCreateSchema, data.d);
 	const body = data.d as StreamCreateSchema;
 
 	// TODO: first check if we are in a voice channel already. cannot create a stream if there's no existing voice connection

src/gateway/opcodes/StreamDelete.ts

@@ -1,16 +1,25 @@
 import { parseStreamKey, Payload, WebSocket } from "@spacebar/gateway";
-import { emitEvent, Stream } from "@spacebar/util";
-
-interface StreamDeleteSchema {
-	stream_key: string;
-}
+import { emitEvent, Stream, StreamDeleteSchema } 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;
 
-	const { userId, channelId, guildId, type } = parseStreamKey(
-		body.stream_key,
-	);
+	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;
 
 	if (this.user_id !== userId) {
 		return this.close(4000, "Cannot delete stream for another user");

src/gateway/opcodes/StreamWatch.ts

@@ -4,13 +4,18 @@ import {
 	Payload,
 	WebSocket,
 } from "@spacebar/gateway";
-import { Config, emitEvent, Stream, StreamSession } from "@spacebar/util";
-
-interface StreamWatchSchema {
-	stream_key: string;
-}
+import {
+	Config,
+	emitEvent,
+	Stream,
+	StreamSession,
+	StreamWatchSchema,
+} from "@spacebar/util";
+import { check } from "./instanceOf";
+import { Not } from "typeorm";
 
 export async function onStreamWatch(this: WebSocket, data: Payload) {
+	check.call(this, StreamWatchSchema, data.d);
 	const body = data.d as StreamWatchSchema;
 
 	// TODO: apply perms: check if user is allowed to watch
@@ -56,8 +61,13 @@ export async function onStreamWatch(this: WebSocket, data: Payload) {
 
 	await streamSession.save();
 
+	// get the viewers: stream session tokens for this stream that have been used but not including stream owner
 	const viewers = await StreamSession.find({
-		where: { stream_id: stream.id },
+		where: {
+			stream_id: stream.id,
+			used: true,
+			user_id: Not(stream.owner_id),
+		},
 	});
 
 	await emitEvent({

src/util/schemas/StreamCreateSchema.ts

@@ -0,0 +1,13 @@
+export interface StreamCreateSchema {
+	type: "guild" | "call";
+	channel_id: string;
+	guild_id?: string;
+	preferred_region?: string;
+}
+
+export const StreamCreateSchema = {
+	type: String,
+	channel_id: String,
+	$guild_id: String,
+	$preferred_region: String,
+};

src/util/schemas/StreamDeleteSchema.ts

@@ -0,0 +1,7 @@
+export interface StreamDeleteSchema {
+	stream_key: string;
+}
+
+export const StreamDeleteSchema = {
+	stream_key: String,
+};

src/util/schemas/StreamWatchSchema.ts

@@ -0,0 +1,7 @@
+export interface StreamWatchSchema {
+	stream_key: string;
+}
+
+export const StreamWatchSchema = {
+	stream_key: String,
+};

src/util/schemas/index.ts

@@ -68,6 +68,9 @@ export * from "./responses";
 export * from "./RoleModifySchema";
 export * from "./RolePositionUpdateSchema";
 export * from "./SelectProtocolSchema";
+export * from "./StreamCreateSchema";
+export * from "./StreamDeleteSchema";
+export * from "./StreamWatchSchema";
 export * from "./TeamCreateSchema";
 export * from "./TemplateCreateSchema";
 export * from "./TemplateModifySchema";