-43

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 - reqwest = { version = "0.12.12", features = [ 18 18 - "rustls-tls", 19 19 - "json", 20 20 - "multipart", 21 21 - ], default-features = false } 22 22 - serde = "1.0.227" 23 23 - serde_json = "1.0.145" 24 24 - tokio = { version = "1.43.0", features = ["full"] } 25 25 - tracing = "0.1.41" 26 26 - dotenv = "0.15.0" 27 27 - sqlx = { version = "0.8.3", features = [ 28 28 - "runtime-tokio", 29 29 - "tls-rustls", 30 30 - "postgres", 31 31 - "chrono", 32 32 - "derive", 33 33 - "macros", 34 34 - ] } 35 35 - tungstenite = { version = "0.26.2", features = ["rustls"] } 36 36 - tokio-tungstenite = { version = "0.26.2", features = [ 37 37 - "tokio-rustls", 38 38 - "rustls-tls-webpki-roots", 39 39 - ] } 40 40 - tokio-stream = { version = "0.1.17", features = ["full"] } 41 41 - async-trait = "0.1.89" 42 42 - xid = "1.1.1" 43 43 - sha256 = "1.6.0"

-31

crates/feed/src/config.rs

··· 1 1 - use std::env; 2 2 - #[derive(Debug, Clone)] 3 3 - /// Configuration values for a Feed service 4 4 - pub struct Config { 5 5 - /// Your account's decentralized identifier (DID) 6 6 - /// A DID is a persistent, long-term identifier for every account. Usually look like did:plc:ewvi7nxzyoun6zhxrhs64oiz. 7 7 - pub publisher_did: String, 8 8 - /// The host name for your feed generator. 9 9 - /// 10 10 - /// For example: if github were to host a feed generator service at their domain they would set this value to `github.com` 11 11 - /// 12 12 - /// You can develop your feed locally without setting this to a real value. However, when publishing, this value must be a domain that: 13 13 - /// - Points to your service. 14 14 - /// - Is secured with SSL (HTTPS). 15 15 - /// - Is accessible on the public internet. 16 16 - pub feed_generator_hostname: String, 17 17 - } 18 18 - 19 19 - impl Config { 20 20 - /// Loads the config from a local .env file containing these variables 21 21 - /// PUBLISHER_DID 22 22 - /// FEED_GENERATOR_HOSTNAME 23 23 - pub fn load_env_config() -> Self { 24 24 - Config { 25 25 - publisher_did: env::var("PUBLISHER_DID") 26 26 - .expect(".env file is missing an entry for PUBLISHER_DID"), 27 27 - feed_generator_hostname: env::var("FEED_GENERATOR_HOSTNAME") 28 28 - .expect(".env file is missing an entry for FEED_GENERATOR_HOSTNAME"), 29 29 - } 30 30 - } 31 31 - }

-40

crates/feed/src/did.rs

··· 1 1 - use anyhow::Error; 2 2 - 3 3 - use crate::types::{Profile, ProfileResponse}; 4 4 - 5 5 - pub async fn did_to_profile(did: &str) -> Result<Profile, Error> { 6 6 - let client = reqwest::Client::new(); 7 7 - let response = client 8 8 - .get(format!("https://plc.directory/{}", did)) 9 9 - .header("Accept", "application/json") 10 10 - .send() 11 11 - .await? 12 12 - .json::<serde_json::Value>() 13 13 - .await?; 14 14 - 15 15 - let handle = response["alsoKnownAs"][0] 16 16 - .as_str() 17 17 - .unwrap_or("") 18 18 - .split("at://") 19 19 - .last() 20 20 - .unwrap_or(""); 21 21 - 22 22 - let service_endpoint = response["service"][0]["serviceEndpoint"] 23 23 - .as_str() 24 24 - .unwrap_or(""); 25 25 - 26 26 - if service_endpoint.is_empty() { 27 27 - return Err(Error::msg("Invalid did")); 28 28 - } 29 29 - 30 30 - let client = reqwest::Client::new(); 31 31 - let mut response = client.get(format!("{}/xrpc/com.atproto.repo.getRecord?repo={}&collection=app.bsky.actor.profile&rkey=self", service_endpoint, did)) 32 32 - .header("Accept", "application/json") 33 33 - .send() 34 34 - .await? 35 35 - .json::<ProfileResponse>() 36 36 - .await?; 37 37 - 38 38 - response.value.handle = Some(handle.to_string()); 39 39 - Ok(response.value) 40 40 - }

-205

crates/feed/src/feed.rs

··· 1 1 - use crate::config::Config; 2 2 - use crate::feed_handler::FeedHandler; 3 3 - use crate::repo::duckdb::DuckdbRepo; 4 4 - use crate::repo::postgres::PostgresRepo; 5 5 - use crate::repo::{Repo, RepoImpl}; 6 6 - use crate::subscriber::ScrobbleSubscriber; 7 7 - use crate::sync::sync_scrobbles; 8 8 - use crate::types::{DidDocument, FeedSkeleton, Request, Service, SkeletonFeedScrobbleData}; 9 9 - use anyhow::Error; 10 10 - use atrium_api::app::bsky::feed::get_feed_skeleton::Parameters as FeedSkeletonQuery; 11 11 - use atrium_api::app::bsky::feed::get_feed_skeleton::ParametersData as FeedSkeletonParameters; 12 12 - use std::env; 13 13 - use std::fmt::Debug; 14 14 - use std::net::SocketAddr; 15 15 - use warp::Filter; 16 16 - 17 17 - /// A `Feed` stores a `FeedHandler`, handles feed server endpoints & connects to the Firehose using the `start` methods. 18 18 - pub trait Feed<Handler: FeedHandler + Clone + Send + Sync + 'static> { 19 19 - fn handler(&mut self) -> Handler; 20 20 - /// Starts the feed generator server & connects to the firehose. 21 21 - /// 22 22 - /// 23 23 - /// - name: The identifying name of your feed. This value is used in the feed URL & when identifying which feed to *unpublish*. This is a separate value from the display name. 24 24 - /// - address: The address to bind the server to 25 25 - /// 26 26 - /// # Panics 27 27 - /// 28 28 - /// Panics if unable to bind to the provided address. 29 29 - fn start( 30 30 - &mut self, 31 31 - name: impl AsRef<str>, 32 32 - address: impl Into<SocketAddr> + Debug + Clone + Send, 33 33 - enable_sync: bool, 34 34 - ) -> impl std::future::Future<Output = Result<(), Error>> + Send { 35 35 - self.start_with_config(name, Config::load_env_config(), address, enable_sync) 36 36 - } 37 37 - 38 38 - /// Starts the feed generator server & connects to the firehose. 39 39 - /// 40 40 - /// - name: The identifying name of your feed. This value is used in the feed URL & when identifying which feed to *unpublish*. This is a separate value from the display name. 41 41 - /// - config: Configuration values, see `Config` 42 42 - /// - address: The address to bind the server to 43 43 - /// 44 44 - /// # Panics 45 45 - /// 46 46 - /// Panics if unable to bind to the provided address. 47 47 - fn start_with_config( 48 48 - &mut self, 49 49 - name: impl AsRef<str>, 50 50 - config: Config, 51 51 - address: impl Into<SocketAddr> + Debug + Clone + Send, 52 52 - enable_sync: bool, 53 53 - ) -> impl std::future::Future<Output = Result<(), Error>> + Send { 54 54 - let handler = self.handler(); 55 55 - let address = address.clone(); 56 56 - let feed_name = name.as_ref().to_string(); 57 57 - 58 58 - async move { 59 59 - let ddb = DuckdbRepo::new().await?; 60 60 - let ddb = RepoImpl::Duckdb(ddb); 61 61 - ddb.clone().create_tables().await?; 62 62 - let ddb_clone = ddb.clone(); 63 63 - 64 64 - let config = config; 65 65 - let pg = RepoImpl::Postgres(PostgresRepo::new().await?); 66 66 - let pg_filter = warp::any().map(move || pg.clone()); 67 67 - 68 68 - let did_config = config.clone(); 69 69 - let did_json = warp::path(".well-known") 70 70 - .and(warp::path("did.json")) 71 71 - .and(warp::get()) 72 72 - .and_then(move || did_json(did_config.clone())); 73 73 - 74 74 - let describe_feed_generator = warp::path("xrpc") 75 75 - .and(warp::path("app.rocksky.feed.describeFeedGenerator")) 76 76 - .and(warp::get()) 77 77 - .and(pg_filter.clone()) 78 78 - .and_then(move |_repo: RepoImpl| describe_feed_generator(feed_name.clone())); 79 79 - 80 80 - let get_feed_handler = handler.clone(); 81 81 - let get_feed_skeleton = warp::path("xrpc") 82 82 - .and(warp::path("app.rocksky.feed.getFeedSkeleton")) 83 83 - .and(warp::get()) 84 84 - .and(warp::query::<FeedSkeletonParameters>()) 85 85 - .and(pg_filter.clone()) 86 86 - .and_then(move |query: FeedSkeletonParameters, repo: RepoImpl| { 87 87 - get_feed_skeleton::<Handler>(query.into(), get_feed_handler.clone(), repo) 88 88 - }); 89 89 - 90 90 - let api = did_json.or(describe_feed_generator).or(get_feed_skeleton); 91 91 - 92 92 - tracing::info!("Serving feed on {}", format!("{:?}", address)); 93 93 - 94 94 - let routes = api.with(warp::log::custom(|info| { 95 95 - let method = info.method(); 96 96 - let path = info.path(); 97 97 - let status = info.status(); 98 98 - let elapsed = info.elapsed().as_millis(); 99 99 - 100 100 - if status.is_success() { 101 101 - tracing::info!( 102 102 - "Method: {}, Path: {}, Status: {}, Elapsed Time: {}ms", 103 103 - method, 104 104 - path, 105 105 - status, 106 106 - elapsed 107 107 - ); 108 108 - } else { 109 109 - tracing::error!( 110 110 - "Method: {}, Path: {}, Status: {}, Elapsed Time: {}ms", 111 111 - method, 112 112 - path, 113 113 - status, 114 114 - elapsed, 115 115 - ); 116 116 - } 117 117 - })); 118 118 - 119 119 - let sync_feed = tokio::spawn(async move { 120 120 - if !enable_sync { 121 121 - return Ok::<(), Error>(()); 122 122 - } 123 123 - sync_scrobbles(ddb_clone).await?; 124 124 - Ok::<(), Error>(()) 125 125 - }); 126 126 - let feed_server = warp::serve(routes); 127 127 - let firehose_listener = tokio::spawn(async move { 128 128 - let jetstream_server = env::var("JETSTREAM_SERVER") 129 129 - .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network".to_string()); 130 130 - let url = format!( 131 131 - "{}/subscribe?wantedCollections=app.rocksky.*", 132 132 - jetstream_server 133 133 - ); 134 134 - let subscriber = ScrobbleSubscriber::new(&url); 135 135 - 136 136 - // loop, reconnecting on failure 137 137 - loop { 138 138 - match subscriber.run(ddb.clone()).await { 139 139 - Ok(_) => {} 140 140 - Err(e) => { 141 141 - tracing::error!(error = %e, "Failed to connect to jetstream server, retrying in 1 second..."); 142 142 - tokio::time::sleep(std::time::Duration::from_secs(1)).await; 143 143 - continue; 144 144 - } 145 145 - } 146 146 - tracing::warn!( 147 147 - "Disconnected from jetstream server, reconnecting in 1 second..." 148 148 - ); 149 149 - tokio::time::sleep(std::time::Duration::from_secs(1)).await; 150 150 - } 151 151 - }); 152 152 - 153 153 - tokio::join!(feed_server.run(address), firehose_listener, sync_feed) 154 154 - .2 155 155 - .expect("Couldn't await tasks")?; 156 156 - 157 157 - Ok::<(), Error>(()) 158 158 - } 159 159 - } 160 160 - } 161 161 - 162 162 - async fn did_json(config: Config) -> Result<impl warp::Reply, warp::Rejection> { 163 163 - Ok(warp::reply::json(&DidDocument { 164 164 - context: vec!["https://www.w3.org/ns/did/v1".to_owned()], 165 165 - id: format!("did:web:{}", config.feed_generator_hostname), 166 166 - service: vec![Service { 167 167 - id: "#rocksky_fg".to_owned(), 168 168 - type_: "RockskyFeedGenerator".to_owned(), 169 169 - service_endpoint: format!("https://{}", config.feed_generator_hostname), 170 170 - }], 171 171 - })) 172 172 - } 173 173 - 174 174 - async fn describe_feed_generator(_feed_name: String) -> Result<impl warp::Reply, warp::Rejection> { 175 175 - Ok(warp::reply::json(&serde_json::json!({}))) 176 176 - } 177 177 - 178 178 - async fn get_feed_skeleton<Handler: FeedHandler>( 179 179 - query: FeedSkeletonQuery, 180 180 - handler: Handler, 181 181 - repo: RepoImpl, 182 182 - ) -> Result<impl warp::Reply, warp::Rejection> { 183 183 - let skeleton = handler 184 184 - .serve_feed( 185 185 - repo, 186 186 - Request { 187 187 - cursor: query.cursor.clone(), 188 188 - feed: query.feed.clone(), 189 189 - limit: query.limit.map(u8::from), 190 190 - }, 191 191 - ) 192 192 - .await; 193 193 - 194 194 - Ok::<warp::reply::Json, warp::Rejection>(warp::reply::json(&FeedSkeleton { 195 195 - cursor: skeleton.cursor, 196 196 - feed: skeleton 197 197 - .feed 198 198 - .into_iter() 199 199 - .map(|uri| SkeletonFeedScrobbleData { 200 200 - feed_context: None, 201 201 - scrobble: uri.0, 202 202 - }) 203 203 - .collect(), 204 204 - })) 205 205 - }

