patch of feat: implement Discord webhook integration for scrobble events · round #0 · pull #1 · rocksky.app/rocksky

+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
        
       }

+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
        
                           }

+1

crates/jetstream/src/webhook/mod.rs

···

       1
       1
       +
       pub mod discord;