patch of feat(feed): implement feed service · round #1 · pull #8 · rocksky.app/rocksky

+35

crates/feed/Cargo.toml

···

       1
       1
       +
       [package]

     

       2
       2
       +
       name = "rocksky-feed"

     

       3
       3
       +
       version = "0.1.0"

     

       4
       4
       +
       authors.workspace = true

     

       5
       5
       +
       edition.workspace = true

     

       6
       6
       +
       license.workspace = true

     

       7
       7
       +
       repository.workspace = true

     

       8
       8
       +
       

     

       9
       9
       +
       [dependencies]

     

       10
       10
       +
       warp = "0.3.7"

     

       11
       11
       +
       owo-colors = "4.1.0"

     

       12
       12
       +
       anyhow = "1.0.100"

     

       13
       13
       +
       atrium-api = "0.25.5"

     

       14
       14
       +
       atrium-xrpc-client = "0.5.14"

     

       15
       15
       +
       chrono = { version = "= 0.4.39", features = ["serde"] }

     

       16
       16
       +
       duckdb = { version = "1.2.0", features = ["chrono"] }

     

       17
       17
       +
       jetstream-oxide = "=0.1.2"

     

       18
       18
       +
       reqwest = { version = "0.12.12", features = [

     

       19
       19
       +
         "rustls-tls",

     

       20
       20
       +
         "json",

     

       21
       21
       +
         "multipart",

     

       22
       22
       +
       ], default-features = false }

     

       23
       23
       +
       serde = "1.0.227"

     

       24
       24
       +
       serde_json = "1.0.145"

     

       25
       25
       +
       tokio = { version = "1.43.0", features = ["full"] }

     

       26
       26
       +
       tracing = "0.1.41"

     

       27
       27
       +
       dotenv = "0.15.0"

     

       28
       28
       +
       sqlx = { version = "0.8.3", features = [

     

       29
       29
       +
         "runtime-tokio",

     

       30
       30
       +
         "tls-rustls",

     

       31
       31
       +
         "postgres",

     

       32
       32
       +
         "chrono",

     

       33
       33
       +
         "derive",

     

       34
       34
       +
         "macros",

     

       35
       35
       +
       ] }

+35

crates/feed/src/config.rs

···

       1
       1
       +
       use std::env;

     

       2
       2
       +
       

     

       3
       3
       +
       use dotenv::dotenv;

     

       4
       4
       +
       

     

       5
       5
       +
       #[derive(Debug, Clone)]

     

       6
       6
       +
       /// Configuration values for a Feed service

     

       7
       7
       +
       pub struct Config {

     

       8
       8
       +
           /// Your account's decentralized identifier (DID)

     

       9
       9
       +
           /// A DID is a persistent, long-term identifier for every account. Usually look like did:plc:ewvi7nxzyoun6zhxrhs64oiz.

     

       10
       10
       +
           pub publisher_did: String,

     

       11
       11
       +
           /// The host name for your feed generator.

     

       12
       12
       +
           ///

     

       13
       13
       +
           /// For example: if github were to host a feed generator service at their domain they would set this value to `github.com`

     

       14
       14
       +
           ///

     

       15
       15
       +
           /// You can develop your feed locally without setting this to a real value. However, when publishing, this value must be a domain that:

     

       16
       16
       +
           /// - Points to your service.

     

       17
       17
       +
           /// - Is secured with SSL (HTTPS).

     

       18
       18
       +
           /// - Is accessible on the public internet.

     

       19
       19
       +
           pub feed_generator_hostname: String,

     

       20
       20
       +
       }

     

       21
       21
       +
       

     

       22
       22
       +
       impl Config {

     

       23
       23
       +
           /// Loads the config from a local .env file containing these variables

     

       24
       24
       +
           /// PUBLISHER_DID

     

       25
       25
       +
           /// FEED_GENERATOR_HOSTNAME

     

       26
       26
       +
           pub fn load_env_config() -> Self {

     

       27
       27
       +
               dotenv().expect("Missing .env");

     

       28
       28
       +
               Config {

     

       29
       29
       +
                   publisher_did: env::var("PUBLISHER_DID")

     

       30
       30
       +
                       .expect(".env file is missing an entry for PUBLISHER_DID"),

     

       31
       31
       +
                   feed_generator_hostname: env::var("FEED_GENERATOR_HOSTNAME")

     

       32
       32
       +
                       .expect(".env file is missing an entry for FEED_GENERATOR_HOSTNAME"),

     

       33
       33
       +
               }

     

       34
       34
       +
           }

     

       35
       35
       +
       }