-17

crates/feed/src/feed_handler.rs

··· 1 1 - use crate::{ 2 2 - repo::RepoImpl, 3 3 - types::{FeedResult, Request, Scrobble, Uri}, 4 4 - }; 5 5 - 6 6 - /// A feed handler is responsible for 7 7 - /// - Storing and managing firehose input. 8 8 - /// - Serving responses to feed requests with `serve_feed` 9 9 - pub trait FeedHandler { 10 10 - fn insert_scrobble(&self, scrobble: Scrobble) -> impl std::future::Future<Output = ()> + Send; 11 11 - fn delete_scrobble(&self, uri: Uri) -> impl std::future::Future<Output = ()> + Send; 12 12 - fn serve_feed( 13 13 - &self, 14 14 - repo: RepoImpl, 15 15 - request: Request, 16 16 - ) -> impl std::future::Future<Output = FeedResult> + Send; 17 17 - }

-1

crates/feed/src/feeds/discover/mod.rs

··· 1 1 -

-4

crates/feed/src/feeds/mod.rs

··· 1 1 - pub mod discover; 2 2 - pub mod new; 3 3 - pub mod scrobbles; 4 4 - pub mod trending;

-1

crates/feed/src/feeds/new/mod.rs

··· 1 1 -

-1

crates/feed/src/feeds/scrobbles/mod.rs

··· 1 1 -

-1

crates/feed/src/feeds/trending/mod.rs

··· 1 1 -

-86

crates/feed/src/lib.rs

··· 1 1 - use anyhow::Error; 2 2 - use duckdb::Connection; 3 3 - use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; 4 4 - use std::{env, net::SocketAddr, sync::Arc}; 5 5 - use tokio::sync::Mutex; 6 6 - 7 7 - use crate::{ 8 8 - feed::Feed, 9 9 - feed_handler::FeedHandler, 10 10 - repo::{duckdb::DB_PATH, RepoImpl}, 11 11 - types::{FeedResult, Scrobble}, 12 12 - }; 13 13 - 14 14 - pub mod config; 15 15 - pub mod did; 16 16 - pub mod feed; 17 17 - pub mod feed_handler; 18 18 - pub mod feeds; 19 19 - pub mod repo; 20 20 - pub mod subscriber; 21 21 - pub mod sync; 22 22 - pub mod types; 23 23 - pub mod xata; 24 24 - 25 25 - pub const SCROBBLE_NSID: &str = "app.rocksky.scrobble"; 26 26 - pub const ARTIST_NSID: &str = "app.rocksky.artist"; 27 27 - pub const ALBUM_NSID: &str = "app.rocksky.album"; 28 28 - pub const SONG_NSID: &str = "app.rocksky.song"; 29 29 - pub const PLAYLIST_NSID: &str = "app.rocksky.playlist"; 30 30 - pub const LIKE_NSID: &str = "app.rocksky.like"; 31 31 - pub const SHOUT_NSID: &str = "app.rocksky.shout"; 32 32 - 33 33 - pub struct RecentlyPlayedFeed { 34 34 - handler: RecentlyPlayedFeedHandler, 35 35 - } 36 36 - 37 37 - impl Feed<RecentlyPlayedFeedHandler> for RecentlyPlayedFeed { 38 38 - fn handler(&mut self) -> RecentlyPlayedFeedHandler { 39 39 - self.handler.clone() 40 40 - } 41 41 - } 42 42 - 43 43 - #[derive(Clone)] 44 44 - pub struct RecentlyPlayedFeedHandler { 45 45 - pub conn: Arc<Mutex<Connection>>, 46 46 - pub pool: Arc<Mutex<Pool<Postgres>>>, 47 47 - } 48 48 - 49 49 - impl FeedHandler for RecentlyPlayedFeedHandler { 50 50 - async fn insert_scrobble(&self, _scrobble: Scrobble) { 51 51 - todo!() 52 52 - } 53 53 - 54 54 - async fn delete_scrobble(&self, _uri: types::Uri) { 55 55 - todo!() 56 56 - } 57 57 - 58 58 - async fn serve_feed(&self, _repo: RepoImpl, _request: types::Request) -> FeedResult { 59 59 - FeedResult { 60 60 - feed: vec![], 61 61 - cursor: None, 62 62 - } 63 63 - } 64 64 - } 65 65 - 66 66 - pub async fn run(enable_sync: bool) -> Result<(), Error> { 67 67 - let conn = Connection::open(DB_PATH)?; 68 68 - let pool = PgPoolOptions::new() 69 69 - .max_connections(5) 70 70 - .connect(&env::var("XATA_POSTGRES_URL")?) 71 71 - .await?; 72 72 - 73 73 - let mut feed = RecentlyPlayedFeed { 74 74 - handler: RecentlyPlayedFeedHandler { 75 75 - conn: Arc::new(Mutex::new(conn)), 76 76 - pool: Arc::new(Mutex::new(pool)), 77 77 - }, 78 78 - }; 79 79 - let host = env::var("FEED_HOST").unwrap_or_else(|_| "127.0.0.1".to_string()); 80 80 - let port = env::var("FEED_PORT").unwrap_or_else(|_| "7885".to_string()); 81 81 - let addr_str = format!("{}:{}", host, port); 82 82 - let addr: SocketAddr = addr_str.parse().expect("Invalid address format"); 83 83 - 84 84 - feed.start("RecentlyPlayed", addr, enable_sync).await?; 85 85 - Ok(()) 86 86 - }

-72

crates/feed/src/repo/duckdb/album.rs

··· 1 1 - use std::sync::{Arc, Mutex}; 2 2 - 3 3 - use crate::types::AlbumRecord; 4 4 - use anyhow::Error; 5 5 - use duckdb::params; 6 6 - 7 7 - pub fn save_album( 8 8 - conn: Arc<Mutex<duckdb::Connection>>, 9 9 - uri: &str, 10 10 - record: AlbumRecord, 11 11 - ) -> Result<(), Error> { 12 12 - let uri = uri.to_string(); 13 13 - let conn = conn.lock().unwrap(); 14 14 - 15 15 - let album_hash = sha256::digest(format!("{} - {}", record.title, record.artist).to_lowercase()); 16 16 - 17 17 - match conn.execute( 18 18 - "UPDATE albums SET uri = ? WHERE title = ? AND artist = ? AND uri IS NULL;", 19 19 - params![uri, record.title, record.artist], 20 20 - ) { 21 21 - Ok(x) => { 22 22 - tracing::info!("Album URI updated successfully: {}", x); 23 23 - return Ok(()); 24 24 - } 25 25 - Err(e) => tracing::error!(error = %e, "Error updating album URI"), 26 26 - } 27 27 - 28 28 - match conn.execute( 29 29 - "UPDATE tracks SET album_uri = ? WHERE album = ? AND album_artist = ? AND album_uri IS NULL;", 30 30 - params![uri, record.title, record.artist], 31 31 - ) { 32 32 - Ok(x) => tracing::info!("Track album_uri updated successfully: {}", x), 33 33 - Err(e) => tracing::error!(error = %e, "Error updating track album_uri"), 34 34 - } 35 35 - 36 36 - match conn.execute( 37 37 - "INSERT OR IGNORE INTO albums ( 38 38 - id, 39 39 - title, 40 40 - artist, 41 41 - release_date, 42 42 - album_art, 43 43 - year, 44 44 - sha256, 45 45 - uri 46 46 - ) VALUES ( 47 47 - ?, 48 48 - ?, 49 49 - ?, 50 50 - ?, 51 51 - ?, 52 52 - ?, 53 53 - ?, 54 54 - ? 55 55 - );", 56 56 - params![ 57 57 - xid::new().to_string(), 58 58 - record.title, 59 59 - record.artist, 60 60 - record.release_date, 61 61 - record.album_art_url, 62 62 - record.year, 63 63 - album_hash, 64 64 - uri 65 65 - ], 66 66 - ) { 67 67 - Ok(x) => tracing::info!("Album successfully inserted or updated: {}", x), 68 68 - Err(e) => tracing::error!(error = %e, "Error inserting/updating album"), 69 69 - } 70 70 - 71 71 - Ok(()) 72 72 - }

-76

crates/feed/src/repo/duckdb/artist.rs

··· 1 1 - use std::sync::{Arc, Mutex}; 2 2 - 3 3 - use crate::types::ArtistRecord; 4 4 - use anyhow::Error; 5 5 - use duckdb::params; 6 6 - 7 7 - pub fn save_artist( 8 8 - conn: Arc<Mutex<duckdb::Connection>>, 9 9 - uri: &str, 10 10 - record: ArtistRecord, 11 11 - ) -> Result<(), Error> { 12 12 - let conn = conn.lock().unwrap(); 13 13 - let uri = uri.to_string(); 14 14 - 15 15 - let artist_hash = sha256::digest(record.name.to_lowercase()); 16 16 - 17 17 - match conn.execute( 18 18 - "UPDATE artists SET uri = ?, picture = ? WHERE name = ? AND uri IS NULL;", 19 19 - params![uri, record.picture_url, record.name], 20 20 - ) { 21 21 - Ok(x) => { 22 22 - tracing::info!("Artist URI updated successfully: {}", x); 23 23 - return Ok(()); 24 24 - } 25 25 - Err(e) => tracing::error!(error = %e, "Error updating artist URI"), 26 26 - } 27 27 - 28 28 - match conn.execute( 29 29 - "UPDATE tracks SET artist_uri = ? WHERE album_artist = ? AND artist_uri IS NULL;", 30 30 - params![uri, record.name], 31 31 - ) { 32 32 - Ok(x) => tracing::info!("Track artist_uri updated successfully: {}", x), 33 33 - Err(e) => tracing::error!(error = %e, "Error updating track artist_uri"), 34 34 - } 35 35 - 36 36 - match conn.execute( 37 37 - &format!( 38 38 - "INSERT OR IGNORE INTO artists ( 39 39 - id, 40 40 - name, 41 41 - picture, 42 42 - sha256, 43 43 - uri, 44 44 - tags 45 45 - ) VALUES ( 46 46 - ?, 47 47 - ?, 48 48 - ?, 49 49 - ?, 50 50 - ?, 51 51 - [{}] 52 52 - );", 53 53 - record 54 54 - .tags 55 55 - .as_ref() 56 56 - .map(|tags| tags 57 57 - .iter() 58 58 - .map(|tag| format!("'{}'", tag)) 59 59 - .collect::<Vec<_>>() 60 60 - .join(", ")) 61 61 - .unwrap_or_default() 62 62 - ), 63 63 - params![ 64 64 - xid::new().to_string(), 65 65 - record.name, 66 66 - record.picture_url, 67 67 - artist_hash, 68 68 - uri 69 69 - ], 70 70 - ) { 71 71 - Ok(x) => tracing::info!("Artist successfully inserted or updated: {}", x), 72 72 - Err(e) => tracing::error!(error = %e, "Error inserting/updating artist"), 73 73 - } 74 74 - 75 75 - Ok(()) 76 76 - }

-371

crates/feed/src/repo/duckdb/mod.rs

