From 54f668508277237a0dcfc920266d4875b61d75f2 Mon Sep 17 00:00:00 2001
From: dank074 <torresefrain10@gmail.com>
Date: Thu, 17 Apr 2025 16:18:46 -0500
Subject: [PATCH] add schemas for stream gateway messages

---
 src/gateway/opcodes/StreamCreate.ts    | 10 +++-------
 src/gateway/opcodes/StreamDelete.ts    | 25 +++++++++++++++++--------
 src/gateway/opcodes/StreamWatch.ts     | 22 ++++++++++++++++------
 src/util/schemas/StreamCreateSchema.ts | 13 +++++++++++++
 src/util/schemas/StreamDeleteSchema.ts |  7 +++++++
 src/util/schemas/StreamWatchSchema.ts  |  7 +++++++
 src/util/schemas/index.ts              |  3 +++
 7 files changed, 66 insertions(+), 21 deletions(-)
 create mode 100644 src/util/schemas/StreamCreateSchema.ts
 create mode 100644 src/util/schemas/StreamDeleteSchema.ts
 create mode 100644 src/util/schemas/StreamWatchSchema.ts

diff --git a/src/gateway/opcodes/StreamCreate.ts b/src/gateway/opcodes/StreamCreate.ts
index 82c2c243..608b6792 100644
--- a/src/gateway/opcodes/StreamCreate.ts
+++ b/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
diff --git a/src/gateway/opcodes/StreamDelete.ts b/src/gateway/opcodes/StreamDelete.ts
index e33204ad..c6f006ef 100644
--- a/src/gateway/opcodes/StreamDelete.ts
+++ b/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");
diff --git a/src/gateway/opcodes/StreamWatch.ts b/src/gateway/opcodes/StreamWatch.ts
index 2b42f46a..119d4cad 100644
--- a/src/gateway/opcodes/StreamWatch.ts
+++ b/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({
diff --git a/src/util/schemas/StreamCreateSchema.ts b/src/util/schemas/StreamCreateSchema.ts
new file mode 100644
index 00000000..bb650791
--- /dev/null
+++ b/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,
+};
diff --git a/src/util/schemas/StreamDeleteSchema.ts b/src/util/schemas/StreamDeleteSchema.ts
new file mode 100644
index 00000000..0e2aff75
--- /dev/null
+++ b/src/util/schemas/StreamDeleteSchema.ts
@@ -0,0 +1,7 @@
+export interface StreamDeleteSchema {
+	stream_key: string;
+}
+
+export const StreamDeleteSchema = {
+	stream_key: String,
+};
diff --git a/src/util/schemas/StreamWatchSchema.ts b/src/util/schemas/StreamWatchSchema.ts
new file mode 100644
index 00000000..263bb11f
--- /dev/null
+++ b/src/util/schemas/StreamWatchSchema.ts
@@ -0,0 +1,7 @@
+export interface StreamWatchSchema {
+	stream_key: string;
+}
+
+export const StreamWatchSchema = {
+	stream_key: String,
+};
diff --git a/src/util/schemas/index.ts b/src/util/schemas/index.ts
index 9701faec..f19eef0d 100644
--- a/src/util/schemas/index.ts
+++ b/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";