+10 -8

Cargo.lock

··· 1509 1509 1510 1510 [[package]] 1511 1511 name = "deranged" 1512 1512 - version = "0.4.0" 1512 1512 + version = "0.5.3" 1513 1513 source = "registry+https://github.com/rust-lang/crates.io-index" 1514 1514 - checksum = "9c9e6a11ca8224451684bc0d7d5a7adbf8f2fd6887261a1cfc3c0432f9d4068e" 1514 1514 + checksum = "d630bccd429a5bb5a64b5e94f693bfc48c9f8566418fda4c494cc94f911f87cc" 1515 1515 dependencies = [ 1516 1516 "powerfmt", 1517 1517 "serde", ··· 4931 4931 "dotenv", 4932 4932 "futures-util", 4933 4933 "owo-colors", 4934 4934 + "redis 0.29.5", 4934 4935 "reqwest", 4935 4936 "serde", 4936 4937 "serde_json", 4937 4938 "sha256", 4938 4939 "sqlx", 4940 4940 + "time", 4939 4941 "tokio", 4940 4942 "tokio-stream", 4941 4943 "tokio-tungstenite", ··· 6440 6442 6441 6443 [[package]] 6442 6444 name = "time" 6443 6443 - version = "0.3.41" 6445 6445 + version = "0.3.44" 6444 6446 source = "registry+https://github.com/rust-lang/crates.io-index" 6445 6445 - checksum = "8a7619e19bc266e0f9c5e6686659d394bc57973859340060a69221e57dbc0c40" 6447 6447 + checksum = "91e7d9e3bb61134e77bde20dd4825b97c010155709965fedf0f49bb138e52a9d" 6446 6448 dependencies = [ 6447 6449 "deranged", 6448 6450 "itoa", ··· 6455 6457 6456 6458 [[package]] 6457 6459 name = "time-core" 6458 6458 - version = "0.1.4" 6460 6460 + version = "0.1.6" 6459 6461 source = "registry+https://github.com/rust-lang/crates.io-index" 6460 6460 - checksum = "c9e9a38711f559d9e3ce1cdb06dd7c5b8ea546bc90052da6d06bb76da74bb07c" 6462 6462 + checksum = "40868e7c1d2f0b8d73e4a8c7f0ff63af4f6d19be117e90bd73eb1d62cf831c6b" 6461 6463 6462 6464 [[package]] 6463 6465 name = "time-macros" 6464 6464 - version = "0.2.22" 6466 6466 + version = "0.2.24" 6465 6467 source = "registry+https://github.com/rust-lang/crates.io-index" 6466 6466 - checksum = "3526739392ec93fd8b359c8e98514cb3e8e021beb4e5f597b00a0221f8ed8a49" 6468 6468 + checksum = "30cfb0125f12d9c277f35663a0a33f8c30190f4e4574868a330595412d34ebf3" 6467 6469 dependencies = [ 6468 6470 "num-conv", 6469 6471 "time-core",

+2

crates/jetstream/Cargo.toml

··· 36 36 "json", 37 37 ], default-features = false } 38 38 sha256 = "1.6.0" 39 39 + time = { version = "0.3.44", features = ["formatting", "macros"] } 40 40 + redis = { version = "0.29.0", features = ["aio", "tokio-comp"] }

+16 -3

crates/jetstream/src/lib.rs

··· 1 1 use anyhow::Error; 2 2 - use std::env; 3 3 - 2 2 + use std::{env, sync::Arc}; 4 3 use subscriber::ScrobbleSubscriber; 4 4 + use tokio::sync::Mutex; 5 5 + 6 6 + use crate::webhook_worker::{start_worker, AppState}; 5 7 6 8 pub mod profile; 7 9 pub mod repo; 8 10 pub mod subscriber; 9 11 pub mod types; 12 12 + pub mod webhook; 13 13 + pub mod webhook_worker; 10 14 pub mod xata; 11 15 12 16 pub async fn subscribe() -> Result<(), Error> { 17 17 + let redis_url = env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); 18 18 + let redis = redis::Client::open(redis_url)?; 19 19 + let queue_key = 20 20 + env::var("WEBHOOK_QUEUE_KEY").unwrap_or_else(|_| "rocksky:webhook_queue".to_string()); 21 21 + 22 22 + let state = Arc::new(Mutex::new(AppState { redis, queue_key })); 23 23 + 24 24 + start_worker(state.clone()).await?; 25 25 + 13 26 let jetstream_server = env::var("JETSTREAM_SERVER") 14 27 .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network".to_string()); 15 28 let url = format!( ··· 18 31 ); 19 32 let subscriber = ScrobbleSubscriber::new(&url); 20 33 21 21 - subscriber.run().await?; 34 34 + subscriber.run(state).await?; 22 35 23 36 Ok(()) 24 37 }

+14 -2

crates/jetstream/src/main.rs

··· 1 1 - use std::env; 1 1 + use std::{env, sync::Arc}; 2 2 3 3 use dotenv::dotenv; 4 4 use subscriber::ScrobbleSubscriber; 5 5 + use tokio::sync::Mutex; 6 6 + 7 7 + use crate::webhook_worker::AppState; 5 8 6 9 pub mod profile; 7 10 pub mod repo; 8 11 pub mod subscriber; 9 12 pub mod types; 13 13 + pub mod webhook; 14 14 + pub mod webhook_worker; 10 15 pub mod xata; 11 16 12 17 #[tokio::main] ··· 20 25 ); 21 26 let subscriber = ScrobbleSubscriber::new(&url); 22 27 23 23 - subscriber.run().await?; 28 28 + let redis_url = env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); 29 29 + let redis = redis::Client::open(redis_url)?; 30 30 + let queue_key = 31 31 + env::var("WEBHOOK_QUEUE_KEY").unwrap_or_else(|_| "rocksky:webhook_queue".to_string()); 32 32 + 33 33 + let state = Arc::new(Mutex::new(AppState { redis, queue_key })); 34 34 + 35 35 + subscriber.run(state).await?; 24 36 Ok(()) 25 37 }

+61

crates/jetstream/src/repo.rs

··· 10 10 profile::did_to_profile, 11 11 subscriber::{ALBUM_NSID, ARTIST_NSID, SCROBBLE_NSID, SONG_NSID}, 12 12 types::{AlbumRecord, ArtistRecord, Commit, ScrobbleRecord, SongRecord}, 13 13 + webhook::discord::{ 14 14 + self, 15 15 + model::{ScrobbleData, WebhookEnvelope}, 16 16 + }, 17 17 + webhook_worker::{push_to_queue, AppState}, 13 18 xata::{ 14 19 album::Album, album_track::AlbumTrack, artist::Artist, artist_album::ArtistAlbum, 15 20 artist_track::ArtistTrack, track::Track, user::User, user_album::UserAlbum, ··· 18 23 }; 19 24 20 25 pub async fn save_scrobble( 26 26 + state: Arc<Mutex<AppState>>, 21 27 pool: Arc<Mutex<Pool<Postgres>>>, 22 28 did: &str, 23 29 commit: Commit, ··· 85 91 .await?; 86 92 87 93 tx.commit().await?; 94 94 + 95 95 + let users: Vec<User> = 96 96 + sqlx::query_as::<_, User>("SELECT * FROM users WHERE did = $1") 97 97 + .bind(did) 98 98 + .fetch_all(&*pool) 99 99 + .await?; 100 100 + 101 101 + if users.is_empty() { 102 102 + return Err(anyhow::anyhow!( 103 103 + "User with DID {} not found in database", 104 104 + did 105 105 + )); 106 106 + } 107 107 + 108 108 + // Push to webhook queue (Discord) 109 109 + match push_to_queue( 110 110 + state, 111 111 + &WebhookEnvelope { 112 112 + r#type: "scrobble.created".to_string(), 113 113 + id: Some(commit.rkey.clone()), 114 114 + data: ScrobbleData { 115 115 + user: discord::model::User { 116 116 + did: did.to_string(), 117 117 + display_name: users[0].display_name.clone(), 118 118 + handle: users[0].handle.clone(), 119 119 + }, 120 120 + track: discord::model::Track { 121 121 + title: scrobble_record.title.clone(), 122 122 + artist: scrobble_record.artist.clone(), 123 123 + album: scrobble_record.album.clone(), 124 124 + duration: scrobble_record.duration, 125 125 + artwork_url: scrobble_record.album_art.clone().map(|x| { 126 126 + format!( 127 127 + "https://cdn.bsky.app/img/feed_thumbnail/plain/{}/{}@{}", 128 128 + did, 129 129 + x.r#ref.link, 130 130 + x.mime_type.split('/').last().unwrap_or("jpeg") 131 131 + ) 132 132 + }), 133 133 + spotify_url: scrobble_record.spotify_link.clone(), 134 134 + tidal_url: scrobble_record.tidal_link.clone(), 135 135 + youtube_url: scrobble_record.youtube_link.clone(), 136 136 + }, 137 137 + played_at: scrobble_record.created_at.clone(), 138 138 + }, 139 139 + delivered_at: Some(chrono::Utc::now().to_rfc3339()), 140 140 + }, 141 141 + ) 142 142 + .await 143 143 + { 144 144 + Ok(_) => {} 145 145 + Err(e) => { 146 146 + eprintln!("Failed to push to webhook queue: {}", e); 147 147 + } 148 148 + } 88 149 } 89 150 90 151 if commit.collection == ARTIST_NSID {

+9 -5

crates/jetstream/src/subscriber.rs

··· 7 7 use tokio::sync::Mutex; 8 8 use tokio_tungstenite::{connect_async, tungstenite::Message}; 9 9 10 10 - use crate::{repo::save_scrobble, types::Root}; 10 10 + use crate::{repo::save_scrobble, types::Root, webhook_worker::AppState}; 11 11 12 12 pub const SCROBBLE_NSID: &str = "app.rocksky.scrobble"; 13 13 pub const ARTIST_NSID: &str = "app.rocksky.artist"; ··· 28 28 } 29 29 } 30 30 31 31 - pub async fn run(&self) -> Result<(), Error> { 31 31 + pub async fn run(&self, state: Arc<Mutex<AppState>>) -> Result<(), Error> { 32 32 // Get the connection string outside of the task 33 33 let db_url = env::var("XATA_POSTGRES_URL") 34 34 .context("Failed to get XATA_POSTGRES_URL environment variable")?; ··· 48 48 while let Some(msg) = ws_stream.next().await { 49 49 match msg { 50 50 Ok(msg) => { 51 51 - if let Err(e) = handle_message(pool.clone(), msg).await { 51 51 + if let Err(e) = handle_message(state.clone(), pool.clone(), msg).await { 52 52 eprintln!("Error handling message: {}", e); 53 53 } 54 54 } ··· 63 63 } 64 64 } 65 65 66 66 - async fn handle_message(pool: Arc<Mutex<sqlx::PgPool>>, msg: Message) -> Result<(), Error> { 66 66 + async fn handle_message( 67 67 + state: Arc<Mutex<AppState>>, 68 68 + pool: Arc<Mutex<sqlx::PgPool>>, 69 69 + msg: Message, 70 70 + ) -> Result<(), Error> { 67 71 tokio::spawn(async move { 68 72 if let Message::Text(text) = msg { 69 73 let message: Root = serde_json::from_str(&text)?; ··· 74 78 75 79 println!("Received message: {:#?}", message); 76 80 if let Some(commit) = message.commit { 77 77 - match save_scrobble(pool, &message.did, commit).await { 81 81 + match save_scrobble(state, pool, &message.did, commit).await { 78 82 Ok(_) => { 79 83 println!("Scrobble saved successfully"); 80 84 }

+54

crates/jetstream/src/webhook/discord/mod.rs

··· 1 1 + pub mod model; 2 2 + 3 3 + use crate::webhook::discord::model::*; 4 4 + use reqwest::Client; 5 5 + 6 6 + pub fn embed_from_scrobble(s: &ScrobbleData) -> DiscordEmbed { 7 7 + let url = s 8 8 + .track 9 9 + .spotify_url 10 10 + .as_ref() 11 11 + .or(s.track.tidal_url.as_ref()) 12 12 + .or(s.track.youtube_url.as_ref()) 13 13 + .cloned(); 14 14 + 15 15 + let mut desc = format!("**{}**\nby {}", esc(&s.track.title), esc(&s.track.artist)); 16 16 + desc.push_str(&format!("\non *{}*", esc(&s.track.album))); 17 17 + 18 18 + DiscordEmbed { 19 19 + title: s.user.display_name.clone(), 20 20 + url, 21 21 + description: Some(desc), 22 22 + timestamp: Some(s.played_at.clone()), 23 23 + thumbnail: s.track.artwork_url.clone().map(|u| DiscordThumb { url: u }), 24 24 + footer: Some(DiscordFooter { 25 25 + text: format!("Rocksky • {}", s.user.handle.clone()), 26 26 + }), 27 27 + } 28 28 + } 29 29 + 30 30 + pub async fn post_embeds( 31 31 + http: &Client, 32 32 + discord_webhook_url: &str, 33 33 + embeds: Vec<DiscordEmbed>, 34 34 + ) -> reqwest::Result<()> { 35 35 + if discord_webhook_url.is_empty() { 36 36 + println!("DISCORD_WEBHOOK_URL is not set, skipping webhook post"); 37 37 + return Ok(()); 38 38 + } 39 39 + 40 40 + let body = DiscordWebhookPayload { 41 41 + content: String::new(), 42 42 + embeds, 43 43 + }; 44 44 + let res = http.post(discord_webhook_url).json(&body).send().await?; 45 45 + if !res.status().is_success() { 46 46 + let text = res.text().await.unwrap_or_default(); 47 47 + eprintln!("Failed to post to Discord webhook: {}", text); 48 48 + } 49 49 + Ok(()) 50 50 + } 51 51 + 52 52 + fn esc(s: &str) -> String { 53 53 + s.replace(['*', '_', '~', '`', '>'], "\\$0") 54 54 + }

+77

crates/jetstream/src/webhook/discord/model.rs

··· 1 1 + use serde::{Deserialize, Serialize}; 2 2 + 3 3 + #[derive(Debug, Deserialize, Serialize, Clone)] 4 4 + pub struct WebhookEnvelope { 5 5 + #[serde(default)] 6 6 + pub r#type: String, 7 7 + #[serde(default)] 8 8 + pub id: Option<String>, 9 9 + #[serde(default)] 10 10 + pub delivered_at: Option<String>, 11 11 + pub data: ScrobbleData, 12 12 + } 13 13 + 14 14 + #[derive(Debug, Deserialize, Serialize, Clone)] 15 15 + pub struct ScrobbleData { 16 16 + pub user: User, 17 17 + pub track: Track, 18 18 + pub played_at: String, // ISO 8601 19 19 + } 20 20 + 21 21 + #[derive(Debug, Deserialize, Serialize, Clone)] 22 22 + pub struct User { 23 23 + pub did: String, 24 24 + pub display_name: String, 25 25 + pub handle: String, 26 26 + } 27 27 + 28 28 + #[derive(Debug, Deserialize, Serialize, Clone)] 29 29 + pub struct Track { 30 30 + pub title: String, 31 31 + pub artist: String, 32 32 + pub album: String, 33 33 + pub duration: i32, 34 34 + #[serde(default)] 35 35 + pub artwork_url: Option<String>, 36 36 + #[serde(default)] 37 37 + pub spotify_url: Option<String>, 38 38 + #[serde(default)] 39 39 + pub tidal_url: Option<String>, 40 40 + #[serde(default)] 41 41 + pub youtube_url: Option<String>, 42 42 + } 43 43 + 44 44 + /* ---------- Discord payloads ---------- */ 45 45 + 46 46 + #[derive(Debug, Serialize)] 47 47 + pub struct DiscordWebhookPayload { 48 48 + #[serde(default)] 49 49 + pub content: String, 50 50 + #[serde(default)] 51 51 + pub embeds: Vec<DiscordEmbed>, 52 52 + } 53 53 + 54 54 + #[derive(Debug, Serialize)] 55 55 + pub struct DiscordEmbed { 56 56 + pub title: String, 57 57 + #[serde(skip_serializing_if = "Option::is_none")] 58 58 + pub url: Option<String>, 59 59 + #[serde(skip_serializing_if = "Option::is_none")] 60 60 + pub description: Option<String>, 61 61 + #[serde(skip_serializing_if = "Option::is_none")] 62 62 + pub timestamp: Option<String>, 63 63 + #[serde(skip_serializing_if = "Option::is_none")] 64 64 + pub thumbnail: Option<DiscordThumb>, 65 65 + #[serde(skip_serializing_if = "Option::is_none")] 66 66 + pub footer: Option<DiscordFooter>, 67 67 + } 68 68 + 69 69 + #[derive(Debug, Serialize)] 70 70 + pub struct DiscordThumb { 71 71 + pub url: String, 72 72 + } 73 73 + 74 74 + #[derive(Debug, Serialize)] 75 75 + pub struct DiscordFooter { 76 76 + pub text: String, 77 77 + }

+1

crates/jetstream/src/webhook/mod.rs

··· 1 1 + pub mod discord;

+139

crates/jetstream/src/webhook_worker.rs

··· 1 1 + use crate::webhook::discord::{self, model::WebhookEnvelope}; 2 2 + use anyhow::Error; 3 3 + use std::{ 4 4 + env, 5 5 + sync::Arc, 6 6 + time::{Duration, Instant}, 7 7 + }; 8 8 + use tokio::{sync::Mutex, time::interval}; 9 9 + 10 10 + #[derive(Clone)] 11 11 + pub struct AppState { 12 12 + pub redis: redis::Client, 13 13 + pub queue_key: String, 14 14 + } 15 15 + 16 16 + pub async fn start_worker(state: Arc<Mutex<AppState>>) -> Result<(), Error> { 17 17 + let max_rps: u32 = env::var("MAX_REQUESTS_PER_SEC") 18 18 + .ok() 19 19 + .and_then(|s| s.parse().ok()) 20 20 + .unwrap_or(5); 21 21 + let max_embeds_per: usize = env::var("MAX_EMBEDS_PER_REQUEST") 22 22 + .ok() 23 23 + .and_then(|s| s.parse().ok()) 24 24 + .unwrap_or(10); 25 25 + let batch_window_ms: u64 = env::var("BATCH_WINDOW_MS") 26 26 + .ok() 27 27 + .and_then(|s| s.parse().ok()) 28 28 + .unwrap_or(400); 29 29 + let discord_webhook_url = env::var("DISCORD_WEBHOOK_URL").unwrap_or(String::new()); 30 30 + 31 31 + tokio::spawn(run_worker( 32 32 + state.clone(), 33 33 + discord_webhook_url, 34 34 + max_rps, 35 35 + max_embeds_per, 36 36 + Duration::from_millis(batch_window_ms), 37 37 + )); 38 38 + 39 39 + Ok(()) 40 40 + } 41 41 + 42 42 + async fn run_worker( 43 43 + st: Arc<Mutex<AppState>>, 44 44 + discord_webhook_url: String, 45 45 + max_rps: u32, 46 46 + max_embeds_per: usize, 47 47 + batch_window: Duration, 48 48 + ) { 49 49 + let http = reqwest::Client::builder() 50 50 + .user_agent("rocksky-discord-bridge/0.1") 51 51 + .build() 52 52 + .expect("http client"); 53 53 + 54 54 + let mut tokens = max_rps as i32; 55 55 + let mut refill = interval(Duration::from_secs(1)); 56 56 + refill.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); 57 57 + 58 58 + loop { 59 59 + tokio::select! { 60 60 + _ = refill.tick() => { 61 61 + tokens = (tokens + max_rps as i32).min(max_rps as i32); 62 62 + } 63 63 + _ = tokio::time::sleep(Duration::from_millis(10)) => { /* tick */ } 64 64 + } 65 65 + 66 66 + if tokens <= 0 { 67 67 + continue; 68 68 + } 69 69 + 70 70 + let start = Instant::now(); 71 71 + let mut embeds = Vec::with_capacity(max_embeds_per); 72 72 + 73 73 + while embeds.len() < max_embeds_per && start.elapsed() < batch_window { 74 74 + match brpop_once(st.clone(), 1).await { 75 75 + Ok(Some(json_str)) => { 76 76 + if let Ok(env) = serde_json::from_str::<WebhookEnvelope>(&json_str) { 77 77 + embeds.push(discord::embed_from_scrobble(&env.data)); 78 78 + } 79 79 + } 80 80 + Ok(None) => break, 81 81 + Err(e) => { 82 82 + eprintln!("Failed to pop from Redis: {}", e); 83 83 + break; 84 84 + } 85 85 + } 86 86 + } 87 87 + 88 88 + if embeds.is_empty() { 89 89 + tokio::time::sleep(Duration::from_millis(50)).await; 90 90 + continue; 91 91 + } 92 92 + 93 93 + tokens -= 1; 94 94 + 95 95 + if let Err(e) = discord::post_embeds(&http, &discord_webhook_url, embeds).await { 96 96 + eprintln!("Failed to post to Discord webhook: {}", e); 97 97 + } 98 98 + } 99 99 + } 100 100 + 101 101 + async fn brpop_once( 102 102 + state: Arc<Mutex<AppState>>, 103 103 + timeout_secs: u64, 104 104 + ) -> redis::RedisResult<Option<String>> { 105 105 + let AppState { 106 106 + redis: client, 107 107 + queue_key: key, 108 108 + } = &*state.lock().await; 109 109 + let mut conn = client.get_multiplexed_async_connection().await?; 110 110 + let res: Option<(String, String)> = redis::cmd("BRPOP") 111 111 + .arg(key) 112 112 + .arg(timeout_secs as usize) 113 113 + .query_async(&mut conn) 114 114 + .await?; 115 115 + Ok(res.map(|(_, v)| v)) 116 116 + } 117 117 + 118 118 + pub async fn push_to_queue( 119 119 + state: Arc<Mutex<AppState>>, 120 120 + item: &WebhookEnvelope, 121 121 + ) -> redis::RedisResult<()> { 122 122 + let payload = serde_json::to_string(item).unwrap(); 123 123 + let AppState { 124 124 + redis: client, 125 125 + queue_key: key, 126 126 + } = &*state.lock().await; 127 127 + let mut conn = client.get_multiplexed_async_connection().await?; 128 128 + let _: () = redis::pipe() 129 129 + .cmd("RPUSH") 130 130 + .arg(key) 131 131 + .arg(payload) 132 132 + .ignore() 133 133 + .cmd("EXPIRE") 134 134 + .arg(key) 135 135 + .arg(60 * 60 * 24) // 24h 136 136 + .query_async(&mut conn) 137 137 + .await?; 138 138 + Ok(()) 139 139 + }