··· 1 1 - use std::sync::Arc; 2 2 - use std::sync::Mutex; 3 3 - 4 4 - use crate::{ 5 5 - repo::duckdb::{ 6 6 - album::save_album, artist::save_artist, scrobble::save_scrobble, track::save_track, 7 7 - user::save_user, 8 8 - }, 9 9 - types::{AlbumRecord, ArtistRecord, ScrobbleRecord, SongRecord}, 10 10 - }; 11 11 - 12 12 - use super::Repo; 13 13 - use anyhow::Error; 14 14 - use async_trait::async_trait; 15 15 - use tokio::sync::mpsc::Sender; 16 16 - 17 17 - pub mod album; 18 18 - pub mod artist; 19 19 - pub mod scrobble; 20 20 - pub mod track; 21 21 - pub mod user; 22 22 - 23 23 - pub const DB_PATH: &str = "./rocksky-feed.ddb"; 24 24 - 25 25 - #[derive(Debug)] 26 26 - enum SaveMessage { 27 27 - Album { 28 28 - uri: String, 29 29 - record: AlbumRecord, 30 30 - }, 31 31 - Artist { 32 32 - uri: String, 33 33 - record: ArtistRecord, 34 34 - }, 35 35 - Scrobble { 36 36 - did: String, 37 37 - uri: String, 38 38 - record: ScrobbleRecord, 39 39 - }, 40 40 - Track { 41 41 - uri: String, 42 42 - record: SongRecord, 43 43 - }, 44 44 - User { 45 45 - did: String, 46 46 - }, 47 47 - } 48 48 - 49 49 - #[derive(Clone)] 50 50 - pub struct DuckdbRepo { 51 51 - conn: Arc<Mutex<duckdb::Connection>>, 52 52 - save_tx: Sender<SaveMessage>, 53 53 - } 54 54 - 55 55 - impl DuckdbRepo { 56 56 - pub async fn new() -> Result<Self, Error> { 57 57 - let (save_tx, mut save_rx) = tokio::sync::mpsc::channel::<SaveMessage>(1); 58 58 - 59 59 - let conn = Arc::new(Mutex::new(duckdb::Connection::open(DB_PATH)?)); 60 60 - let conn_clone = conn.clone(); 61 61 - tokio::spawn(async move { 62 62 - while let Some(msg) = save_rx.recv().await { 63 63 - let result = match msg { 64 64 - SaveMessage::Album { uri, record } => { 65 65 - save_album(conn_clone.clone(), &uri, record) 66 66 - } 67 67 - SaveMessage::Artist { uri, record } => { 68 68 - save_artist(conn_clone.clone(), &uri, record) 69 69 - } 70 70 - SaveMessage::Scrobble { did, uri, record } => { 71 71 - save_scrobble(conn_clone.clone(), &did, &uri, record) 72 72 - } 73 73 - SaveMessage::Track { uri, record } => { 74 74 - save_track(conn_clone.clone(), &uri, record) 75 75 - } 76 76 - SaveMessage::User { did } => save_user(conn_clone.clone(), &did), 77 77 - }; 78 78 - 79 79 - if let Err(e) = result { 80 80 - tracing::error!(error = %e, "Error saving record to DuckDB"); 81 81 - } 82 82 - } 83 83 - }); 84 84 - 85 85 - Ok(Self { conn, save_tx }) 86 86 - } 87 87 - } 88 88 - 89 89 - #[async_trait] 90 90 - impl Repo for DuckdbRepo { 91 91 - async fn insert_album(self, uri: &str, record: AlbumRecord) -> Result<(), anyhow::Error> { 92 92 - self.save_tx 93 93 - .send(SaveMessage::Album { 94 94 - uri: uri.to_string(), 95 95 - record, 96 96 - }) 97 97 - .await 98 98 - .map_err(|e| anyhow::anyhow!("Failed to send album save message: {}", e))?; 99 99 - Ok(()) 100 100 - } 101 101 - 102 102 - async fn insert_artist(self, uri: &str, record: ArtistRecord) -> Result<(), anyhow::Error> { 103 103 - self.save_tx 104 104 - .send(SaveMessage::Artist { 105 105 - uri: uri.to_string(), 106 106 - record, 107 107 - }) 108 108 - .await 109 109 - .map_err(|e| anyhow::anyhow!("Failed to send artist save message: {}", e))?; 110 110 - Ok(()) 111 111 - } 112 112 - 113 113 - async fn insert_scrobble( 114 114 - self, 115 115 - did: &str, 116 116 - uri: &str, 117 117 - record: ScrobbleRecord, 118 118 - ) -> Result<(), anyhow::Error> { 119 119 - self.save_tx 120 120 - .send(SaveMessage::Scrobble { 121 121 - did: did.to_string(), 122 122 - uri: uri.to_string(), 123 123 - record, 124 124 - }) 125 125 - .await 126 126 - .map_err(|e| anyhow::anyhow!("Failed to send scrobble save message: {}", e))?; 127 127 - Ok(()) 128 128 - } 129 129 - 130 130 - async fn insert_track(self, uri: &str, record: SongRecord) -> Result<(), anyhow::Error> { 131 131 - self.save_tx 132 132 - .send(SaveMessage::Track { 133 133 - uri: uri.to_string(), 134 134 - record, 135 135 - }) 136 136 - .await 137 137 - .map_err(|e| anyhow::anyhow!("Failed to send track save message: {}", e))?; 138 138 - Ok(()) 139 139 - } 140 140 - 141 141 - async fn insert_user(self, did: &str) -> Result<(), anyhow::Error> { 142 142 - self.save_tx 143 143 - .send(SaveMessage::User { 144 144 - did: did.to_string(), 145 145 - }) 146 146 - .await 147 147 - .map_err(|e| anyhow::anyhow!("Failed to send user save message: {}", e))?; 148 148 - Ok(()) 149 149 - } 150 150 - 151 151 - async fn get_albums(self) -> Result<(), anyhow::Error> { 152 152 - todo!() 153 153 - } 154 154 - 155 155 - async fn get_artists(self) -> Result<(), anyhow::Error> { 156 156 - todo!() 157 157 - } 158 158 - 159 159 - async fn get_scrobbles(self) -> Result<(), anyhow::Error> { 160 160 - todo!() 161 161 - } 162 162 - 163 163 - async fn get_tracks(self) -> Result<(), anyhow::Error> { 164 164 - todo!() 165 165 - } 166 166 - 167 167 - async fn get_users(self) -> Result<(), anyhow::Error> { 168 168 - todo!() 169 169 - } 170 170 - 171 171 - async fn get_album(self) -> Result<(), anyhow::Error> { 172 172 - todo!() 173 173 - } 174 174 - 175 175 - async fn get_artist(self) -> Result<(), anyhow::Error> { 176 176 - todo!() 177 177 - } 178 178 - 179 179 - async fn get_track(self) -> Result<(), anyhow::Error> { 180 180 - todo!() 181 181 - } 182 182 - 183 183 - async fn get_user(self) -> Result<(), anyhow::Error> { 184 184 - todo!() 185 185 - } 186 186 - 187 187 - async fn create_tables(self) -> Result<(), anyhow::Error> { 188 188 - let conn = self.conn.lock().unwrap(); 189 189 - conn.execute_batch( 190 190 - "BEGIN; 191 191 - CREATE TABLE IF NOT EXISTS artists ( 192 192 - id VARCHAR PRIMARY KEY, 193 193 - name VARCHAR NOT NULL, 194 194 - biography TEXT, 195 195 - born DATE, 196 196 - born_in VARCHAR, 197 197 - died DATE, 198 198 - picture VARCHAR, 199 199 - sha256 VARCHAR UNIQUE NOT NULL, 200 200 - spotify_link VARCHAR, 201 201 - tidal_link VARCHAR, 202 202 - youtube_link VARCHAR, 203 203 - apple_music_link VARCHAR, 204 204 - uri VARCHAR UNIQUE, 205 205 - tags VARCHAR[], 206 206 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 207 207 - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 208 208 - ); 209 209 - CREATE TABLE IF NOT EXISTS albums ( 210 210 - id VARCHAR PRIMARY KEY, 211 211 - title VARCHAR NOT NULL, 212 212 - artist VARCHAR NOT NULL, 213 213 - release_date DATE, 214 214 - album_art VARCHAR, 215 215 - year INTEGER, 216 216 - spotify_link VARCHAR, 217 217 - tidal_link VARCHAR, 218 218 - youtube_link VARCHAR, 219 219 - apple_music_link VARCHAR, 220 220 - sha256 VARCHAR UNIQUE NOT NULL, 221 221 - uri VARCHAR UNIQUE, 222 222 - artist_uri VARCHAR, 223 223 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 224 224 - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 225 225 - ); 226 226 - CREATE TABLE IF NOT EXISTS tracks ( 227 227 - id VARCHAR PRIMARY KEY, 228 228 - title VARCHAR, 229 229 - artist VARCHAR, 230 230 - album_artist VARCHAR, 231 231 - album_art VARCHAR, 232 232 - album VARCHAR, 233 233 - track_number INTEGER, 234 234 - duration INTEGER, 235 235 - mb_id VARCHAR, 236 236 - youtube_link VARCHAR, 237 237 - spotify_link VARCHAR, 238 238 - tidal_link VARCHAR, 239 239 - apple_music_link VARCHAR, 240 240 - sha256 VARCHAR UNIQUE NOT NULL, 241 241 - lyrics TEXT, 242 242 - composer VARCHAR, 243 243 - genre VARCHAR, 244 244 - disc_number INTEGER, 245 245 - copyright_message VARCHAR, 246 246 - label VARCHAR, 247 247 - uri VARCHAR UNIQUE, 248 248 - artist_uri VARCHAR, 249 249 - album_uri VARCHAR, 250 250 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 251 251 - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 252 252 - ); 253 253 - CREATE TABLE IF NOT EXISTS album_tracks ( 254 254 - id VARCHAR PRIMARY KEY, 255 255 - album_id VARCHAR, 256 256 - track_id VARCHAR, 257 257 - FOREIGN KEY (album_id) REFERENCES albums(id), 258 258 - FOREIGN KEY (track_id) REFERENCES tracks(id), 259 259 - ); 260 260 - CREATE TABLE IF NOT EXISTS users ( 261 261 - id VARCHAR PRIMARY KEY, 262 262 - display_name VARCHAR, 263 263 - did VARCHAR UNIQUE NOT NULL, 264 264 - handle VARCHAR UNIQUE NOT NULL, 265 265 - avatar VARCHAR, 266 266 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 267 267 - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 268 268 - ); 269 269 - CREATE TABLE IF NOT EXISTS playlists ( 270 270 - id VARCHAR PRIMARY KEY, 271 271 - name VARCHAR NOT NULL, 272 272 - description TEXT, 273 273 - picture VARCHAR, 274 274 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 275 275 - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 276 276 - uri VARCHAR UNIQUE, 277 277 - created_by VARCHAR NOT NULL, 278 278 - FOREIGN KEY (created_by) REFERENCES users(id), 279 279 - ); 280 280 - CREATE TABLE IF NOT EXISTS playlist_tracks ( 281 281 - id VARCHAR PRIMARY KEY, 282 282 - playlist_id VARCHAR NOT NULL, 283 283 - track_id VARCHAR NOT NULL, 284 284 - added_by VARCHAR, 285 285 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 286 286 - FOREIGN KEY (playlist_id) REFERENCES playlists(id), 287 287 - FOREIGN KEY (track_id) REFERENCES tracks(id), 288 288 - ); 289 289 - CREATE TABLE IF NOT EXISTS user_tracks ( 290 290 - id VARCHAR PRIMARY KEY, 291 291 - user_id VARCHAR NOT NULL, 292 292 - track_id VARCHAR NOT NULL, 293 293 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 294 294 - FOREIGN KEY (user_id) REFERENCES users(id), 295 295 - FOREIGN KEY (track_id) REFERENCES tracks(id), 296 296 - ); 297 297 - CREATE TABLE IF NOT EXISTS user_albums ( 298 298 - id VARCHAR PRIMARY KEY, 299 299 - user_id VARCHAR, 300 300 - album_id VARCHAR, 301 301 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 302 302 - FOREIGN KEY (user_id) REFERENCES users(id), 303 303 - FOREIGN KEY (album_id) REFERENCES albums(id), 304 304 - ); 305 305 - CREATE TABLE IF NOT EXISTS user_artists ( 306 306 - id VARCHAR PRIMARY KEY, 307 307 - user_id VARCHAR NOT NULL, 308 308 - artist_id VARCHAR NOT NULL, 309 309 - created_at TIMESTAMP, 310 310 - FOREIGN KEY (user_id) REFERENCES users(id), 311 311 - FOREIGN KEY (artist_id) REFERENCES artists(id), 312 312 - ); 313 313 - CREATE TABLE IF NOT EXISTS user_playlists ( 314 314 - id VARCHAR PRIMARY KEY, 315 315 - user_id VARCHAR NOT NULL, 316 316 - playlist_id VARCHAR NOT NULL, 317 317 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 318 318 - FOREIGN KEY (user_id) REFERENCES users(id), 319 319 - FOREIGN KEY (playlist_id) REFERENCES playlists(id), 320 320 - ); 321 321 - CREATE TABLE IF NOT EXISTS loved_tracks ( 322 322 - id VARCHAR PRIMARY KEY, 323 323 - user_id VARCHAR NOT NULL, 324 324 - track_id VARCHAR NOT NULL, 325 325 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 326 326 - FOREIGN KEY (user_id) REFERENCES users(id), 327 327 - FOREIGN KEY (track_id) REFERENCES tracks(id), 328 328 - ); 329 329 - CREATE TABLE IF NOT EXISTS artist_tracks ( 330 330 - id VARCHAR PRIMARY KEY, 331 331 - artist_id VARCHAR NOT NULL, 332 332 - track_id VARCHAR NOT NULL, 333 333 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 334 334 - FOREIGN KEY (artist_id) REFERENCES artists(id), 335 335 - FOREIGN KEY (track_id) REFERENCES tracks(id), 336 336 - ); 337 337 - CREATE TABLE IF NOT EXISTS artist_albums ( 338 338 - id VARCHAR PRIMARY KEY, 339 339 - artist_id VARCHAR NOT NULL, 340 340 - album_id VARCHAR NOT NULL, 341 341 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 342 342 - FOREIGN KEY (artist_id) REFERENCES artists(id), 343 343 - FOREIGN KEY (album_id) REFERENCES albums(id), 344 344 - ); 345 345 - CREATE TABLE IF NOT EXISTS album_tracks ( 346 346 - id VARCHAR PRIMARY KEY, 347 347 - album_id VARCHAR NOT NULL, 348 348 - track_id VARCHAR NOT NULL, 349 349 - FOREIGN KEY (album_id) REFERENCES albums(id), 350 350 - FOREIGN KEY (track_id) REFERENCES tracks(id), 351 351 - ); 352 352 - CREATE TABLE IF NOT EXISTS scrobbles ( 353 353 - id VARCHAR PRIMARY KEY, 354 354 - user_id VARCHAR NOT NULL, 355 355 - track_id VARCHAR NOT NULL, 356 356 - album_id VARCHAR NOT NULL, 357 357 - artist_id VARCHAR NOT NULL, 358 358 - uri VARCHAR UNIQUE, 359 359 - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 360 360 - FOREIGN KEY (user_id) REFERENCES users(id), 361 361 - FOREIGN KEY (track_id) REFERENCES tracks(id), 362 362 - FOREIGN KEY (album_id) REFERENCES albums(id), 363 363 - FOREIGN KEY (artist_id) REFERENCES artists(id), 364 364 - ); 365 365 - COMMIT; 366 366 - ", 367 367 - )?; 368 368 - 369 369 - Ok(()) 370 370 - } 371 371 - }