+10

crates/feed/src/feed_handler.rs

···

       1
       1
       +
       use crate::types::{FeedResult, Request, Scrobble, Uri};

     

       2
       2
       +
       

     

       3
       3
       +
       /// A feed handler is responsible for

     

       4
       4
       +
       /// - Storing and managing firehose input.

     

       5
       5
       +
       /// - Serving responses to feed requests with `serve_feed`

     

       6
       6
       +
       pub trait FeedHandler {

     

       7
       7
       +
           fn insert_scrobble(&self, scrobble: Scrobble) -> impl std::future::Future<Output = ()> + Send;

     

       8
       8
       +
           fn delete_scrobble(&self, uri: Uri) -> impl std::future::Future<Output = ()> + Send;

     

       9
       9
       +
           fn serve_feed(&self, request: Request) -> impl std::future::Future<Output = FeedResult> + Send;

     

       10
       10
       +
       }

+54

crates/feed/src/types.rs

···

       1
       1
       +
       #[derive(Debug, Clone)]

     

       2
       2
       +
       pub struct Request {

     

       3
       3
       +
           pub cursor: Option<String>,

     

       4
       4
       +
           pub feed: String,

     

       5
       5
       +
           pub limit: Option<u8>,

     

       6
       6
       +
       }

     

       7
       7
       +
       

     

       8
       8
       +
       #[derive(Debug, Clone)]

     

       9
       9
       +
       pub struct Cid(pub String);

     

       10
       10
       +
       

     

       11
       11
       +
       #[derive(Debug, Clone)]

     

       12
       12
       +
       pub struct Did(pub String);

     

       13
       13
       +
       

     

       14
       14
       +
       #[derive(Debug, Clone, Hash, PartialEq, Eq)]

     

       15
       15
       +
       pub struct Uri(pub String);

     

       16
       16
       +
       

     

       17
       17
       +
       #[derive(Debug, Clone)]

     

       18
       18
       +
       pub struct FeedResult {

     

       19
       19
       +
           pub cursor: Option<String>,

     

       20
       20
       +
           pub feed: Vec<Uri>,

     

       21
       21
       +
       }

     

       22
       22
       +
       

     

       23
       23
       +
       pub struct FeedSkeletonQuery {}

     

       24
       24
       +
       

     

       25
       25
       +
       #[derive(Deserialize)]

     

       26
       26
       +
       pub struct FeedSkeletonParameters {}

     

       27
       27
       +
       

     

       28
       28
       +
       impl Into<FeedSkeletonQuery> for FeedSkeletonParameters {

     

       29
       29
       +
           fn into(self) -> FeedSkeletonQuery {

     

       30
       30
       +
               FeedSkeletonQuery {}

     

       31
       31
       +
           }

     

       32
       32
       +
       }

     

       33
       33
       +
       

     

       34
       34
       +
       #[derive(Debug, Clone)]

     

       35
       35
       +
       pub struct Scrobble {}

     

       36
       36
       +
       

     

       37
       37
       +
       use serde::{Deserialize, Serialize};

     

       38
       38
       +
       

     

       39
       39
       +
       #[derive(Serialize)]

     

       40
       40
       +
       pub(crate) struct DidDocument {

     

       41
       41
       +
           #[serde(rename = "@context")]

     

       42
       42
       +
           pub(crate) context: Vec<String>,

     

       43
       43
       +
           pub(crate) id: String,

     

       44
       44
       +
           pub(crate) service: Vec<Service>,

     

       45
       45
       +
       }

     

       46
       46
       +
       

     

       47
       47
       +
       #[derive(Serialize)]

     

       48
       48
       +
       pub(crate) struct Service {

     

       49
       49
       +
           pub(crate) id: String,

     

       50
       50
       +
           #[serde(rename = "type")]

     

       51
       51
       +
           pub(crate) type_: String,

     

       52
       52
       +
           #[serde(rename = "serviceEndpoint")]

     

       53
       53
       +
           pub(crate) service_endpoint: String,

     

       54
       54
       +
       }

crates/rockskyd/Cargo.toml

···

       20
       20
        
       rocksky-spotify = { path = "../spotify" }

     

       21
       21
        
       rocksky-tracklist = { path = "../tracklist" }

     

       22
       22
        
       rocksky-webscrobbler = { path = "../webscrobbler" }

     

       23
       23
       +
       rocksky-feed = { path = "../feed" }

     

       23
       24
        
       tracing = "0.1.41"

     

       24
       25
        
       tracing-subscriber = "0.3.20"

     

       25
       26
        
       tracing-log = "0.2.0"

