feat(youtube): deserialize youtube event

workers/twitter/Cargo.toml

@@ -16,7 +16,7 @@ parking_lot = "0.12"
 reqwest = "0.11"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-sg-core = { package = "core", path = "../../core", features=["mq"] }
+sg-core = { package = "core", path = "../../core", features = ["mq"] }
 humantime-serde = "1.0"
 tap = "1.0"
 tarpc = { version = "0.27", features = ["serde1", "tokio1"] }

workers/youtube/Cargo.toml

@@ -2,30 +2,30 @@
 name = "youtube"
 version = "0.1.0"
 edition = "2021"
-
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 axum = "0.4"
 axum-xml = "0.1"
-serde = { version = "1.0", features = ["derive"] }
-reqwest = "0.11"
-sg-core = { package = "core", path = "../../core", features = ["mq"] }
-figment = { version = "0.10", features = ["env", "test"] }
-uuid = "0.8"
-tarpc = { version = "0.27", features = ["serde1", "tokio1"] }
-tokio = { version = "1.17", features = ["rt", "rt-multi-thread", "parking_lot", "time", "net", "macros"] }
-eyre = "0.6"
 color-eyre = "0.6"
-tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+eyre = "0.6"
+figment = { version = "0.10", features = ["env", "test"] }
+futures = "0.3"
 humantime-serde = "1.1"
 once_cell = "1.10"
+parking_lot = "0.12"
+quick-xml = "0.22"
+reqwest = "0.11"
 reqwest-middleware = "0.1"
 reqwest-retry = "0.1"
-parking_lot = "0.12"
+serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-futures = "0.3"
+serde_urlencoded = "0.7"
+sg-core = { package = "core", path = "../../core", features = ["mq"] }
 tap = "1.0"
-url = "2.2"
-serde_urlencoded = "0.7"
+tarpc = { version = "0.27", features = ["serde1", "tokio1"] }
+tokio = { version = "1.17", features = ["rt", "rt-multi-thread", "parking_lot", "time", "net", "macros"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+url = { version = "2.2", features = ["serde"] }
+uuid = "0.8"

workers/youtube/src/models.rs

@@ -1,7 +1,7 @@
 use serde::{Deserialize, Serialize};
 use url::Url;
 
-#[derive(Deserialize)]
+#[derive(Debug, Deserialize)]
 pub struct ChallengeQuery {
     #[serde(rename = "hub.topic")]
     pub topic: Url,
@@ -38,3 +38,30 @@ pub struct SubscribeForm {
     #[serde(rename = "hub.lease_seconds")]
     pub lease_seconds: u64,
 }
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct Feed {
+    pub entry: Option<Entry>,
+    pub deleted_entry: Option<DeletedEntry>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Entry {
+    pub video_id: String,
+    pub link: Link,
+    pub title: String,
+    pub channel_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Link {
+    pub href: Url,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DeletedEntry {
+    #[serde(rename = "ref")]
+    pub video_id: String,
+}

workers/youtube/src/server.rs

@@ -6,14 +6,15 @@ use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::routing::get;
 use axum::{Router, Server};
+use axum_xml::Xml;
 use eyre::Result;
 use parking_lot::RwLock;
 use serde::Deserialize;
 use tracing::{error, info, warn};
 
 use sg_core::mq::MessageQueue;
 
-use crate::models::{ChallengeQuery, Mode};
+use crate::models::{ChallengeQuery, Feed, Mode};
 use crate::registry::Registry;
 use crate::Config;
 
@@ -33,7 +34,7 @@ impl IntoResponse for ChallengeError {
 
 #[derive(Deserialize)]
 struct TopicQuery {
-    topic: String,
+    channel_id: String,
 }
 
 #[allow(clippy::unused_async)]
@@ -42,7 +43,8 @@ async fn challenge(
     Extension(registry): Extension<Arc<RwLock<Registry>>>,
 ) -> Result<String, ChallengeError> {
     let channel_id =
-        serde_urlencoded::from_str::<TopicQuery>(query.topic.query().unwrap_or_default())?.topic;
+        serde_urlencoded::from_str::<TopicQuery>(query.topic.query().unwrap_or_default())?
+            .channel_id;
     let has_task = registry.read().contains_channel(&channel_id);
     let mode = query.mode;
     if (mode == Mode::Subscribe && has_task) || (mode == Mode::Unsubscribe && !has_task) {
@@ -70,8 +72,13 @@ impl IntoResponse for EventError {
 }
 
 #[allow(clippy::unused_async)]
-async fn event(Extension(_registry): Extension<Arc<RwLock<Registry>>>) -> Result<(), EventError> {
-    todo!()
+async fn event(
+    Extension(_registry): Extension<Arc<RwLock<Registry>>>,
+    feed: Xml<Feed>,
+) -> Result<(), EventError> {
+    eprintln!("Event received.");
+    eprintln!("{:?}", feed);
+    Ok(())
 }
 
 pub async fn serve(