-366

crates/feed/src/repo/duckdb/scrobble.rs

··· 1 1 - use std::{sync::Arc, thread}; 2 2 - 3 3 - use anyhow::Error; 4 4 - use duckdb::{params, OptionalExt}; 5 5 - use owo_colors::OwoColorize; 6 6 - use std::sync::Mutex; 7 7 - 8 8 - use crate::{did::did_to_profile, types::ScrobbleRecord}; 9 9 - 10 10 - pub fn save_scrobble( 11 11 - conn: Arc<Mutex<duckdb::Connection>>, 12 12 - did: &str, 13 13 - uri: &str, 14 14 - record: ScrobbleRecord, 15 15 - ) -> Result<(), Error> { 16 16 - let did = did.to_string(); 17 17 - let cloned_did = did.clone(); 18 18 - 19 19 - let uri = uri.to_string(); 20 20 - let conn = conn.lock().unwrap(); 21 21 - 22 22 - tracing::info!("Inserting scrobble for user: {}, scrobble: {}", did, uri); 23 23 - let mut user = conn.prepare("SELECT id FROM users WHERE did = ?")?; 24 24 - let user_id: Option<String> = user.query_row(params![did], |row| row.get(0)).optional()?; 25 25 - 26 26 - let did_clone = did.clone(); 27 27 - if user_id.is_none() { 28 28 - let handle = thread::spawn(move || { 29 29 - let rt = tokio::runtime::Runtime::new()?; 30 30 - let profile = rt.block_on(did_to_profile(&did_clone))?; 31 31 - 32 32 - let avatar = profile.avatar.map(|blob| { 33 33 - format!( 34 34 - "https://cdn.bsky.app/img/avatar/plain/{}/{}@{}", 35 35 - did_clone, 36 36 - blob.r#ref.link, 37 37 - blob.mime_type.split('/').last().unwrap_or("jpeg") 38 38 - ) 39 39 - }); 40 40 - Ok::<(Option<String>, Option<String>, Option<String>), Error>(( 41 41 - profile.display_name, 42 42 - profile.handle, 43 43 - avatar, 44 44 - )) 45 45 - }); 46 46 - 47 47 - let (display_name, handle, avatar) = handle 48 48 - .join() 49 49 - .map_err(|e| anyhow::anyhow!("Thread panicked: {:?}", e))??; 50 50 - 51 51 - conn.execute( 52 52 - "INSERT OR IGNORE INTO users ( 53 53 - id, 54 54 - display_name, 55 55 - did, 56 56 - handle, 57 57 - avatar 58 58 - ) VALUES ( 59 59 - ?, 60 60 - ?, 61 61 - ?, 62 62 - ?, 63 63 - ?)", 64 64 - params![ 65 65 - xid::new().to_string(), 66 66 - display_name.unwrap_or_default(), 67 67 - did, 68 68 - handle.unwrap_or_default(), 69 69 - avatar, 70 70 - ], 71 71 - )?; 72 72 - } 73 73 - 74 74 - let album_hash = 75 75 - sha256::digest(format!("{} - {}", record.album, record.album_artist).to_lowercase()); 76 76 - 77 77 - match conn.execute( 78 78 - "INSERT OR IGNORE INTO albums ( 79 79 - id, 80 80 - title, 81 81 - artist, 82 82 - release_date, 83 83 - album_art, 84 84 - year, 85 85 - uri, 86 86 - sha256 87 87 - ) VALUES ( 88 88 - ?, 89 89 - ?, 90 90 - ?, 91 91 - ?, 92 92 - ?, 93 93 - ?, 94 94 - ?, 95 95 - ? 96 96 - )", 97 97 - params![ 98 98 - xid::new().to_string(), 99 99 - record.album, 100 100 - record.album_artist, 101 101 - record.release_date, 102 102 - record.album_art_url, 103 103 - record.year, 104 104 - record.album_uri, 105 105 - album_hash, 106 106 - ], 107 107 - ) { 108 108 - Ok(x) => tracing::info!("Album inserted or already exists {}", x), 109 109 - Err(e) => tracing::error!(error = %e, "Error inserting album"), 110 110 - } 111 111 - 112 112 - let artist_hash = sha256::digest(record.album_artist.to_lowercase()); 113 113 - match conn.execute( 114 114 - &format!( 115 115 - "INSERT OR IGNORE INTO artists ( 116 116 - id, 117 117 - name, 118 118 - sha256, 119 119 - picture, 120 120 - uri, 121 121 - tags 122 122 - ) VALUES ( 123 123 - ?, 124 124 - ?, 125 125 - ?, 126 126 - ?, 127 127 - ?, 128 128 - [{}] 129 129 - )", 130 130 - record 131 131 - .tags 132 132 - .as_ref() 133 133 - .map(|tags| tags 134 134 - .iter() 135 135 - .map(|tag| tag.replace('\'', "''")) 136 136 - .map(|tag| format!("'{}'", tag)) 137 137 - .collect::<Vec<_>>() 138 138 - .join(", ")) 139 139 - .unwrap_or_default() 140 140 - ), 141 141 - params![ 142 142 - xid::new().to_string(), 143 143 - record.album_artist, 144 144 - artist_hash, 145 145 - record.artist_picture, 146 146 - record.artist_uri 147 147 - ], 148 148 - ) { 149 149 - Ok(x) => tracing::info!("Artist inserted or already exists {}", x), 150 150 - Err(e) => tracing::error!(error = %e, "Error inserting artist"), 151 151 - } 152 152 - 153 153 - let track_hash = sha256::digest( 154 154 - format!("{} - {} - {}", record.title, record.artist, record.album).to_lowercase(), 155 155 - ); 156 156 - match conn.execute( 157 157 - "INSERT OR IGNORE INTO tracks ( 158 158 - id, 159 159 - title, 160 160 - artist, 161 161 - album_artist, 162 162 - album_art, 163 163 - album, 164 164 - track_number, 165 165 - disc_number, 166 166 - spotify_link, 167 167 - tidal_link, 168 168 - youtube_link, 169 169 - apple_music_link, 170 170 - copyright_message, 171 171 - label, 172 172 - lyrics, 173 173 - composer, 174 174 - duration, 175 175 - mb_id, 176 176 - uri, 177 177 - album_uri, 178 178 - artist_uri, 179 179 - sha256 180 180 - ) VALUES ( 181 181 - ?, 182 182 - ?, 183 183 - ?, 184 184 - ?, 185 185 - ?, 186 186 - ?, 187 187 - ?, 188 188 - ?, 189 189 - ?, 190 190 - ?, 191 191 - ?, 192 192 - ?, 193 193 - ?, 194 194 - ?, 195 195 - ?, 196 196 - ?, 197 197 - ?, 198 198 - ?, 199 199 - ?, 200 200 - ?, 201 201 - ?, 202 202 - ? 203 203 - )", 204 204 - params![ 205 205 - xid::new().to_string(), 206 206 - record.title, 207 207 - record.artist, 208 208 - record.album_artist, 209 209 - record.album_art_url, 210 210 - record.album, 211 211 - record.track_number, 212 212 - record.disc_number, 213 213 - record.spotify_link, 214 214 - record.tidal_link, 215 215 - record.youtube_link, 216 216 - record.apple_music_link, 217 217 - record.copyright_message, 218 218 - record.label, 219 219 - record.lyrics, 220 220 - record.composer, 221 221 - record.duration, 222 222 - record.mbid, 223 223 - record.song_uri, 224 224 - record.album_uri, 225 225 - record.artist_uri, 226 226 - track_hash, 227 227 - ], 228 228 - ) { 229 229 - Ok(x) => tracing::info!("Track inserted or already exists {}", x), 230 230 - Err(e) => tracing::error!(error = %e, "Error inserting track"), 231 231 - } 232 232 - 233 233 - match conn.execute( 234 234 - "INSERT OR IGNORE INTO album_tracks ( 235 235 - id, 236 236 - album_id, 237 237 - track_id 238 238 - ) VALUES ( 239 239 - ?, 240 240 - (SELECT id FROM albums WHERE sha256 = ?), 241 241 - (SELECT id FROM tracks WHERE sha256 = ?), 242 242 - )", 243 243 - params![xid::new().to_string(), album_hash, track_hash], 244 244 - ) { 245 245 - Ok(x) => tracing::info!("Album-Track relation inserted or already exists {}", x), 246 246 - Err(e) => tracing::error!(error = %e, "Error inserting album-track relation"), 247 247 - } 248 248 - 249 249 - match conn.execute( 250 250 - "INSERT OR IGNORE INTO user_artists ( 251 251 - id, 252 252 - user_id, 253 253 - artist_id 254 254 - ) VALUES ( 255 255 - ?, 256 256 - (SELECT id FROM users WHERE did = ?), 257 257 - (SELECT id FROM artists WHERE sha256 = ?), 258 258 - )", 259 259 - params![xid::new().to_string(), cloned_did, artist_hash], 260 260 - ) { 261 261 - Ok(x) => tracing::info!("User-Artist relation inserted or already exists {}", x), 262 262 - Err(e) => tracing::error!(error = %e, "Error inserting user-artist relation"), 263 263 - } 264 264 - 265 265 - match conn.execute( 266 266 - "INSERT OR IGNORE INTO user_albums ( 267 267 - id, 268 268 - user_id, 269 269 - album_id 270 270 - ) VALUES ( 271 271 - ?, 272 272 - (SELECT id FROM users WHERE did = ?), 273 273 - (SELECT id FROM albums WHERE sha256 = ?), 274 274 - )", 275 275 - params![xid::new().to_string(), cloned_did, album_hash], 276 276 - ) { 277 277 - Ok(x) => tracing::info!("User-Album relation inserted or already exists {}", x), 278 278 - Err(e) => tracing::error!(error = %e, "Error inserting user-album relation"), 279 279 - } 280 280 - 281 281 - match conn.execute( 282 282 - "INSERT OR IGNORE INTO user_tracks ( 283 283 - id, 284 284 - user_id, 285 285 - track_id 286 286 - ) VALUES ( 287 287 - ?, 288 288 - (SELECT id FROM users WHERE did = ?), 289 289 - (SELECT id FROM tracks WHERE sha256 = ?), 290 290 - )", 291 291 - params![xid::new().to_string(), cloned_did, track_hash], 292 292 - ) { 293 293 - Ok(x) => tracing::info!("User-Track relation inserted or already exists {}", x), 294 294 - Err(e) => tracing::error!(error = %e, "Error inserting user-track relation"), 295 295 - } 296 296 - 297 297 - match conn.execute( 298 298 - "INSERT OR IGNORE INTO artist_albums ( 299 299 - id, 300 300 - artist_id, 301 301 - album_id 302 302 - ) VALUES ( 303 303 - ?, 304 304 - (SELECT id FROM artists WHERE sha256 = ?), 305 305 - (SELECT id FROM albums WHERE sha256 = ?), 306 306 - )", 307 307 - params![xid::new().to_string(), artist_hash, album_hash], 308 308 - ) { 309 309 - Ok(x) => tracing::info!("Artist-Album relation inserted or already exists {}", x), 310 310 - Err(e) => tracing::error!(error = %e, "Error inserting artist-album relation"), 311 311 - } 312 312 - 313 313 - match conn.execute( 314 314 - "INSERT OR IGNORE INTO artist_tracks ( 315 315 - id, 316 316 - artist_id, 317 317 - track_id 318 318 - ) VALUES ( 319 319 - ?, 320 320 - (SELECT id FROM artists WHERE sha256 = ?), 321 321 - (SELECT id FROM tracks WHERE sha256 = ?), 322 322 - )", 323 323 - params![xid::new().to_string(), artist_hash, track_hash], 324 324 - ) { 325 325 - Ok(x) => tracing::info!("Artist-Track relation inserted or already exists {}", x), 326 326 - Err(e) => tracing::error!(error = %e, "Error inserting artist-track relation"), 327 327 - } 328 328 - 329 329 - match conn.execute( 330 330 - "INSERT OR IGNORE INTO scrobbles ( 331 331 - id, 332 332 - user_id, 333 333 - track_id, 334 334 - album_id, 335 335 - artist_id, 336 336 - uri, 337 337 - created_at 338 338 - ) VALUES ( 339 339 - ?, 340 340 - (SELECT id FROM users WHERE did = ?), 341 341 - (SELECT id FROM tracks WHERE sha256 = ?), 342 342 - (SELECT id FROM albums WHERE sha256 = ?), 343 343 - (SELECT id FROM artists WHERE sha256 = ?), 344 344 - ?, 345 345 - ?, 346 346 - )", 347 347 - params![ 348 348 - xid::new().to_string(), 349 349 - cloned_did, 350 350 - track_hash, 351 351 - album_hash, 352 352 - artist_hash, 353 353 - uri, 354 354 - record.created_at, 355 355 - ], 356 356 - ) { 357 357 - Ok(x) => tracing::info!(title = %record.title.cyan(), "Scrobble inserted {}", x), 358 358 - Err(e) => { 359 359 - tracing::error!(title = %record.title.cyan(),error = %e, "Error inserting scrobble") 360 360 - } 361 361 - } 362 362 - 363 363 - tracing::info!("Scrobble insertion process completed for user: {}", did); 364 364 - 365 365 - Ok(()) 366 366 - }

-101

crates/feed/src/repo/duckdb/track.rs

··· 1 1 - use std::sync::{Arc, Mutex}; 2 2 - 3 3 - use crate::types::SongRecord; 4 4 - use anyhow::Error; 5 5 - use duckdb::params; 6 6 - 7 7 - pub fn save_track( 8 8 - conn: Arc<Mutex<duckdb::Connection>>, 9 9 - uri: &str, 10 10 - record: SongRecord, 11 11 - ) -> Result<(), Error> { 12 12 - let uri = uri.to_string(); 13 13 - let conn = conn.lock().unwrap(); 14 14 - let track_hash = sha256::digest( 15 15 - format!("{} - {} - {}", record.title, record.artist, record.album).to_lowercase(), 16 16 - ); 17 17 - 18 18 - match conn.execute( 19 19 - "UPDATE tracks SET uri = ? WHERE title = ? AND artist = ? AND album = ? AND uri IS NULL;", 20 20 - params![uri, record.title, record.artist, record.album], 21 21 - ) { 22 22 - Ok(x) => { 23 23 - tracing::info!("Track URI updated successfully: {}", x); 24 24 - return Ok(()); 25 25 - } 26 26 - Err(e) => tracing::error!(error = %e, "Error updating track URI"), 27 27 - } 28 28 - 29 29 - match conn.execute( 30 30 - "INSERT OR IGNORE INTO tracks ( 31 31 - id, 32 32 - title, 33 33 - artist, 34 34 - album_artist, 35 35 - album_art, 36 36 - album, 37 37 - track_number, 38 38 - disc_number, 39 39 - spotify_link, 40 40 - tidal_link, 41 41 - youtube_link, 42 42 - apple_music_link, 43 43 - copyright_message, 44 44 - label, 45 45 - lyrics, 46 46 - composer, 47 47 - duration, 48 48 - mb_id, 49 49 - sha256, 50 50 - uri 51 51 - ) VALUES ( 52 52 - ?, 53 53 - ?, 54 54 - ?, 55 55 - ?, 56 56 - ?, 57 57 - ?, 58 58 - ?, 59 59 - ?, 60 60 - ?, 61 61 - ?, 62 62 - ?, 63 63 - ?, 64 64 - ?, 65 65 - ?, 66 66 - ?, 67 67 - ?, 68 68 - ?, 69 69 - ?, 70 70 - ?, 71 71 - ? 72 72 - );", 73 73 - params![ 74 74 - xid::new().to_string(), 75 75 - record.title, 76 76 - record.artist, 77 77 - record.album_artist, 78 78 - record.album_art_url, 79 79 - record.album, 80 80 - record.track_number, 81 81 - record.disc_number, 82 82 - record.spotify_link, 83 83 - record.tidal_link, 84 84 - record.youtube_link, 85 85 - record.apple_music_link, 86 86 - record.copyright_message, 87 87 - record.label, 88 88 - record.lyrics, 89 89 - record.composer, 90 90 - record.duration, 91 91 - record.mbid, 92 92 - track_hash, 93 93 - uri 94 94 - ], 95 95 - ) { 96 96 - Ok(x) => tracing::info!("Track successfully inserted or updated: {}", x), 97 97 - Err(e) => tracing::error!(error = %e, "Error inserting/updating track"), 98 98 - } 99 99 - 100 100 - Ok(()) 101 101 - }

-5

crates/feed/src/repo/duckdb/user.rs

··· 1 1 - use std::sync::{Arc, Mutex}; 2 2 - 3 3 - pub fn save_user(_conn: Arc<Mutex<duckdb::Connection>>, _did: &str) -> Result<(), anyhow::Error> { 4 4 - todo!() 5 5 - }

-154

crates/feed/src/repo/mod.rs

··· 1 1 - use anyhow::Error; 2 2 - use async_trait::async_trait; 3 3 - 4 4 - use crate::{ 5 5 - repo::{duckdb::DuckdbRepo, postgres::PostgresRepo}, 6 6 - types::{AlbumRecord, ArtistRecord, ScrobbleRecord, SongRecord}, 7 7 - }; 8 8 - 9 9 - pub mod duckdb; 10 10 - pub mod postgres; 11 11 - 12 12 - #[async_trait] 13 13 - pub trait Repo: Send + Sync + Clone { 14 14 - async fn insert_album(self, uri: &str, record: AlbumRecord) -> Result<(), Error>; 15 15 - async fn insert_artist(self, uri: &str, record: ArtistRecord) -> Result<(), Error>; 16 16 - async fn insert_scrobble( 17 17 - self, 18 18 - did: &str, 19 19 - uri: &str, 20 20 - record: ScrobbleRecord, 21 21 - ) -> Result<(), Error>; 22 22 - async fn insert_track(self, uri: &str, record: SongRecord) -> Result<(), Error>; 23 23 - async fn insert_user(self, did: &str) -> Result<(), Error>; 24 24 - async fn get_albums(self) -> Result<(), Error>; 25 25 - async fn get_artists(self) -> Result<(), Error>; 26 26 - async fn get_scrobbles(self) -> Result<(), Error>; 27 27 - async fn get_tracks(self) -> Result<(), Error>; 28 28 - async fn get_users(self) -> Result<(), Error>; 29 29 - async fn get_album(self) -> Result<(), Error>; 30 30 - async fn get_artist(self) -> Result<(), Error>; 31 31 - async fn get_track(self) -> Result<(), Error>; 32 32 - async fn get_user(self) -> Result<(), Error>; 33 33 - async fn create_tables(self) -> Result<(), Error>; 34 34 - } 35 35 - 36 36 - pub enum RepoImpl { 37 37 - Duckdb(DuckdbRepo), 38 38 - Postgres(PostgresRepo), 39 39 - } 40 40 - 41 41 - impl Clone for RepoImpl { 42 42 - fn clone(&self) -> Self { 43 43 - match self { 44 44 - RepoImpl::Duckdb(repo) => RepoImpl::Duckdb(repo.clone()), 45 45 - RepoImpl::Postgres(repo) => RepoImpl::Postgres(repo.clone()), 46 46 - } 47 47 - } 48 48 - } 49 49 - 50 50 - #[async_trait] 51 51 - impl Repo for RepoImpl { 52 52 - async fn insert_album(self, uri: &str, record: AlbumRecord) -> Result<(), Error> { 53 53 - match self { 54 54 - RepoImpl::Duckdb(repo) => repo.insert_album(uri, record).await, 55 55 - RepoImpl::Postgres(repo) => repo.insert_album(uri, record).await, 56 56 - } 57 57 - } 58 58 - 59 59 - async fn insert_artist(self, uri: &str, record: ArtistRecord) -> Result<(), Error> { 60 60 - match self { 61 61 - RepoImpl::Duckdb(repo) => repo.insert_artist(uri, record).await, 62 62 - RepoImpl::Postgres(repo) => repo.insert_artist(uri, record).await, 63 63 - } 64 64 - } 65 65 - 66 66 - async fn insert_scrobble( 67 67 - self, 68 68 - did: &str, 69 69 - uri: &str, 70 70 - record: ScrobbleRecord, 71 71 - ) -> Result<(), Error> { 72 72 - match self { 73 73 - RepoImpl::Duckdb(repo) => repo.insert_scrobble(did, uri, record).await, 74 74 - RepoImpl::Postgres(repo) => repo.insert_scrobble(did, uri, record).await, 75 75 - } 76 76 - } 77 77 - 78 78 - async fn insert_track(self, uri: &str, record: SongRecord) -> Result<(), Error> { 79 79 - match self { 80 80 - RepoImpl::Duckdb(repo) => repo.insert_track(uri, record).await, 81 81 - RepoImpl::Postgres(repo) => repo.insert_track(uri, record).await, 82 82 - } 83 83 - } 84 84 - 85 85 - async fn insert_user(self, did: &str) -> Result<(), Error> { 86 86 - match self { 87 87 - RepoImpl::Duckdb(repo) => repo.insert_user(did).await, 88 88 - RepoImpl::Postgres(repo) => repo.insert_user(did).await, 89 89 - } 90 90 - } 91 91 - 92 92 - async fn get_albums(self) -> Result<(), Error> { 93 93 - match self { 94 94 - RepoImpl::Duckdb(repo) => repo.get_albums().await, 95 95 - RepoImpl::Postgres(repo) => repo.get_albums().await, 96 96 - } 97 97 - } 98 98 - 99 99 - async fn get_artists(self) -> Result<(), Error> { 100 100 - match self { 101 101 - RepoImpl::Duckdb(repo) => repo.get_artists().await, 102 102 - RepoImpl::Postgres(repo) => repo.get_artists().await, 103 103 - } 104 104 - } 105 105 - 106 106 - async fn get_scrobbles(self) -> Result<(), Error> { 107 107 - match self { 108 108 - RepoImpl::Duckdb(repo) => repo.get_scrobbles().await, 109 109 - RepoImpl::Postgres(repo) => repo.get_scrobbles().await, 110 110 - } 111 111 - } 112 112 - async fn get_tracks(self) -> Result<(), Error> { 113 113 - match self { 114 114 - RepoImpl::Duckdb(repo) => repo.get_tracks().await, 115 115 - RepoImpl::Postgres(repo) => repo.get_tracks().await, 116 116 - } 117 117 - } 118 118 - async fn get_users(self) -> Result<(), Error> { 119 119 - match self { 120 120 - RepoImpl::Duckdb(repo) => repo.get_users().await, 121 121 - RepoImpl::Postgres(repo) => repo.get_users().await, 122 122 - } 123 123 - } 124 124 - async fn get_album(self) -> Result<(), Error> { 125 125 - match self { 126 126 - RepoImpl::Duckdb(repo) => repo.get_album().await, 127 127 - RepoImpl::Postgres(repo) => repo.get_album().await, 128 128 - } 129 129 - } 130 130 - async fn get_artist(self) -> Result<(), Error> { 131 131 - match self { 132 132 - RepoImpl::Duckdb(repo) => repo.get_artist().await, 133 133 - RepoImpl::Postgres(repo) => repo.get_artist().await, 134 134 - } 135 135 - } 136 136 - async fn get_track(self) -> Result<(), Error> { 137 137 - match self { 138 138 - RepoImpl::Duckdb(repo) => repo.get_track().await, 139 139 - RepoImpl::Postgres(repo) => repo.get_track().await, 140 140 - } 141 141 - } 142 142 - async fn get_user(self) -> Result<(), Error> { 143 143 - match self { 144 144 - RepoImpl::Duckdb(repo) => repo.get_user().await, 145 145 - RepoImpl::Postgres(repo) => repo.get_user().await, 146 146 - } 147 147 - } 148 148 - async fn create_tables(self) -> Result<(), Error> { 149 149 - match self { 150 150 - RepoImpl::Duckdb(repo) => repo.create_tables().await, 151 151 - RepoImpl::Postgres(repo) => repo.create_tables().await, 152 152 - } 153 153 - } 154 154 - }

-25

crates/feed/src/repo/postgres/album.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use sqlx::{Pool, Postgres}; 4 4 - use tokio::sync::Mutex; 5 5 - 6 6 - use crate::types::AlbumRecord; 7 7 - 8 8 - #[derive(Clone)] 9 9 - pub struct AlbumRepo { 10 10 - pub pool: Arc<Mutex<Pool<Postgres>>>, 11 11 - } 12 12 - 13 13 - impl AlbumRepo { 14 14 - pub fn new(pool: Arc<Mutex<Pool<Postgres>>>) -> Self { 15 15 - Self { pool } 16 16 - } 17 17 - 18 18 - pub async fn save_album(&self, _uri: &str, _record: AlbumRecord) -> Result<(), anyhow::Error> { 19 19 - todo!() 20 20 - } 21 21 - 22 22 - pub async fn get_albums(&self) -> Result<Vec<AlbumRecord>, anyhow::Error> { 23 23 - todo!() 24 24 - } 25 25 - }

-29

crates/feed/src/repo/postgres/artist.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use sqlx::{Pool, Postgres}; 4 4 - use tokio::sync::Mutex; 5 5 - 6 6 - use crate::{types::ArtistRecord, xata::artist::Artist}; 7 7 - 8 8 - #[derive(Clone)] 9 9 - pub struct ArtistRepo { 10 10 - pub pool: Arc<Mutex<Pool<Postgres>>>, 11 11 - } 12 12 - 13 13 - impl ArtistRepo { 14 14 - pub fn new(pool: Arc<Mutex<Pool<Postgres>>>) -> Self { 15 15 - Self { pool } 16 16 - } 17 17 - 18 18 - pub async fn save_artist( 19 19 - &self, 20 20 - _uri: &str, 21 21 - _record: ArtistRecord, 22 22 - ) -> Result<(), anyhow::Error> { 23 23 - todo!() 24 24 - } 25 25 - 26 26 - pub async fn get_artists(&self) -> Result<Vec<Artist>, anyhow::Error> { 27 27 - todo!() 28 28 - } 29 29 - }

-116

crates/feed/src/repo/postgres/mod.rs

··· 1 1 - use std::{env, sync::Arc}; 2 2 - 3 3 - use anyhow::Error; 4 4 - use async_trait::async_trait; 5 5 - use sqlx::postgres::PgPoolOptions; 6 6 - use tokio::sync::Mutex; 7 7 - 8 8 - use crate::{ 9 9 - repo::postgres::{ 10 10 - album::AlbumRepo, artist::ArtistRepo, scrobble::ScrobbleRepo, track::TrackRepo, 11 11 - user::UserRepo, 12 12 - }, 13 13 - types::{AlbumRecord, ArtistRecord, ScrobbleRecord, SongRecord}, 14 14 - }; 15 15 - 16 16 - use super::Repo; 17 17 - 18 18 - pub mod album; 19 19 - pub mod artist; 20 20 - pub mod scrobble; 21 21 - pub mod track; 22 22 - pub mod user; 23 23 - 24 24 - #[derive(Clone)] 25 25 - pub struct PostgresRepo { 26 26 - pub album: AlbumRepo, 27 27 - pub artist: ArtistRepo, 28 28 - pub scrobble: ScrobbleRepo, 29 29 - pub track: TrackRepo, 30 30 - pub user: UserRepo, 31 31 - } 32 32 - 33 33 - impl PostgresRepo { 34 34 - pub async fn new() -> Result<Self, Error> { 35 35 - let pool = PgPoolOptions::new() 36 36 - .max_connections(5) 37 37 - .connect(&env::var("XATA_POSTGRES_URL")?) 38 38 - .await?; 39 39 - let pool = Arc::new(Mutex::new(pool)); 40 40 - Ok(Self { 41 41 - album: AlbumRepo::new(pool.clone()), 42 42 - artist: ArtistRepo::new(pool.clone()), 43 43 - scrobble: ScrobbleRepo::new(pool.clone()), 44 44 - track: TrackRepo::new(pool.clone()), 45 45 - user: UserRepo::new(pool.clone()), 46 46 - }) 47 47 - } 48 48 - } 49 49 - 50 50 - #[async_trait] 51 51 - impl Repo for PostgresRepo { 52 52 - async fn insert_album(self, uri: &str, record: AlbumRecord) -> Result<(), anyhow::Error> { 53 53 - self.album.save_album(uri, record).await 54 54 - } 55 55 - 56 56 - async fn insert_artist(self, uri: &str, record: ArtistRecord) -> Result<(), anyhow::Error> { 57 57 - self.artist.save_artist(uri, record).await 58 58 - } 59 59 - 60 60 - async fn insert_scrobble( 61 61 - self, 62 62 - did: &str, 63 63 - uri: &str, 64 64 - record: ScrobbleRecord, 65 65 - ) -> Result<(), anyhow::Error> { 66 66 - self.scrobble.save_scrobble(did, uri, record).await 67 67 - } 68 68 - 69 69 - async fn insert_track(self, uri: &str, record: SongRecord) -> Result<(), anyhow::Error> { 70 70 - self.track.save_track(uri, record).await 71 71 - } 72 72 - 73 73 - async fn insert_user(self, did: &str) -> Result<(), anyhow::Error> { 74 74 - self.user.save_user(did).await 75 75 - } 76 76 - 77 77 - async fn get_albums(self) -> Result<(), anyhow::Error> { 78 78 - todo!() 79 79 - } 80 80 - 81 81 - async fn get_artists(self) -> Result<(), anyhow::Error> { 82 82 - todo!() 83 83 - } 84 84 - 85 85 - async fn get_scrobbles(self) -> Result<(), anyhow::Error> { 86 86 - todo!() 87 87 - } 88 88 - 89 89 - async fn get_tracks(self) -> Result<(), anyhow::Error> { 90 90 - todo!() 91 91 - } 92 92 - 93 93 - async fn get_users(self) -> Result<(), anyhow::Error> { 94 94 - todo!() 95 95 - } 96 96 - 97 97 - async fn get_album(self) -> Result<(), anyhow::Error> { 98 98 - todo!() 99 99 - } 100 100 - 101 101 - async fn get_artist(self) -> Result<(), anyhow::Error> { 102 102 - todo!() 103 103 - } 104 104 - 105 105 - async fn get_track(self) -> Result<(), anyhow::Error> { 106 106 - todo!() 107 107 - } 108 108 - 109 109 - async fn get_user(self) -> Result<(), anyhow::Error> { 110 110 - todo!() 111 111 - } 112 112 - 113 113 - async fn create_tables(self) -> Result<(), Error> { 114 114 - todo!() 115 115 - } 116 116 - }

-30

crates/feed/src/repo/postgres/scrobble.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use sqlx::{Pool, Postgres}; 4 4 - use tokio::sync::Mutex; 5 5 - 6 6 - use crate::{types::ScrobbleRecord, xata::scrobble::Scrobble}; 7 7 - 8 8 - #[derive(Clone)] 9 9 - pub struct ScrobbleRepo { 10 10 - pub pool: Arc<Mutex<Pool<Postgres>>>, 11 11 - } 12 12 - 13 13 - impl ScrobbleRepo { 14 14 - pub fn new(pool: Arc<Mutex<Pool<Postgres>>>) -> Self { 15 15 - Self { pool } 16 16 - } 17 17 - 18 18 - pub async fn save_scrobble( 19 19 - &self, 20 20 - _did: &str, 21 21 - _uri: &str, 22 22 - _record: ScrobbleRecord, 23 23 - ) -> Result<(), anyhow::Error> { 24 24 - todo!() 25 25 - } 26 26 - 27 27 - pub async fn get_scrobbles(&self) -> Result<Vec<Scrobble>, anyhow::Error> { 28 28 - todo!() 29 29 - } 30 30 - }

-21

crates/feed/src/repo/postgres/track.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use sqlx::{Pool, Postgres}; 4 4 - use tokio::sync::Mutex; 5 5 - 6 6 - use crate::types::SongRecord; 7 7 - 8 8 - #[derive(Clone)] 9 9 - pub struct TrackRepo { 10 10 - pub pool: Arc<Mutex<Pool<Postgres>>>, 11 11 - } 12 12 - 13 13 - impl TrackRepo { 14 14 - pub fn new(pool: Arc<Mutex<Pool<Postgres>>>) -> Self { 15 15 - Self { pool } 16 16 - } 17 17 - 18 18 - pub async fn save_track(&self, _uri: &str, _record: SongRecord) -> Result<(), anyhow::Error> { 19 19 - todo!() 20 20 - } 21 21 - }

-19

crates/feed/src/repo/postgres/user.rs

··· 1 1 - use std::sync::Arc; 2 2 - 3 3 - use sqlx::{Pool, Postgres}; 4 4 - use tokio::sync::Mutex; 5 5 - 6 6 - #[derive(Clone)] 7 7 - pub struct UserRepo { 8 8 - pub pool: Arc<Mutex<Pool<Postgres>>>, 9 9 - } 10 10 - 11 11 - impl UserRepo { 12 12 - pub fn new(pool: Arc<Mutex<Pool<Postgres>>>) -> Self { 13 13 - Self { pool } 14 14 - } 15 15 - 16 16 - pub async fn save_user(&self, _did: &str) -> Result<(), anyhow::Error> { 17 17 - todo!() 18 18 - } 19 19 - }

-104

crates/feed/src/subscriber.rs

··· 1 1 - use anyhow::Error; 2 2 - use owo_colors::OwoColorize; 3 3 - use tokio_stream::StreamExt; 4 4 - use tokio_tungstenite::connect_async; 5 5 - use tungstenite::Message; 6 6 - 7 7 - use crate::{ 8 8 - repo::{Repo, RepoImpl}, 9 9 - types::{AlbumRecord, ArtistRecord, Commit, Root, ScrobbleRecord, SongRecord}, 10 10 - ALBUM_NSID, ARTIST_NSID, SCROBBLE_NSID, SONG_NSID, 11 11 - }; 12 12 - 13 13 - pub struct ScrobbleSubscriber { 14 14 - pub service_url: String, 15 15 - } 16 16 - 17 17 - impl ScrobbleSubscriber { 18 18 - pub fn new(service: &str) -> Self { 19 19 - Self { 20 20 - service_url: service.to_string(), 21 21 - } 22 22 - } 23 23 - 24 24 - pub async fn run(&self, repo: RepoImpl) -> Result<(), Error> { 25 25 - let (mut ws_stream, _) = connect_async(&self.service_url).await?; 26 26 - tracing::info!(url = %self.service_url.bright_green(), "Connected to jetstream at"); 27 27 - 28 28 - while let Some(msg) = ws_stream.next().await { 29 29 - match msg { 30 30 - Ok(msg) => { 31 31 - if let Err(e) = handle_message(&repo, msg) { 32 32 - tracing::error!(error = %e, "Error handling message"); 33 33 - } 34 34 - } 35 35 - Err(e) => { 36 36 - tracing::error!(error = %e, "WebSocket error"); 37 37 - break; 38 38 - } 39 39 - } 40 40 - } 41 41 - 42 42 - Ok(()) 43 43 - } 44 44 - } 45 45 - 46 46 - fn handle_message(repo: &RepoImpl, msg: Message) -> Result<(), Error> { 47 47 - let repo = repo.clone(); 48 48 - tokio::spawn(async move { 49 49 - if let Message::Text(text) = msg { 50 50 - let message: Root = serde_json::from_str(&text)?; 51 51 - 52 52 - if message.kind != "commit" { 53 53 - return Ok::<(), Error>(()); 54 54 - } 55 55 - 56 56 - tracing::info!(message = %text, "Received message"); 57 57 - 58 58 - if let Some(commit) = message.commit { 59 59 - match commit.operation.as_str() { 60 60 - "create" => save_scrobble(repo, &message.did, commit).await?, 61 61 - _ => tracing::warn!(operation = %commit.operation, "Unknown operation"), 62 62 - } 63 63 - } 64 64 - } 65 65 - Ok::<(), Error>(()) 66 66 - }); 67 67 - Ok(()) 68 68 - } 69 69 - 70 70 - async fn save_scrobble(repo: RepoImpl, did: &str, commit: Commit) -> Result<(), Error> { 71 71 - if !vec![SCROBBLE_NSID, ARTIST_NSID, ALBUM_NSID, SONG_NSID] 72 72 - .contains(&commit.collection.as_str()) 73 73 - { 74 74 - return Ok(()); 75 75 - } 76 76 - 77 77 - match commit.collection.as_str() { 78 78 - SCROBBLE_NSID => { 79 79 - let record = serde_json::from_value::<ScrobbleRecord>(commit.record)?; 80 80 - let uri = format!("at://{}/app.rocksky.scrobble/{}", did, commit.rkey); 81 81 - repo.insert_scrobble(did, &uri, record.clone()).await?; 82 82 - } 83 83 - ARTIST_NSID => { 84 84 - let record = serde_json::from_value::<ArtistRecord>(commit.record)?; 85 85 - let uri = format!("at://{}/app.rocksky.artist/{}", did, commit.rkey); 86 86 - repo.insert_artist(&uri, record).await?; 87 87 - } 88 88 - ALBUM_NSID => { 89 89 - let record = serde_json::from_value::<AlbumRecord>(commit.record)?; 90 90 - let uri = format!("at://{}/app.rocksky.album/{}", did, commit.rkey); 91 91 - repo.insert_album(&uri, record).await?; 92 92 - } 93 93 - SONG_NSID => { 94 94 - let record = serde_json::from_value::<SongRecord>(commit.record)?; 95 95 - let uri = format!("at://{}/app.rocksky.song/{}", did, commit.rkey); 96 96 - repo.insert_track(&uri, record).await?; 97 97 - } 98 98 - _ => { 99 99 - tracing::warn!(collection = %commit.collection, "Unknown collection"); 100 100 - } 101 101 - } 102 102 - 103 103 - Ok(()) 104 104 - }

-147

crates/feed/src/sync.rs

··· 1 1 - use std::env; 2 2 - 3 3 - use anyhow::Error; 4 4 - use owo_colors::OwoColorize; 5 5 - use sqlx::postgres::PgPoolOptions; 6 6 - use sqlx::Row; 7 7 - 8 8 - use crate::repo::{Repo, RepoImpl}; 9 9 - use crate::types::ScrobbleRecord; 10 10 - 11 11 - pub async fn sync_scrobbles(ddb: RepoImpl) -> Result<(), Error> { 12 12 - tracing::info!("Starting scrobble synchronization..."); 13 13 - 14 14 - let repo = ddb.clone(); 15 15 - repo.create_tables().await?; 16 16 - 17 17 - let pool = PgPoolOptions::new() 18 18 - .max_connections(10) 19 19 - .connect(&env::var("XATA_POSTGRES_URL")?) 20 20 - .await?; 21 21 - 22 22 - let pool_clone = pool.clone(); 23 23 - 24 24 - const BATCH_SIZE: i64 = 1000; 25 25 - 26 26 - let total_scrobbles: (i64,) = 27 27 - sqlx::query_as("SELECT COUNT(*) FROM scrobbles WHERE uri IS NOT NULL") 28 28 - .fetch_one(&pool_clone) 29 29 - .await?; 30 30 - let total_scrobbles = total_scrobbles.0; 31 31 - tracing::info!(total = %total_scrobbles.magenta(), "Total scrobbles to sync"); 32 32 - 33 33 - let start = env::var("SYNC_START_OFFSET") 34 34 - .ok() 35 35 - .and_then(|s| s.parse::<i64>().ok()) 36 36 - .unwrap_or(0); 37 37 - 38 38 - let mut i = 1; 39 39 - for offset in (start..total_scrobbles).step_by(BATCH_SIZE as usize) { 40 40 - tracing::info!( 41 41 - offset = %(offset).magenta(), 42 42 - end = %(offset + BATCH_SIZE).magenta(), 43 43 - "Syncing scrobbles batch:", 44 44 - ); 45 45 - let result = sqlx::query( 46 46 - r#" 47 47 - SELECT 48 48 - s.xata_id, 49 49 - s.user_id, 50 50 - s.track_id, 51 51 - s.album_id, 52 52 - s.artist_id, 53 53 - s.uri, 54 54 - s.timestamp, 55 55 - a.*, 56 56 - ar.*, 57 57 - t.*, 58 58 - u.*, 59 59 - a.uri AS album_uri, 60 60 - ar.uri AS artist_uri, 61 61 - t.uri AS track_uri, 62 62 - a.title AS album_title, 63 63 - a.youtube_link AS album_youtube_link, 64 64 - a.spotify_link AS album_spotify_link, 65 65 - a.tidal_link AS album_tidal_link, 66 66 - a.apple_music_link AS album_apple_music_link 67 67 - FROM scrobbles s 68 68 - LEFT JOIN albums a ON s.album_id = a.xata_id 69 69 - LEFT JOIN artists ar ON s.artist_id = ar.xata_id 70 70 - LEFT JOIN tracks t ON s.track_id = t.xata_id 71 71 - LEFT JOIN users u ON s.user_id = u.xata_id 72 72 - WHERE s.uri IS NOT NULL 73 73 - ORDER BY s.timestamp DESC 74 74 - LIMIT $1 OFFSET $2 75 75 - "#, 76 76 - ) 77 77 - .bind(BATCH_SIZE) 78 78 - .bind(offset as i64) 79 79 - .fetch_all(&pool_clone) 80 80 - .await?; 81 81 - 82 82 - for row in result { 83 83 - tracing::info!( 84 84 - count = %i.magenta(), 85 85 - total = %total_scrobbles.magenta(), 86 86 - title = %row.get::<String, _>("title").cyan(), 87 87 - did = %row.get::<String, _>("did"), 88 88 - "Inserting scrobble..." 89 89 - ); 90 90 - 91 91 - let scrobble_uri = row.get::<Option<String>, _>("uri"); 92 92 - if scrobble_uri.is_none() { 93 93 - tracing::warn!(count = %i.magenta(), "Skipping scrobble with no URI"); 94 94 - i += 1; 95 95 - continue; 96 96 - } 97 97 - let scrobble_uri = scrobble_uri.unwrap(); 98 98 - 99 99 - let did = row.get::<String, _>("did"); 100 100 - let record: ScrobbleRecord = ScrobbleRecord { 101 101 - track_number: row.get::<Option<i32>, _>("track_number"), 102 102 - disc_number: row.get::<Option<i32>, _>("disc_number"), 103 103 - title: row.get::<String, _>("title"), 104 104 - artist: row.get::<String, _>("artist"), 105 105 - album_artist: row.get::<String, _>("album_artist"), 106 106 - album_art_url: row.get::<Option<String>, _>("album_art"), 107 107 - album: row.get::<String, _>("album"), 108 108 - duration: row.get::<i32, _>("duration"), 109 109 - release_date: row.get::<Option<String>, _>("release_date"), 110 110 - year: row.get::<Option<i32>, _>("year"), 111 111 - genre: row.get::<Option<String>, _>("genre"), 112 112 - tags: row.get::<Option<Vec<String>>, _>("genres"), 113 113 - composer: row.get::<Option<String>, _>("composer"), 114 114 - lyrics: row.get::<Option<String>, _>("lyrics"), 115 115 - copyright_message: row.get::<Option<String>, _>("copyright_message"), 116 116 - wiki: None, 117 117 - youtube_link: row.get::<Option<String>, _>("youtube_link"), 118 118 - spotify_link: row.get::<Option<String>, _>("spotify_link"), 119 119 - tidal_link: row.get::<Option<String>, _>("tidal_link"), 120 120 - apple_music_link: row.get::<Option<String>, _>("apple_music_link"), 121 121 - created_at: row 122 122 - .get::<chrono::DateTime<chrono::Utc>, _>("timestamp") 123 123 - .to_rfc3339(), 124 124 - label: row.get::<Option<String>, _>("label"), 125 125 - mbid: row.get::<Option<String>, _>("mb_id"), 126 126 - artist_picture: row.get::<Option<String>, _>("picture"), 127 127 - artist_uri: row.get::<Option<String>, _>("artist_uri"), 128 128 - album_uri: row.get::<Option<String>, _>("album_uri"), 129 129 - song_uri: row.get::<Option<String>, _>("track_uri"), 130 130 - }; 131 131 - 132 132 - let repo = ddb.clone(); 133 133 - match repo.insert_scrobble(&did, &scrobble_uri, record).await { 134 134 - Ok(_) => { 135 135 - tracing::info!(count = %i.magenta(), "Scrobble inserted successfully") 136 136 - } 137 137 - Err(e) => { 138 138 - tracing::error!(error = %e, "Error inserting scrobble"); 139 139 - } 140 140 - } 141 141 - 142 142 - i += 1; 143 143 - } 144 144 - } 145 145 - 146 146 - Ok(()) 147 147 - }

-267

crates/feed/src/types.rs

··· 1 1 - use serde::{Deserialize, Serialize}; 2 2 - use serde_json::Value; 3 3 - 4 4 - #[derive(Debug, Clone)] 5 5 - pub struct Request { 6 6 - pub cursor: Option<String>, 7 7 - pub feed: String, 8 8 - pub limit: Option<u8>, 9 9 - } 10 10 - 11 11 - #[derive(Debug, Clone)] 12 12 - pub struct Cid(pub String); 13 13 - 14 14 - #[derive(Debug, Clone)] 15 15 - pub struct Did(pub String); 16 16 - 17 17 - #[derive(Debug, Clone, Hash, PartialEq, Eq)] 18 18 - pub struct Uri(pub String); 19 19 - 20 20 - #[derive(Debug, Clone)] 21 21 - pub struct FeedResult { 22 22 - pub cursor: Option<String>, 23 23 - pub feed: Vec<Uri>, 24 24 - } 25 25 - 26 26 - #[derive(Debug, Clone)] 27 27 - pub struct Scrobble {} 28 28 - 29 29 - #[derive(Serialize)] 30 30 - pub(crate) struct DidDocument { 31 31 - #[serde(rename = "@context")] 32 32 - pub(crate) context: Vec<String>, 33 33 - pub(crate) id: String, 34 34 - pub(crate) service: Vec<Service>, 35 35 - } 36 36 - 37 37 - #[derive(Serialize)] 38 38 - pub(crate) struct Service { 39 39 - pub(crate) id: String, 40 40 - #[serde(rename = "type")] 41 41 - pub(crate) type_: String, 42 42 - #[serde(rename = "serviceEndpoint")] 43 43 - pub(crate) service_endpoint: String, 44 44 - } 45 45 - 46 46 - #[derive(Debug, Deserialize)] 47 47 - pub struct Commit { 48 48 - pub rev: String, 49 49 - pub operation: String, 50 50 - pub collection: String, 51 51 - pub rkey: String, 52 52 - pub record: Value, 53 53 - pub cid: String, 54 54 - } 55 55 - 56 56 - #[derive(Debug, Deserialize)] 57 57 - pub struct Root { 58 58 - pub did: String, 59 59 - pub time_us: i64, 60 60 - pub kind: String, 61 61 - pub commit: Option<Commit>, 62 62 - } 63 63 - 64 64 - #[derive(Serialize, Deserialize)] 65 65 - pub struct SkeletonFeedScrobbleData { 66 66 - #[serde(skip_serializing_if = "core::option::Option::is_none")] 67 67 - pub feed_context: core::option::Option<String>, 68 68 - pub scrobble: String, 69 69 - } 70 70 - 71 71 - #[derive(Serialize, Deserialize)] 72 72 - pub struct FeedSkeleton { 73 73 - pub cursor: Option<String>, 74 74 - pub feed: Vec<SkeletonFeedScrobbleData>, 75 75 - } 76 76 - 77 77 - // 78 78 - // Jetstream types 79 79 - // 80 80 - 81 81 - #[derive(Debug, Deserialize, Clone)] 82 82 - #[serde(rename_all = "camelCase")] 83 83 - pub struct ScrobbleRecord { 84 84 - #[serde(skip_serializing_if = "Option::is_none")] 85 85 - pub track_number: Option<i32>, 86 86 - #[serde(skip_serializing_if = "Option::is_none")] 87 87 - pub disc_number: Option<i32>, 88 88 - pub title: String, 89 89 - pub artist: String, 90 90 - pub album_artist: String, 91 91 - pub album: String, 92 92 - pub duration: i32, 93 93 - #[serde(skip_serializing_if = "Option::is_none")] 94 94 - pub release_date: Option<String>, 95 95 - #[serde(skip_serializing_if = "Option::is_none")] 96 96 - pub year: Option<i32>, 97 97 - #[serde(skip_serializing_if = "Option::is_none")] 98 98 - pub genre: Option<String>, 99 99 - #[serde(skip_serializing_if = "Option::is_none")] 100 100 - pub tags: Option<Vec<String>>, 101 101 - #[serde(skip_serializing_if = "Option::is_none")] 102 102 - pub composer: Option<String>, 103 103 - #[serde(skip_serializing_if = "Option::is_none")] 104 104 - pub lyrics: Option<String>, 105 105 - #[serde(skip_serializing_if = "Option::is_none")] 106 106 - pub copyright_message: Option<String>, 107 107 - #[serde(skip_serializing_if = "Option::is_none")] 108 108 - pub wiki: Option<String>, 109 109 - #[serde(skip_serializing_if = "Option::is_none")] 110 110 - pub album_art_url: Option<String>, 111 111 - #[serde(skip_serializing_if = "Option::is_none")] 112 112 - pub youtube_link: Option<String>, 113 113 - #[serde(skip_serializing_if = "Option::is_none")] 114 114 - pub spotify_link: Option<String>, 115 115 - #[serde(skip_serializing_if = "Option::is_none")] 116 116 - pub tidal_link: Option<String>, 117 117 - #[serde(skip_serializing_if = "Option::is_none")] 118 118 - pub apple_music_link: Option<String>, 119 119 - pub created_at: String, 120 120 - #[serde(skip_serializing_if = "Option::is_none")] 121 121 - pub label: Option<String>, 122 122 - #[serde(skip_serializing_if = "Option::is_none")] 123 123 - pub mbid: Option<String>, 124 124 - #[serde(skip_serializing_if = "Option::is_none")] 125 125 - pub artist_picture: Option<String>, 126 126 - #[serde(skip_serializing_if = "Option::is_none")] 127 127 - pub artist_uri: Option<String>, 128 128 - #[serde(skip_serializing_if = "Option::is_none")] 129 129 - pub album_uri: Option<String>, 130 130 - #[serde(skip_serializing_if = "Option::is_none")] 131 131 - pub song_uri: Option<String>, 132 132 - } 133 133 - 134 134 - #[derive(Debug, Deserialize, Clone)] 135 135 - #[serde(rename_all = "camelCase")] 136 136 - pub struct ArtistRecord { 137 137 - pub name: String, 138 138 - #[serde(skip_serializing_if = "Option::is_none")] 139 139 - pub bio: Option<String>, 140 140 - #[serde(skip_serializing_if = "Option::is_none")] 141 141 - pub picture_url: Option<String>, 142 142 - #[serde(skip_serializing_if = "Option::is_none")] 143 143 - pub tags: Option<Vec<String>>, 144 144 - #[serde(skip_serializing_if = "Option::is_none")] 145 145 - pub born: Option<String>, 146 146 - #[serde(skip_serializing_if = "Option::is_none")] 147 147 - pub died: Option<String>, 148 148 - #[serde(skip_serializing_if = "Option::is_none")] 149 149 - pub born_in: Option<String>, 150 150 - pub created_at: String, 151 151 - } 152 152 - 153 153 - #[derive(Debug, Deserialize, Clone)] 154 154 - #[serde(rename_all = "camelCase")] 155 155 - pub struct AlbumRecord { 156 156 - pub title: String, 157 157 - pub artist: String, 158 158 - #[serde(skip_serializing_if = "Option::is_none")] 159 159 - pub duration: Option<i32>, 160 160 - #[serde(skip_serializing_if = "Option::is_none")] 161 161 - pub release_date: Option<String>, 162 162 - #[serde(skip_serializing_if = "Option::is_none")] 163 163 - pub year: Option<i32>, 164 164 - #[serde(skip_serializing_if = "Option::is_none")] 165 165 - pub genre: Option<String>, 166 166 - #[serde(skip_serializing_if = "Option::is_none")] 167 167 - pub album_art_url: Option<String>, 168 168 - #[serde(skip_serializing_if = "Option::is_none")] 169 169 - pub tags: Option<Vec<String>>, 170 170 - #[serde(skip_serializing_if = "Option::is_none")] 171 171 - pub youtube_link: Option<String>, 172 172 - #[serde(skip_serializing_if = "Option::is_none")] 173 173 - pub spotify_link: Option<String>, 174 174 - #[serde(skip_serializing_if = "Option::is_none")] 175 175 - pub tidal_link: Option<String>, 176 176 - #[serde(skip_serializing_if = "Option::is_none")] 177 177 - pub apple_music_link: Option<String>, 178 178 - pub created_at: String, 179 179 - } 180 180 - 181 181 - #[derive(Debug, Deserialize, Clone)] 182 182 - #[serde(rename_all = "camelCase")] 183 183 - pub struct SongRecord { 184 184 - pub title: String, 185 185 - pub artist: String, 186 186 - pub album: String, 187 187 - pub album_artist: String, 188 188 - pub duration: i32, 189 189 - pub created_at: String, 190 190 - #[serde(skip_serializing_if = "Option::is_none")] 191 191 - pub track_number: Option<i32>, 192 192 - #[serde(skip_serializing_if = "Option::is_none")] 193 193 - pub disc_number: Option<i32>, 194 194 - #[serde(skip_serializing_if = "Option::is_none")] 195 195 - pub genre: Option<String>, 196 196 - #[serde(skip_serializing_if = "Option::is_none")] 197 197 - pub release_date: Option<String>, 198 198 - #[serde(skip_serializing_if = "Option::is_none")] 199 199 - pub year: Option<i32>, 200 200 - #[serde(skip_serializing_if = "Option::is_none")] 201 201 - pub tags: Option<Vec<String>>, 202 202 - #[serde(skip_serializing_if = "Option::is_none")] 203 203 - pub composer: Option<String>, 204 204 - #[serde(skip_serializing_if = "Option::is_none")] 205 205 - pub lyrics: Option<String>, 206 206 - #[serde(skip_serializing_if = "Option::is_none")] 207 207 - pub copyright_message: Option<String>, 208 208 - #[serde(skip_serializing_if = "Option::is_none")] 209 209 - pub wiki: Option<String>, 210 210 - #[serde(skip_serializing_if = "Option::is_none")] 211 211 - pub album_art_url: Option<String>, 212 212 - #[serde(skip_serializing_if = "Option::is_none")] 213 213 - pub youtube_link: Option<String>, 214 214 - #[serde(skip_serializing_if = "Option::is_none")] 215 215 - pub spotify_link: Option<String>, 216 216 - #[serde(skip_serializing_if = "Option::is_none")] 217 217 - pub tidal_link: Option<String>, 218 218 - #[serde(skip_serializing_if = "Option::is_none")] 219 219 - pub apple_music_link: Option<String>, 220 220 - #[serde(skip_serializing_if = "Option::is_none")] 221 221 - pub label: Option<String>, 222 222 - #[serde(skip_serializing_if = "Option::is_none")] 223 223 - pub mbid: Option<String>, 224 224 - } 225 225 - 226 226 - #[derive(Debug, Deserialize, Clone)] 227 227 - pub struct Ref { 228 228 - #[serde(rename = "$link")] 229 229 - pub link: String, 230 230 - } 231 231 - 232 232 - #[derive(Debug, Deserialize, Clone)] 233 233 - #[serde(rename_all = "camelCase")] 234 234 - pub struct Blob { 235 235 - #[serde(rename = "$type")] 236 236 - pub r#type: String, 237 237 - pub r#ref: Ref, 238 238 - pub mime_type: String, 239 239 - pub size: i32, 240 240 - } 241 241 - 242 242 - #[derive(Debug, Deserialize)] 243 243 - pub struct PinnedPost { 244 244 - pub cid: String, 245 245 - pub uri: String, 246 246 - } 247 247 - 248 248 - #[derive(Debug, Deserialize)] 249 249 - #[serde(rename_all = "camelCase")] 250 250 - pub struct Profile { 251 251 - #[serde(rename = "$type")] 252 252 - pub r#type: String, 253 253 - pub avatar: Option<Blob>, 254 254 - pub banner: Option<Blob>, 255 255 - pub created_at: Option<String>, 256 256 - pub pinned_post: Option<PinnedPost>, 257 257 - pub description: Option<String>, 258 258 - pub display_name: Option<String>, 259 259 - pub handle: Option<String>, 260 260 - } 261 261 - 262 262 - #[derive(Debug, Deserialize)] 263 263 - pub struct ProfileResponse { 264 264 - pub uri: String, 265 265 - pub cid: String, 266 266 - pub value: Profile, 267 267 - }

-21

crates/feed/src/xata/album.rs

··· 1 1 - use chrono::{DateTime, Utc}; 2 2 - use serde::Deserialize; 3 3 - 4 4 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 5 5 - pub struct Album { 6 6 - pub xata_id: String, 7 7 - pub title: String, 8 8 - pub artist: String, 9 9 - pub release_date: Option<String>, 10 10 - pub album_art: Option<String>, 11 11 - pub year: Option<i32>, 12 12 - pub spotify_link: Option<String>, 13 13 - pub tidal_link: Option<String>, 14 14 - pub youtube_link: Option<String>, 15 15 - pub apple_music_link: Option<String>, 16 16 - pub sha256: String, 17 17 - pub uri: Option<String>, 18 18 - pub artist_uri: Option<String>, 19 19 - #[serde(with = "chrono::serde::ts_seconds")] 20 20 - pub xata_createdat: DateTime<Utc>, 21 21 - }

-8

crates/feed/src/xata/album_track.rs

··· 1 1 - use serde::Deserialize; 2 2 - 3 3 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 4 4 - pub struct AlbumTrack { 5 5 - pub xata_id: String, 6 6 - pub album_id: String, 7 7 - pub track_id: String, 8 8 - }

-24

crates/feed/src/xata/artist.rs

··· 1 1 - use chrono::{DateTime, Utc}; 2 2 - use serde::Deserialize; 3 3 - 4 4 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 5 5 - pub struct Artist { 6 6 - pub xata_id: String, 7 7 - pub name: String, 8 8 - pub biography: Option<String>, 9 9 - #[serde(with = "chrono::serde::ts_seconds_option")] 10 10 - pub born: Option<DateTime<Utc>>, 11 11 - pub born_in: Option<String>, 12 12 - #[serde(with = "chrono::serde::ts_seconds_option")] 13 13 - pub died: Option<DateTime<Utc>>, 14 14 - pub picture: Option<String>, 15 15 - pub sha256: String, 16 16 - pub spotify_link: Option<String>, 17 17 - pub tidal_link: Option<String>, 18 18 - pub youtube_link: Option<String>, 19 19 - pub apple_music_link: Option<String>, 20 20 - pub uri: Option<String>, 21 21 - pub genres: Option<Vec<String>>, 22 22 - #[serde(with = "chrono::serde::ts_seconds")] 23 23 - pub xata_createdat: DateTime<Utc>, 24 24 - }

-10

crates/feed/src/xata/artist_track.rs

··· 1 1 - use serde::Deserialize; 2 2 - 3 3 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 4 4 - pub struct ArtistTrack { 5 5 - pub xata_id: String, 6 6 - pub artist_id: String, 7 7 - pub track_id: String, 8 8 - #[serde(with = "chrono::serde::ts_seconds")] 9 9 - pub xata_createdat: chrono::DateTime<chrono::Utc>, 10 10 - }

-7

crates/feed/src/xata/mod.rs

··· 1 1 - pub mod album; 2 2 - pub mod album_track; 3 3 - pub mod artist; 4 4 - pub mod artist_track; 5 5 - pub mod scrobble; 6 6 - pub mod track; 7 7 - pub mod user;

-16

crates/feed/src/xata/scrobble.rs

··· 1 1 - use chrono::{DateTime, Utc}; 2 2 - use serde::Deserialize; 3 3 - 4 4 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 5 5 - pub struct Scrobble { 6 6 - pub xata_id: String, 7 7 - pub user_id: String, 8 8 - pub track_id: String, 9 9 - pub album_id: Option<String>, 10 10 - pub artist_id: Option<String>, 11 11 - pub uri: Option<String>, 12 12 - #[serde(with = "chrono::serde::ts_seconds")] 13 13 - pub xata_createdat: DateTime<Utc>, 14 14 - #[serde(with = "chrono::serde::ts_seconds")] 15 15 - pub timestamp: DateTime<Utc>, 16 16 - }

-31

crates/feed/src/xata/track.rs

··· 1 1 - use chrono::{DateTime, Utc}; 2 2 - use serde::{Deserialize, Serialize}; 3 3 - 4 4 - #[derive(Debug, sqlx::FromRow, Serialize, Deserialize, Clone)] 5 5 - pub struct Track { 6 6 - pub xata_id: String, 7 7 - pub title: String, 8 8 - pub artist: String, 9 9 - pub album_artist: String, 10 10 - pub album_art: Option<String>, 11 11 - pub album: String, 12 12 - pub track_number: i32, 13 13 - pub duration: i32, 14 14 - pub mb_id: Option<String>, 15 15 - pub youtube_link: Option<String>, 16 16 - pub spotify_link: Option<String>, 17 17 - pub tidal_link: Option<String>, 18 18 - pub apple_music_link: Option<String>, 19 19 - pub sha256: String, 20 20 - pub lyrics: Option<String>, 21 21 - pub composer: Option<String>, 22 22 - pub genre: Option<String>, 23 23 - pub disc_number: i32, 24 24 - pub copyright_message: Option<String>, 25 25 - pub label: Option<String>, 26 26 - pub uri: Option<String>, 27 27 - pub artist_uri: Option<String>, 28 28 - pub album_uri: Option<String>, 29 29 - #[serde(with = "chrono::serde::ts_seconds")] 30 30 - pub xata_createdat: DateTime<Utc>, 31 31 - }

-13

crates/feed/src/xata/user.rs

··· 1 1 - use chrono::{DateTime, Utc}; 2 2 - use serde::Deserialize; 3 3 - 4 4 - #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] 5 5 - pub struct User { 6 6 - pub xata_id: String, 7 7 - pub display_name: String, 8 8 - pub did: String, 9 9 - pub handle: String, 10 10 - pub avatar: String, 11 11 - #[serde(with = "chrono::serde::ts_seconds")] 12 12 - pub xata_createdat: DateTime<Utc>, 13 13 - }