crates/rockskyd/src/cmd/mod.rs

···

       1
       1
        
       pub mod analytics;

     

       2
       2
        
       pub mod dropbox;

     

       3
       3
       +
       pub mod feed;

     

       3
       4
        
       pub mod googledrive;

     

       4
       5
        
       pub mod jetstream;

     

       5
       6
        
       pub mod playlist;

crates/rockskyd/src/main.rs

···

       32
       32
        
               .subcommand(Command::new("spotify").about("Start Spotify Listener Service"))

     

       33
       33
        
               .subcommand(Command::new("tracklist").about("Start User Current Track Queue Service"))

     

       34
       34
        
               .subcommand(Command::new("webscrobbler").about("Start Webscrobbler API"))

     

       35
       35
       +
               .subcommand(

     

       36
       36
       +
                   Command::new("feed")

     

       37
       37
       +
                       .about("Feed related commands")

     

       38
       38
       +
                       .subcommand(Command::new("serve").about("Serve the Rocksky Feed API")),

     

       39
       39
       +
               )

     

       35
       40
        
       }

     

       36
       41
        
       

     

       37
       42
        
       #[tokio::main]

     
···

       85
       90
        
               Some(("webscrobbler", _)) => {

     

       86
       91
        
                   cmd::webscrobbler::start_webscrobbler_service().await?;

     

       87
       92
        
               }

     

       93
       93
       +
               Some(("feed", sub_m)) => match sub_m.subcommand() {

     

       94
       94
       +
                   Some(("serve", _)) => cmd::feed::serve().await?,

     

       95
       95
       +
                   _ => println!("Unknown feed command"),

     

       96
       96
       +
               },

     

       88
       97
        
               _ => {

     

       89
       98
        
                   println!("No valid subcommand was used. Use --help to see available commands.");

     

       90
       99
        
               }

+25 -6

crates/feed/src/feed.rs

···

       1
       1
        
       use crate::config::Config;

     

       2
       2
        
       use crate::types::{DidDocument, FeedSkeletonParameters, Service};

     

       3
       3
        
       use crate::{feed_handler::FeedHandler, types::FeedSkeletonQuery};

     

       4
       4
       +
       use anyhow::Error;

     

       5
       5
       +
       use sqlx::postgres::PgPoolOptions;

     

       6
       6
       +
       use sqlx::{Pool, Postgres};

     

       7
       7
       +
       use std::env;

     

       4
       8
        
       use std::fmt::Debug;

     

       5
       9
        
       use std::net::SocketAddr;

     

       10
       10
       +
       use std::sync::Arc;

     

       6
       11
        
       use warp::Filter;

     

       7
       12
        
       

     

       8
       13
        
       /// A `Feed` stores a `FeedHandler`, handles feed server endpoints & connects to the Firehose using the `start` methods.

     
···

       21
       26
        
               &mut self,

     

       22
       27
        
               name: impl AsRef<str>,

     

       23
       28
        
               address: impl Into<SocketAddr> + Debug + Clone + Send,

     

       24
       24
       -
           ) -> impl std::future::Future<Output = ()> + Send {

     

       29
       29
       +
           ) -> impl std::future::Future<Output = Result<(), Error>> + Send {

     

       25
       30
        
               self.start_with_config(name, Config::load_env_config(), address)

     

       26
       31
        
           }

     

       27
       32
        
       

     
