feat: remove old event system

For some reason we had an event system in place already. Who knew?! I removed it and converted it to the NEW event system.
ScuffleTV · Jan 1, 2024 · 826d6ac · 826d6ac
1 parent 762d35a
commit 826d6ac
Show file tree

Hide file tree

Showing 35 changed files with 371 additions and 418 deletions.
diff --git a/platform/api/Cargo.toml b/platform/api/Cargo.toml
@@ -50,11 +50,11 @@ http-body = "1.0"
 http-body-util = "0.1"
 hyper-util = "0.1"
 pin-project = "1.1"
+base64 = "0.21"
 
 config = { workspace = true }
 pb = { workspace = true }
 binary-helper = { workspace = true }
-base64 = "0.21.5"
 
 [dev-dependencies]
 tempfile = "3.8"

diff --git a/platform/api/src/main.rs b/platform/api/src/main.rs
@@ -7,7 +7,7 @@ use binary_helper::{bootstrap, grpc_health, grpc_server, impl_global_traits};
 use common::context::Context;
 use common::dataloader::DataLoader;
 use common::global::*;
-use platform_api::config::{ApiConfig, JwtConfig, TurnstileConfig, VideoApiConfig};
+use platform_api::config::{ApiConfig, ImageUploaderConfig, JwtConfig, TurnstileConfig, VideoApiConfig};
 use platform_api::dataloader::category::CategoryByIdLoader;
 use platform_api::dataloader::global_state::GlobalStateLoader;
 use platform_api::dataloader::role::RoleByIdLoader;
@@ -184,7 +184,6 @@ impl binary_helper::Global<AppConfig> for GlobalState {
 
 		let subscription_manager = SubscriptionManager::default();
 
-
 		let image_processor_s3 = config
 			.extra
 			.image_uploader

diff --git a/platform/website/src/components/dev-banner.svelte b/platform/website/src/components/dev-banner.svelte
@@ -20,7 +20,7 @@
 		PUBLIC_BASE_URL,
 		PUBLIC_TWITTER_HANDLE,
 		PUBLIC_ORG_ID,
-		PUBLIC_EDGE_ENDPOINT
+		PUBLIC_EDGE_ENDPOINT,
 	} from "$env/static/public";
 	import { websocketOpen } from "$/store/websocket";
 	import { dev } from "$app/environment";

