Save streams representation to the database (#59)

* Add stream DB module + migration

* Add processing to other routes for twitch callback

* Fix rollback

* Add user FK to stream

* Move ChatMessagePayload to new vendors module

* Move NATS to its own module

* Move events to their own module

* Move queue to its own module

* Delete workers module

* Add event stream support for stream.online|offline

* Add support for creating stream in db on stream.online event

* Check for stream online event when saving new stream

justfile

@@ -39,7 +39,7 @@ create-db:
 drop-db:
     sqlx database drop
 
-mig_source := "crates/common/migrations"
+mig_source := "./migrations"
 
 mig: migrate
 migrate:
@@ -49,7 +49,7 @@ mig-add mig_name:
     sqlx migrate add {{ mig_name }} --source {{ mig_source }}
 
 revert:
-    sqlx migrate run --source {{ mig_source }}
+    sqlx migrate revert --source {{ mig_source }}
 
 # Utility commands
 sync: sync-web

migrations/20250224032704_add_stream.down.sql

@@ -0,0 +1,3 @@
+DROP TRIGGER IF EXISTS update_streams_updated_at ON streams;
+
+DROP TABLE IF EXISTS streams;

migrations/20250224032704_add_stream.up.sql

@@ -0,0 +1,27 @@
+CREATE TABLE IF NOT EXISTS streams (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid (),
+    user_id UUID NOT NULL REFERENCES users (id) ON DELETE CASCADE,
+    start_time TIMESTAMP
+    WITH
+        TIME ZONE NOT NULL,
+        end_time TIMESTAMP
+    WITH
+        TIME ZONE,
+        event_log_url TEXT,
+        video_url TEXT,
+        created_at TIMESTAMP
+    WITH
+        TIME ZONE NOT NULL DEFAULT NOW (),
+        updated_at TIMESTAMP
+    WITH
+        TIME ZONE NOT NULL DEFAULT NOW ()
+);
+
+-- Add an index on start_time for efficient querying
+CREATE INDEX idx_streams_start_time ON streams (start_time);
+
+CREATE INDEX idx_streams_user_id ON streams (user_id);
+
+-- Create trigger using existing function
+CREATE TRIGGER update_streams_updated_at BEFORE
+UPDATE ON streams FOR EACH ROW EXECUTE FUNCTION update_updated_at_column ();

src/api/app_state.rs

@@ -1,9 +1,7 @@
 use super::config::Config;
 use crate::{
-    db::connect_to_database,
+    db::connect_to_database, event::Stream, nats::create_nats_client, queue::Queue,
     storage::s3::create_s3_client,
-    workers::events::Stream,
-    workers::{create_nats_client, Queue},
 };
 use sqlx::PgPool;
 

src/api/twitch/eventsub/callback.rs

@@ -4,46 +4,20 @@ use axum::{
     response::IntoResponse,
 };
 use bytes::Bytes;
-use chrono::{DateTime, Utc};
 use hmac::{Hmac, Mac};
-use serde::{Deserialize, Serialize};
 use sha2::Sha256;
 use std::sync::Arc;
 
 use crate::{
     api::{app_state::AppState, routes::auth::oauth::twitch::TwitchCredentials},
-    workers::events::{chat::ChatMessagePayload, Event},
+    db::streams::Stream,
+    event::Event,
+    twitch::{subscription::Notification, ChatMessagePayload, StreamStatusPayload},
 };
 
 type HmacSha256 = Hmac<Sha256>;
 const HMAC_PREFIX: &str = "sha256=";
 
-#[derive(Debug, Deserialize, Serialize)]
-struct Subscription {
-    id: String,
-    status: String,
-    #[serde(rename = "type")]
-    event_type: String,
-    version: String,
-    cost: i32,
-    condition: serde_json::Value,
-    transport: Transport,
-    created_at: DateTime<Utc>,
-}
-
-#[derive(Debug, Deserialize, Serialize)]
-struct Transport {
-    method: String,
-    callback: String,
-}
-
-#[derive(Debug, Deserialize, Serialize)]
-struct Notification {
-    subscription: Subscription,
-    event: Option<serde_json::Value>,
-    challenge: Option<String>,
-}
-
 pub async fn handle_webhook(
     State(state): State<Arc<AppState>>,
     headers: HeaderMap,
@@ -100,6 +74,74 @@ pub async fn handle_webhook(
             tracing::debug!("Event type: {}", notification.subscription.event_type);
             let notification_type = notification.subscription.event_type;
             match notification_type.as_str() {
+                "stream.online" | "stream.offline" => {
+                    let Some(raw_payload) = notification.event else {
+                        tracing::error!("Received stream status notification without event");
+                        return (StatusCode::BAD_REQUEST, "Missing event data").into_response();
+                    };
+
+                    let stream_payload =
+                        serde_json::from_value::<StreamStatusPayload>(raw_payload.clone());
+                    let Ok(stream_payload) = stream_payload else {
+                        tracing::error!("Failed to parse stream status notification");
+                        return (StatusCode::BAD_REQUEST, "Invalid event data").into_response();
+                    };
+
+                    // If the stream is online, then we also want to start a new stream in the database for that user
+                    if stream_payload.is_online() {
+                        // Start by getting the user account by the payload
+                        let Ok(user_account) =
+                            stream_payload.find_broadcaster_account(&state.db).await
+                        else {
+                            tracing::error!("Failed to find broadcaster account");
+                            return (
+                                StatusCode::BAD_REQUEST,
+                                "Failed to find broadcaster account",
+                            )
+                                .into_response();
+                        };
+                        // Parse the start time from the payload
+                        let Some(start_time) = stream_payload.event.started_at else {
+                            tracing::error!("Failed to find stream start time");
+                            return (StatusCode::BAD_REQUEST, "Failed to find stream start time")
+                                .into_response();
+                        };
+                        // Save the stream to the database
+                        let Ok(_user_stream) =
+                            Stream::create(user_account.user_id, start_time, &state.db).await
+                        else {
+                            tracing::error!("Failed to create stream");
+                            return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to create stream")
+                                .into_response();
+                        };
+                    }
+
+                    // Lastly, publish the stream status event
+                    let subject = Event::from(stream_payload).get_subject();
+                    state
+                        .event_stream
+                        .publish(subject.to_string(), raw_payload.to_string()) // Pass the original payload so we can skip serialization
+                        .await
+                        .map_err(|e| {
+                            tracing::error!("Failed to publish stream status event: {}", e);
+                            StatusCode::INTERNAL_SERVER_ERROR
+                        })
+                        .expect("Failed to publish stream status event");
+                }
+                "channel.follow" => {
+                    return (
+                        StatusCode::NOT_IMPLEMENTED,
+                        "Channel follow not implemented",
+                    )
+                        .into_response()
+                }
+                "channel.subscribe" => {
+                    return (
+                        StatusCode::NOT_IMPLEMENTED,
+                        "Channel subscription not implemented",
+                    )
+                        .into_response()
+                }
                 "channel.chat.message" => {
                     tracing::debug!("Channel chat message received");
                     // Pull the raw payload out of the notification

src/bin/down.rs

@@ -1,8 +1,5 @@
 use anyhow::Result;
-use farmhand::{
-    db,
-    workers::{create_nats_client, Queue, Stream},
-};
+use farmhand::{db, event::Stream, nats::create_nats_client, queue::Queue};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 #[tokio::main]

src/bin/job_runner.rs

@@ -1,6 +1,10 @@
 use anyhow::Result;
 use async_nats::jetstream::AckKind;
-use farmhand::workers::{self, events::MESSAGE_PREFIX, queue::process_message};
+use farmhand::{
+    event::MESSAGE_PREFIX,
+    nats::create_nats_client,
+    queue::{process_message, Queue},
+};
 use futures::StreamExt;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -14,9 +18,9 @@ async fn main() -> Result<()> {
         .init();
     // Connect to the stream
     tracing::debug!("Connecting to NATS server");
-    let nats_client = workers::create_nats_client().await?;
+    let nats_client = create_nats_client().await?;
     tracing::debug!("Connecting to queue");
-    let queue = workers::Queue::connect(nats_client)
+    let queue = Queue::connect(nats_client)
         .await
         .expect("Failed to create worker queue");
 

src/bin/stream_listener.rs

@@ -2,7 +2,10 @@
 //! It intentionally does not do anything else
 
 use anyhow::Result;
-use farmhand::workers::{self, events::MESSAGE_PREFIX};
+use farmhand::{
+    event::{Stream, MESSAGE_PREFIX},
+    nats::create_nats_client,
+};
 use futures::StreamExt;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -17,9 +20,9 @@ async fn main() -> Result<()> {
         .init();
     // Create the NATS client
     tracing::debug!("Connecting to NATS server");
-    let nats_client = workers::create_nats_client().await?;
+    let nats_client = create_nats_client().await?;
     // Setup the Jetstream queue
-    let listener = workers::Stream::connect(nats_client)
+    let listener = Stream::connect(nats_client)
         .await
         .expect("Failed to create worker queue");
 

src/bin/up.rs

@@ -1,10 +1,9 @@
 use anyhow::Result;
 use farmhand::{
     db,
-    workers::{
-        self,
-        events::{EVENT_PREFIX, EVENT_STREAM, JOB_PREFIX, JOB_STREAM, MESSAGE_PREFIX},
-    },
+    event::{Stream, EVENT_PREFIX, EVENT_STREAM, JOB_PREFIX, JOB_STREAM, MESSAGE_PREFIX},
+    nats::create_nats_client,
+    queue::Queue,
 };
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -49,13 +48,13 @@ async fn init_project_nats() {
 
     // Connect to the NATS server
     tracing::debug!("Connecting to NATS server");
-    let nats_client = workers::create_nats_client()
+    let nats_client = create_nats_client()
         .await
         .expect("Failed to connect to NATS");
 
     // Create the event stream
     let all_events_subject = format!("{}.{}.>", MESSAGE_PREFIX, EVENT_PREFIX);
-    workers::Stream::new(
+    Stream::new(
         EVENT_STREAM.to_string(),
         Some("All Farmhand events".to_string()),
         vec![all_events_subject],
@@ -66,7 +65,7 @@ async fn init_project_nats() {
 
     // Create the job stream
     let all_jobs_subject = format!("{}.{}.>", MESSAGE_PREFIX, JOB_PREFIX);
-    workers::Queue::new(
+    Queue::new(
         JOB_STREAM.to_string(),
         Some("All Farmhand jobs".to_string()),
         vec![all_jobs_subject],

src/db/mod.rs

@@ -1,4 +1,5 @@
 pub mod accounts;
+pub mod streams;
 pub mod users;
 pub mod videos;