···

       39
       44
        
               name: impl AsRef<str>,

     

       40
       45
        
               config: Config,

     

       41
       46
        
               address: impl Into<SocketAddr> + Debug + Clone + Send,

     

       42
       42
       -
           ) -> impl std::future::Future<Output = ()> + Send {

     

       47
       47
       +
           ) -> impl std::future::Future<Output = Result<(), Error>> + Send {

     

       43
       48
        
               let handler = self.handler();

     

       44
       49
        
               let address = address.clone();

     

       45
       50
        
               let feed_name = name.as_ref().to_string();

     

       46
       51
        
       

     

       47
       52
        
               async move {

     

       48
       53
        
                   let config = config;

     

       54
       54
       +
                   let pool = PgPoolOptions::new()

     

       55
       55
       +
                       .max_connections(5)

     

       56
       56
       +
                       .connect(&env::var("XATA_POSTGRES_URL")?)

     

       57
       57
       +
                       .await?;

     

       58
       58
       +
                   let pool = Arc::new(pool);

     

       59
       59
       +
                   let db_filter = warp::any().map(move || pool.clone());

     

       49
       60
        
       

     

       50
       61
        
                   let did_config = config.clone();

     

       51
       62
        
                   let did_json = warp::path(".well-known")

     
···

       56
       67
        
                   let describe_feed_generator = warp::path("xrpc")

     

       57
       68
        
                       .and(warp::path("app.rocksky.feed.describeFeedGenerator"))

     

       58
       69
        
                       .and(warp::get())

     

       59
       59
       -
                       .and_then(move || describe_feed_generator(feed_name.clone()));

     

       70
       70
       +
                       .and(db_filter.clone())

     

       71
       71
       +
                       .and_then(move |_pool: Arc<Pool<Postgres>>| {

     

       72
       72
       +
                           describe_feed_generator(feed_name.clone())

     

       73
       73
       +
                       });

     

       60
       74
        
       

     

       61
       75
        
                   let get_feed_handler = handler.clone();

     

       62
       76
        
                   let get_feed_skeleton = warp::path("xrpc")

     

       63
       77
        
                       .and(warp::path("app.rocksky.feed.getFeedSkeleton"))

     

       64
       78
        
                       .and(warp::get())

     

       65
       79
        
                       .and(warp::query::<FeedSkeletonParameters>())

     

       66
       66
       -
                       .and_then(move |query: FeedSkeletonParameters| {

     

       67
       67
       -
                           get_feed_skeleton::<Handler>(query.into(), get_feed_handler.clone())

     

       68
       68
       -
                       });

     

       80
       80
       +
                       .and(db_filter.clone())

     

       81
       81
       +
                       .and_then(

     

       82
       82
       +
                           move |query: FeedSkeletonParameters, _pool: Arc<Pool<Postgres>>| {

     

       83
       83
       +
                               get_feed_skeleton::<Handler>(query.into(), get_feed_handler.clone())

     

       84
       84
       +
                           },

     

       85
       85
       +
                       );

     

       69
       86
        
       

     

       70
       87
        
                   let api = did_json.or(describe_feed_generator).or(get_feed_skeleton);

     

       71
       88
        
       

     
···

       101
       118
        
                   tokio::join!(feed_server.run(address), firehose_listener)

     

       102
       119
        
                       .1

     

       103
       120
        
                       .expect("Couldn't await tasks");

     

       121
       121
       +
       

     

       122
       122
       +
                   Ok::<(), Error>(())

     

       104
       123
        
               }

     

       105
       124
        
           }

     

       106
       125
        
       }

+4 -3

crates/feed/src/lib.rs