diff --git a/platform/website/src/components/home/big-stream-preview.svelte b/platform/website/src/components/home/big-stream-preview.svelte
@@ -38,11 +38,7 @@
 	</div>
 	<a class="video" href="/{user.username}">
 		{#if playing}
-			<Player
-			roomId={user.channel.roomId}
-				showPip={false}
-				showTheater={false}
-			/>
+			<Player roomId={user.channel.roomId} showPip={false} showTheater={false} />
 		{:else}
 			<img src={preview} alt="Stream Thumbnail" class="blurred" aria-hidden="true" />
 			<img src={preview} alt="Stream Thumbnail" class="thumbnail" />

diff --git a/proto/scuffle/video/internal/events/organization_event.proto b/proto/scuffle/video/internal/events/organization_event.proto
diff --git a/proto/scuffle/video/v1/types/event.proto b/proto/scuffle/video/v1/types/event.proto
@@ -96,25 +96,8 @@ message Event {
       // If the disconnection was clean. Meaning the client disconnected and it
       // was handled gracefully.
       bool clean = 2;
-
-      // The cause of the disconnection.
-      enum Cause {
-        // The cause is unknown.
-        DISCONNECTED_CAUSE_UNKNOWN = 0;
-        // The server requested the disconnection. This is usually because an
-        // API request on Room.Disconnect was made. Or because another client
-        // connected to the room, or the room was deleted.
-        DISCONNECTED_CAUSE_DISCONNECT_REQUEST = 1;
-        // The ingest bitrate was too high.
-        DISCONNECTED_CAUSE_HIGH_BITRATE = 2;
-        // The ingest resolution was too high.
-        DISCONNECTED_CAUSE_HIGH_RESOLUTION = 3;
-        // The ingest FPS was too high.
-        DISCONNECTED_CAUSE_HIGH_FPS = 4;
-      }
-
       // The cause of the disconnection.
-      Cause cause = 3;
+      optional string cause = 3;
     }
 
     // If the room is ready to be watched.
@@ -131,8 +114,9 @@ message Event {
 
     // If the room failed to be go live.
     message Failed {
+      scuffle.types.Ulid connection_id = 1;
       // The error that occurred.
-      string error = 1;
+      string error = 2;
     }
 
     // The event that occurred.

diff --git a/video/api/src/api/access_token/create.rs b/video/api/src/api/access_token/create.rs
@@ -107,7 +107,7 @@ impl ApiRequest<AccessTokenCreateResponse> for tonic::Request<AccessTokenCreateR
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::AccessToken,
 			event::Event::AccessToken(event::AccessToken {

diff --git a/video/api/src/api/access_token/delete.rs b/video/api/src/api/access_token/delete.rs
@@ -103,7 +103,7 @@ impl ApiRequest<AccessTokenDeleteResponse> for tonic::Request<AccessTokenDeleteR
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::AccessToken,
 				event::Event::AccessToken(event::AccessToken {

diff --git a/video/api/src/api/playback_key_pair/create.rs b/video/api/src/api/playback_key_pair/create.rs
@@ -86,7 +86,7 @@ impl ApiRequest<PlaybackKeyPairCreateResponse> for tonic::Request<PlaybackKeyPai
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::PlaybackKeyPair,
 			event::Event::PlaybackKeyPair(event::PlaybackKeyPair {

diff --git a/video/api/src/api/playback_key_pair/delete.rs b/video/api/src/api/playback_key_pair/delete.rs
@@ -63,7 +63,7 @@ impl ApiRequest<PlaybackKeyPairDeleteResponse> for tonic::Request<PlaybackKeyPai
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::PlaybackKeyPair,
 				event::Event::PlaybackKeyPair(event::PlaybackKeyPair {

diff --git a/video/api/src/api/playback_key_pair/modify.rs b/video/api/src/api/playback_key_pair/modify.rs
@@ -90,7 +90,7 @@ impl ApiRequest<PlaybackKeyPairModifyResponse> for tonic::Request<PlaybackKeyPai
 		match result {
 			Some(result) => {
 				video_common::events::emit(
-					global.jetstream(),
+					global.nats(),
 					access_token.organization_id.0,
 					Target::PlaybackKeyPair,
 					event::Event::PlaybackKeyPair(event::PlaybackKeyPair {

diff --git a/video/api/src/api/recording/modify.rs b/video/api/src/api/recording/modify.rs
@@ -114,7 +114,7 @@ impl ApiRequest<RecordingModifyResponse> for tonic::Request<RecordingModifyReque
 			.unwrap_or_default();
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::Recording,
 			event::Event::Recording(event::Recording {

diff --git a/video/api/src/api/recording_config/create.rs b/video/api/src/api/recording_config/create.rs
@@ -119,7 +119,7 @@ impl ApiRequest<RecordingConfigCreateResponse> for tonic::Request<RecordingConfi
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::RecordingConfig,
 			event::Event::RecordingConfig(event::RecordingConfig {

diff --git a/video/api/src/api/recording_config/delete.rs b/video/api/src/api/recording_config/delete.rs
@@ -102,7 +102,7 @@ impl ApiRequest<RecordingConfigDeleteResponse> for tonic::Request<RecordingConfi
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::RecordingConfig,
 				event::Event::RecordingConfig(event::RecordingConfig {

diff --git a/video/api/src/api/recording_config/modify.rs b/video/api/src/api/recording_config/modify.rs
@@ -129,7 +129,7 @@ impl ApiRequest<RecordingConfigModifyResponse> for tonic::Request<RecordingConfi
 		match result {
 			Some(result) => {
 				video_common::events::emit(
-					global.jetstream(),
+					global.nats(),
 					access_token.organization_id.0,
 					Target::RecordingConfig,
 					event::Event::RecordingConfig(event::RecordingConfig {

diff --git a/video/api/src/api/room/create.rs b/video/api/src/api/room/create.rs
@@ -123,7 +123,7 @@ impl ApiRequest<RoomCreateResponse> for tonic::Request<RoomCreateRequest> {
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::Room,
 			event::Event::Room(event::Room {

diff --git a/video/api/src/api/room/delete.rs b/video/api/src/api/room/delete.rs
@@ -95,7 +95,7 @@ impl ApiRequest<RoomDeleteResponse> for tonic::Request<RoomDeleteRequest> {
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::Room,
 				event::Event::Room(event::Room {

diff --git a/video/api/src/api/room/modify.rs b/video/api/src/api/room/modify.rs
@@ -144,7 +144,7 @@ impl ApiRequest<RoomModifyResponse> for tonic::Request<RoomModifyRequest> {
 		};
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::Room,
 			event::Event::Room(event::Room {

diff --git a/video/api/src/api/room/reset_key.rs b/video/api/src/api/room/reset_key.rs
@@ -93,7 +93,7 @@ impl ApiRequest<RoomResetKeyResponse> for tonic::Request<RoomResetKeyRequest> {
 		for RoomKeyPair { id, .. } in rooms.iter() {
 			if let Some(id) = id {
 				video_common::events::emit(
-					global.jetstream(),
+					global.nats(),
 					access_token.organization_id.0,
 					Target::Room,
 					event::Event::Room(event::Room {

diff --git a/video/api/src/api/s3_bucket/create.rs b/video/api/src/api/s3_bucket/create.rs
@@ -111,7 +111,7 @@ impl ApiRequest<S3BucketCreateResponse> for tonic::Request<S3BucketCreateRequest
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::S3Bucket,
 			event::Event::S3Bucket(event::S3Bucket {

diff --git a/video/api/src/api/s3_bucket/delete.rs b/video/api/src/api/s3_bucket/delete.rs
@@ -102,7 +102,7 @@ impl ApiRequest<S3BucketDeleteResponse> for tonic::Request<S3BucketDeleteRequest
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::S3Bucket,
 				event::Event::S3Bucket(event::S3Bucket {

diff --git a/video/api/src/api/s3_bucket/modify.rs b/video/api/src/api/s3_bucket/modify.rs
@@ -131,7 +131,7 @@ impl ApiRequest<S3BucketModifyResponse> for tonic::Request<S3BucketModifyRequest
 		match result {
 			Some(result) => {
 				video_common::events::emit(
-					global.jetstream(),
+					global.nats(),
 					access_token.organization_id.0,
 					Target::S3Bucket,
 					event::Event::S3Bucket(event::S3Bucket {

diff --git a/video/api/src/api/transcoding_config/create.rs b/video/api/src/api/transcoding_config/create.rs
@@ -88,7 +88,7 @@ impl ApiRequest<TranscodingConfigCreateResponse> for tonic::Request<TranscodingC
 			})?;
 
 		video_common::events::emit(
-			global.jetstream(),
+			global.nats(),
 			access_token.organization_id.0,
 			Target::TranscodingConfig,
 			event::Event::TranscodingConfig(event::TranscodingConfig {

diff --git a/video/api/src/api/transcoding_config/delete.rs b/video/api/src/api/transcoding_config/delete.rs
@@ -95,7 +95,7 @@ impl ApiRequest<TranscodingConfigDeleteResponse> for tonic::Request<TranscodingC
 
 		for id in deleted_ids.iter().copied() {
 			video_common::events::emit(
-				global.jetstream(),
+				global.nats(),
 				access_token.organization_id.0,
 				Target::TranscodingConfig,
 				event::Event::TranscodingConfig(event::TranscodingConfig {

diff --git a/video/api/src/api/transcoding_config/modify.rs b/video/api/src/api/transcoding_config/modify.rs
@@ -98,7 +98,7 @@ impl ApiRequest<TranscodingConfigModifyResponse> for tonic::Request<TranscodingC
 		match result {
 			Some(result) => {
 				video_common::events::emit(
-					global.jetstream(),
+					global.nats(),
 					access_token.organization_id.0,
 					Target::TranscodingConfig,
 					event::Event::TranscodingConfig(event::TranscodingConfig {

diff --git a/video/api/src/api/utils/tags.rs b/video/api/src/api/utils/tags.rs
@@ -179,7 +179,7 @@ macro_rules! impl_tag_req {
 				})?;
 
 				let $id = pb::ext::UlidExt::into_ulid(req.id);
-				video_common::events::emit(global.jetstream(), access_token.organization_id.0, $event_target, $event).await;
+				video_common::events::emit(global.nats(), access_token.organization_id.0, $event_target, $event).await;
 
 				Ok(tonic::Response::new($resp {
 					tags: Some(result.into_tags()?)
@@ -218,7 +218,7 @@ macro_rules! impl_untag_req {
 				})?;
 
 				let $id = pb::ext::UlidExt::into_ulid(req.id);
-				video_common::events::emit(global.jetstream(), access_token.organization_id.0, $event_target, $event).await;
+				video_common::events::emit(global.nats(), access_token.organization_id.0, $event_target, $event).await;
 
 				Ok(tonic::Response::new($resp {
 					tags: Some(result.into_tags()?)

diff --git a/video/cli/src/cli/events/fetch.rs b/video/cli/src/cli/events/fetch.rs
@@ -287,7 +287,7 @@ impl Invokable for Fetch {
 								action: "disconnected".to_owned(),
 								connection_id: Some(disconnected.connection_id.into_ulid()),
 								clean: Some(disconnected.clean),
-								cause: Some(disconnected.cause().as_str_name().into()),
+								cause: disconnected.cause,
 								..Default::default()
 							},
 							Some(event::room::Event::Ready(ready)) => EventPayload {

diff --git a/video/common/src/events.rs b/video/common/src/events.rs
@@ -4,21 +4,20 @@ use prost::Message;
 
 use crate::keys::event_subject;
 
-pub async fn emit(jetstream: &async_nats::jetstream::Context, org_id: ulid::Ulid, target: Target, event: event::Event) {
-	jetstream
-		.publish(
-			event_subject(org_id, target),
-			Event {
-				timestamp: chrono::Utc::now().timestamp_millis(),
-				event_id: Some(ulid::Ulid::new().into()),
-				event: Some(event),
-			}
-			.encode_to_vec()
-			.into(),
-		)
-		.await
-		.map_err(|e| {
-			tracing::error!(err = %e, "failed to publish event");
-		})
-		.ok();
+pub async fn emit(nats: &async_nats::Client, org_id: ulid::Ulid, target: Target, event: event::Event) {
+	nats.publish(
+		event_subject(org_id, target),
+		Event {
+			timestamp: chrono::Utc::now().timestamp_millis(),
+			event_id: Some(ulid::Ulid::new().into()),
+			event: Some(event),
+		}
+		.encode_to_vec()
+		.into(),
+	)
+	.await
+	.map_err(|e| {
+		tracing::error!(err = %e, "failed to publish event");
+	})
+	.ok();
 }
diff --git a/video/ingest/src/config.rs b/video/ingest/src/config.rs
@@ -28,9 +28,6 @@ pub struct IngestConfig {
 	/// NATS subject to send transcoder requests to
 	pub transcoder_request_subject: String,
 
-	/// NATS subject for events
-	pub events_subject: String,
-
 	/// The interval in to update the bitrate for a room
 	pub bitrate_update_interval: Duration,
 
@@ -58,7 +55,6 @@ impl Default for IngestConfig {
 	fn default() -> Self {
 		Self {
 			transcoder_request_subject: "transcoder-request".to_string(),
-			events_subject: "events".to_string(),
 			bitrate_update_interval: Duration::from_secs(5),
 			max_bitrate: 12000 * 1024,
 			max_bytes_between_keyframes: 5 * 12000 * 1024 / 8,