···

       1
       1
       +
       use anyhow::Error;

     

       1
       2
        
       use std::{env, net::SocketAddr, sync::Arc};

     

       2
       2
       -
       

     

       3
       3
        
       use tokio::sync::Mutex;

     

       4
       4
        
       

     

       5
       5
        
       use crate::{

     
···

       42
       42
        
           }

     

       43
       43
        
       }

     

       44
       44
        
       

     

       45
       45
       -
       pub async fn run() {

     

       45
       45
       +
       pub async fn run() -> Result<(), Error> {

     

       46
       46
        
           let mut feed = RecentlyPlayedFeed {

     

       47
       47
        
               handler: RecentlyPlayedFeedHandler {

     

       48
       48
        
                   scrobbles: Arc::new(Mutex::new(Vec::new())),

     
···

       53
       53
        
           let addr_str = format!("{}:{}", host, port);

     

       54
       54
        
           let addr: SocketAddr = addr_str.parse().expect("Invalid address format");

     

       55
       55
        
       

     

       56
       56
       -
           feed.start("RecentlyPlayed", addr).await;

     

       56
       56
       +
           feed.start("RecentlyPlayed", addr).await?;

     

       57
       57
       +
           Ok(())

     

       57
       58
        
       }

+87

Cargo.lock

···

       3466
       3466
        
        "windows-sys 0.59.0",

     

       3467
       3467
        
       ]

     

       3468
       3468
        
       

     

       3469
       3469
       +
       [[package]]

     

       3470
       3470
       +
       name = "moka"

     

       3471
       3471
       +
       version = "0.12.11"

     

       3472
       3472
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3473
       3473
       +
       checksum = "8261cd88c312e0004c1d51baad2980c66528dfdb2bee62003e643a4d8f86b077"

     

       3474
       3474
       +
       dependencies = [

     

       3475
       3475
       +
        "async-lock",

     

       3476
       3476
       +
        "crossbeam-channel",

     

       3477
       3477
       +
        "crossbeam-epoch",

     

       3478
       3478
       +
        "crossbeam-utils",

     

       3479
       3479
       +
        "equivalent",

     

       3480
       3480
       +
        "event-listener",

     

       3481
       3481
       +
        "futures-util",

     

       3482
       3482
       +
        "parking_lot",

     

       3483
       3483
       +
        "portable-atomic",

     

       3484
       3484
       +
        "rustc_version",

     

       3485
       3485
       +
        "smallvec",

     

       3486
       3486
       +
        "tagptr",

     

       3487
       3487
       +
        "uuid",

     

       3488
       3488
       +
       ]

     

       3489
       3489
       +
       

     

       3490
       3490
       +
       [[package]]

     

       3491
       3491
       +
       name = "multer"

     

       3492
       3492
       +
       version = "2.1.0"

     

       3493
       3493
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3494
       3494
       +
       checksum = "01acbdc23469fd8fe07ab135923371d5f5a422fbf9c522158677c8eb15bc51c2"

     

       3495
       3495
       +
       dependencies = [

     

       3496
       3496
       +
        "bytes",

     

       3497
       3497
       +
        "encoding_rs",

     

       3498
       3498
       +
        "futures-util",

     

       3499
       3499
       +
        "http 0.2.12",

     

       3500
       3500
       +
        "httparse",

     

       3501
       3501
       +
        "log",

     

       3502
       3502
       +
        "memchr",

     

       3503
       3503
       +
        "mime",

     

       3504
       3504
       +
        "spin",

     

       3505
       3505
       +
        "version_check",

     

       3506
       3506
       +
       ]

     

       3507
       3507
       +
       

     

       3508
       3508
       +
       [[package]]

     

       3509
       3509
       +
       name = "multibase"

     

       3510
       3510
       +
       version = "0.9.1"

     

       3511
       3511
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3512
       3512
       +
       checksum = "9b3539ec3c1f04ac9748a260728e855f261b4977f5c3406612c884564f329404"

     

       3513
       3513
       +
       dependencies = [

     

       3514
       3514
       +
        "base-x",

     

       3515
       3515
       +
        "data-encoding",

     

       3516
       3516
       +
        "data-encoding-macro",

     

       3517
       3517
       +
       ]

     

       3518
       3518
       +
       

     

       3519
       3519
       +
       [[package]]

     

       3520
       3520
       +
       name = "multihash"

     

       3521
       3521
       +
       version = "0.19.3"

     

       3522
       3522
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3523
       3523
       +
       checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d"

     

       3524
       3524
       +
       dependencies = [

     

       3525
       3525
       +
        "core2",

     

       3526
       3526
       +
        "serde",

     

       3527
       3527
       +
        "unsigned-varint",

     

       3528
       3528
       +
       ]

     

       3529
       3529
       +
       

     

       3469
       3530
        
       [[package]]

     

       3470
       3531
        
       name = "nanoid"

     

       3471
       3532
        
       version = "0.4.0"

     
···

       3475
       3536
        
        "rand 0.8.5",

     

       3476
       3537
        
       ]

     

       3477
       3538
        
       

     

       3539
       3539
       +
       [[package]]

     

       3540
       3540
       +
       name = "nanorand"

     

       3541
       3541
       +
       version = "0.7.0"

     

       3542
       3542
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3543
       3543
       +
       checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3"

     

       3544
       3544
       +
       dependencies = [

     

       3545
       3545
       +
        "getrandom 0.2.16",

     

       3546
       3546
       +
       ]

     

       3547
       3547
       +
       

     

       3548
       3548
       +
       [[package]]

     

       3549
       3549
       +
       name = "native-tls"

     

       3550
       3550
       +
       version = "0.2.14"

     

       3551
       3551
       +
       source = "registry+https://github.com/rust-lang/crates.io-index"

     

       3552
       3552
       +
       checksum = "87de3442987e9dbec73158d5c715e7ad9072fda936bb03d19d7fa10e00520f0e"

     

       3553
       3553
       +
       dependencies = [

     

       3554
       3554
       +
        "libc",

     

       3555
       3555
       +
        "log",

     

       3556
       3556
       +
        "openssl",

     

       3557
       3557
       +
        "openssl-probe",

     

       3558
       3558
       +
        "openssl-sys",

     

       3559
       3559
       +
        "schannel",

     

       3560
       3560
       +
        "security-framework 2.11.1",

     

       3561
       3561
       +
        "security-framework-sys",

     

       3562
       3562
       +
        "tempfile",

     

       3563
       3563
       +
       ]

     

       3564
       3564
       +
       

     

       3478
       3565
        
       [[package]]

     

       3479
       3566
        
       name = "nkeys"

     

       3480
       3567
        
       version = "0.4.4"