From cab317ba6cd3c842874e030c2aac2ecd7a245792 Mon Sep 17 00:00:00 2001 From: Tsiry Sandratraina Date: Wed, 23 Apr 2025 17:36:47 +0300 Subject: [PATCH] [api] simplify scrobble logic, listen to jetstream --- crates/jetstream/src/main.rs | 5 +- crates/jetstream/src/profile.rs | 17 +- crates/jetstream/src/repo.rs | 546 +++++++++++++++++- crates/jetstream/src/subscriber.rs | 76 +-- crates/jetstream/src/types.rs | 101 +++- crates/jetstream/src/xata/mod.rs | 1 + crates/jetstream/src/xata/user_album.rs | 1 + crates/jetstream/src/xata/user_artist.rs | 1 + crates/jetstream/src/xata/user_playlist.rs | 11 + crates/jetstream/src/xata/user_track.rs | 5 +- rockskyapi/rocksky-auth/src/index.ts | 2 +- .../src/nowplaying/nowplaying.service.ts | 260 +-------- 12 files changed, 698 insertions(+), 328 deletions(-) create mode 100644 crates/jetstream/src/xata/user_playlist.rs diff --git a/crates/jetstream/src/main.rs b/crates/jetstream/src/main.rs index c117d29e..66e089f0 100644 --- a/crates/jetstream/src/main.rs +++ b/crates/jetstream/src/main.rs @@ -1,6 +1,6 @@ use std::env; -use subscriber::{ScrobbleSubscriber, ALBUM_NSID, ARTIST_NSID, SCROBBLE_NSID, SONG_NSID}; +use subscriber::ScrobbleSubscriber; use dotenv::dotenv; pub mod subscriber; @@ -13,8 +13,9 @@ pub mod profile; async fn main() -> Result<(), anyhow::Error> { dotenv()?; let jetstream_server = env::var("JETSTREAM_SERVER").unwrap_or_else(|_| "wss://jetstream2.us-east.bsky.network".to_string()); - let url = format!("{}/subscribe?wantedCollections={},{},{},{}", jetstream_server, SCROBBLE_NSID, ARTIST_NSID, ALBUM_NSID, SONG_NSID); + let url = format!("{}/subscribe?wantedCollections=app.rocksky.*", jetstream_server); let subscriber = ScrobbleSubscriber::new(&url); + subscriber.run().await?; Ok(()) } diff --git a/crates/jetstream/src/profile.rs b/crates/jetstream/src/profile.rs index d8b9c797..2f1c7317 100644 --- a/crates/jetstream/src/profile.rs +++ b/crates/jetstream/src/profile.rs @@ -1,5 +1,4 @@ use anyhow::Error; -use tokio::io::split; use crate::types::{Profile, ProfileResponse}; @@ -49,6 +48,22 @@ mod tests { assert_eq!(profile.r#type, "app.bsky.actor.profile"); assert!(profile.display_name.map(|s| s.starts_with("Tsiry Sandratraina")).unwrap_or(false)); assert!(profile.handle.map(|s| s == "tsiry-sandratraina.com").unwrap_or(false)); + + let did = "did:plc:fgvx5xqinqoqgpfhito5er3s"; + let profile = did_to_profile(did).await?; + + assert_eq!(profile.r#type, "app.bsky.actor.profile"); + assert!(profile.display_name.map(|s| s.starts_with("Lixtrix")).unwrap_or(false)); + assert!(profile.handle.map(|s| s == "lixtrix.art").unwrap_or(false)); + + let did = "did:plc:d5jvs7uo4z6lw63zzreukgt4"; + let profile = did_to_profile(did).await?; + assert_eq!(profile.r#type, "app.bsky.actor.profile"); + + let did = "did:plc:gwxwdfmun3aqaiu5mx7nnyof"; + let profile = did_to_profile(did).await?; + assert_eq!(profile.r#type, "app.bsky.actor.profile"); + Ok(()) } } \ No newline at end of file diff --git a/crates/jetstream/src/repo.rs b/crates/jetstream/src/repo.rs index 7b99a680..2d97dd1d 100644 --- a/crates/jetstream/src/repo.rs +++ b/crates/jetstream/src/repo.rs @@ -1,25 +1,34 @@ +use std::sync::Arc; + use anyhow::Error; +use owo_colors::OwoColorize; use sqlx::{Pool, Postgres}; +use tokio::sync::Mutex; -use crate::{profile::did_to_profile, subscriber::{ALBUM_NSID, ARTIST_NSID, SCROBBLE_NSID, SONG_NSID}, types::{AlbumRecord, ArtistRecord, Commit, ScrobbleRecord, SongRecord}, xata::{album_track::AlbumTrack, artist::Artist, track::{self, Track}, user::User}}; +use crate::{profile::did_to_profile, subscriber::{ALBUM_NSID, ARTIST_NSID, SCROBBLE_NSID, SONG_NSID}, types::{AlbumRecord, ArtistRecord, Commit, ScrobbleRecord, SongRecord}, xata::{album::Album, album_track::AlbumTrack, artist::Artist, artist_album::ArtistAlbum, artist_track::ArtistTrack, track::Track, user::User, user_album::UserAlbum, user_artist::UserArtist, user_track::UserTrack}}; -pub async fn save_scrobble(pool: &Pool, did: &str, commit: Commit) -> Result<(), Error> { +pub async fn save_scrobble(pool: Arc>>, did: &str, commit: Commit) -> Result<(), Error> { // skip unknown collection if !vec![ SCROBBLE_NSID, + ARTIST_NSID, + ALBUM_NSID, + SONG_NSID, ].contains(&commit.collection.as_str()) { return Ok(()); } + let pool = pool.lock().await; + match commit.operation.as_str() { "create" => { if commit.collection == SCROBBLE_NSID { let mut tx = pool.begin().await?; let scrobble_record: ScrobbleRecord = serde_json::from_value(commit.record.clone())?; - let album_id = save_album(&mut tx, scrobble_record.clone()).await?; + let album_id = save_album(&mut tx, scrobble_record.clone(), did).await?; let artist_id = save_artist(&mut tx, scrobble_record.clone()).await?; - let track_id = save_track(&mut tx, scrobble_record.clone()).await?; + let track_id = save_track(&mut tx, scrobble_record.clone(), did).await?; save_album_track(&mut tx, &album_id, &track_id).await?; save_artist_track(&mut tx, &artist_id, &track_id).await?; @@ -29,13 +38,15 @@ pub async fn save_scrobble(pool: &Pool, did: &str, commit: Commit) -> let user_id = save_user(&mut tx, did).await?; + println!("Saving scrobble: {} ", format!("{} - {} - {}", scrobble_record.title, scrobble_record.artist, scrobble_record.album).magenta()); + sqlx::query(r#" INSERT INTO scrobbles ( album_id, artist_id, track_id, uri, - user_id, + user_id ) VALUES ($1, $2, $3, $4, $5) "#) .bind(album_id) @@ -52,9 +63,11 @@ pub async fn save_scrobble(pool: &Pool, did: &str, commit: Commit) -> let mut tx = pool.begin().await?; let user_id = save_user(&mut tx, did).await?; + let uri = format!("at://{}/app.rocksky.artist/{}", did, commit.rkey); let artist_record: ArtistRecord = serde_json::from_value(commit.record.clone())?; - save_user_artist(&mut tx, &user_id, artist_record).await?; + save_user_artist(&mut tx, &user_id, artist_record.clone(), &uri).await?; + update_artist_uri(&mut tx, &user_id, artist_record, &uri).await?; tx.commit().await?; } @@ -62,9 +75,11 @@ pub async fn save_scrobble(pool: &Pool, did: &str, commit: Commit) -> if commit.collection == ALBUM_NSID { let mut tx = pool.begin().await?; let user_id = save_user(&mut tx, did).await?; + let uri = format!("at://{}/app.rocksky.album/{}", did, commit.rkey); let album_record: AlbumRecord = serde_json::from_value(commit.record.clone())?; - save_user_album(&mut tx, &user_id, album_record).await?; + save_user_album(&mut tx, &user_id, album_record.clone(), &uri).await?; + update_album_uri(&mut tx, &user_id, album_record, &uri).await?; tx.commit().await?; } @@ -73,9 +88,11 @@ pub async fn save_scrobble(pool: &Pool, did: &str, commit: Commit) -> let mut tx = pool.begin().await?; let user_id = save_user(&mut tx, did).await?; + let uri = format!("at://{}/app.rocksky.song/{}", did, commit.rkey); let song_record: SongRecord = serde_json::from_value(commit.record.clone())?; - save_user_track(&mut tx, &user_id, song_record).await?; + save_user_track(&mut tx, &user_id, song_record.clone(), &uri).await?; + update_track_uri(&mut tx, &user_id, song_record, &uri).await?; tx.commit().await?; } @@ -117,7 +134,7 @@ pub async fn save_user(tx: &mut sqlx::Transaction<'_, Postgres>, did: &str) -> R Ok(users[0].xata_id.clone()) } -pub async fn save_track(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_record: ScrobbleRecord) -> Result { +pub async fn save_track(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_record: ScrobbleRecord, did: &str) -> Result { let uri: Option = None; let hash = sha256::digest( format!( @@ -137,7 +154,6 @@ pub async fn save_track(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor return Ok(tracks[0].xata_id.clone()); } - let did = ""; sqlx::query(r#" INSERT INTO tracks ( title, @@ -155,7 +171,7 @@ pub async fn save_track(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor copyright_message, uri, spotify_link, - label, + label ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16 ) @@ -167,19 +183,26 @@ pub async fn save_track(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor .bind(scrobble_record.album_artist) .bind(scrobble_record.track_number) .bind(scrobble_record.duration) + .bind(scrobble_record.mbid) .bind(scrobble_record.composer) .bind(scrobble_record.lyrics) .bind(scrobble_record.disc_number) - .bind(hash) + .bind(&hash) .bind(scrobble_record.copyright_message) .bind(uri) .bind(scrobble_record.spotify_link) .bind(scrobble_record.label) .execute(&mut **tx).await?; - todo!() + + let tracks: Vec = sqlx::query_as("SELECT * FROM tracks WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + Ok(tracks[0].xata_id.clone()) } -pub async fn save_album(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_record: ScrobbleRecord) -> Result { +pub async fn save_album(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_record: ScrobbleRecord, did: &str) -> Result { let hash = sha256::digest(format!( "{} - {}", scrobble_record.album, @@ -188,18 +211,20 @@ pub async fn save_album(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor .to_lowercase() ); - let albums: Vec = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") + let albums: Vec = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") .bind(&hash) .fetch_all(&mut **tx) .await?; if !albums.is_empty() { + println!("Album already exists: {}", albums[0].title.magenta()); return Ok(albums[0].xata_id.clone()); } + println!("Saving album: {}", scrobble_record.album.magenta()); + let uri: Option = None; let artist_uri: Option = None; - let did = ""; sqlx::query(r#" INSERT INTO albums ( title, @@ -209,7 +234,7 @@ pub async fn save_album(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor release_date, sha256, uri, - artist_uri, + artist_uri ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8 ) @@ -219,11 +244,17 @@ pub async fn save_album(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_recor .bind(scrobble_record.album_art.map(|x| format!("https://cdn.bsky.app/img/feed_thumbnail/plain/{}/{}@{}", did, x.r#ref.link, x.mime_type.split('/').last().unwrap_or("jpeg")))) .bind(scrobble_record.year) .bind(scrobble_record.release_date) - .bind(hash) + .bind(&hash) .bind(uri) .bind(artist_uri) .execute(&mut **tx).await?; - todo!() + + let albums: Vec = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + Ok(albums[0].xata_id.clone()) } pub async fn save_artist(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_record: ScrobbleRecord) -> Result { @@ -234,9 +265,12 @@ pub async fn save_artist(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_reco .await?; if !artists.is_empty() { + println!("Artist already exists: {}", artists[0].name.magenta()); return Ok(artists[0].xata_id.clone()); } + println!("Saving artist: {}", scrobble_record.album_artist.magenta()); + let uri: Option = None; let picture = ""; sqlx::query(r#" @@ -244,17 +278,23 @@ pub async fn save_artist(tx: &mut sqlx::Transaction<'_, Postgres>, scrobble_reco name, sha256, uri, - picture, + picture ) VALUES ( $1, $2, $3, $4 ) "#) .bind(scrobble_record.artist) - .bind(hash) + .bind(&hash) .bind(uri) .bind(picture) .execute(&mut **tx).await?; - todo!() + + let artists: Vec = sqlx::query_as("SELECT * FROM artists WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + Ok(artists[0].xata_id.clone()) } pub async fn save_album_track(tx: &mut sqlx::Transaction<'_, Postgres>, album_id: &str, track_id: &str) -> Result<(), Error> { @@ -265,13 +305,16 @@ pub async fn save_album_track(tx: &mut sqlx::Transaction<'_, Postgres>, album_id .await?; if !album_tracks.is_empty() { + println!("Album track already exists: {}", format!("{} - {}", album_id, track_id).magenta()); return Ok(()); } + println!("Saving album track: {}", format!("{} - {}", album_id, track_id).magenta()); + sqlx::query(r#" INSERT INTO album_tracks ( album_id, - track_id, + track_id ) VALUES ( $1, $2 ) @@ -283,20 +326,23 @@ pub async fn save_album_track(tx: &mut sqlx::Transaction<'_, Postgres>, album_id } pub async fn save_artist_track(tx: &mut sqlx::Transaction<'_, Postgres>, artist_id: &str, track_id: &str) -> Result<(), Error> { - let artist_tracks : Vec = sqlx::query_as("SELECT * FROM artist_tracks WHERE artist_id = $1 AND track_id = $2") + let artist_tracks : Vec = sqlx::query_as("SELECT * FROM artist_tracks WHERE artist_id = $1 AND track_id = $2") .bind(artist_id) .bind(track_id) .fetch_all(&mut **tx) .await?; if !artist_tracks.is_empty() { + println!("Artist track already exists: {}", format!("{} - {}", artist_id, track_id).magenta()); return Ok(()); } + println!("Saving artist track: {}", format!("{} - {}", artist_id, track_id).magenta()); + sqlx::query(r#" INSERT INTO artist_tracks ( artist_id, - track_id, + track_id ) VALUES ( $1, $2 ) @@ -308,20 +354,23 @@ pub async fn save_artist_track(tx: &mut sqlx::Transaction<'_, Postgres>, artist_ } pub async fn save_artist_album(tx: &mut sqlx::Transaction<'_, Postgres>, artist_id: &str, album_id: &str) -> Result<(), Error> { - let artist_albums : Vec = sqlx::query_as("SELECT * FROM artist_albums WHERE artist_id = $1 AND album_id = $2") + let artist_albums : Vec = sqlx::query_as("SELECT * FROM artist_albums WHERE artist_id = $1 AND album_id = $2") .bind(artist_id) .bind(album_id) .fetch_all(&mut **tx) .await?; if !artist_albums.is_empty() { + println!("Artist album already exists: {}", format!("{} - {}", artist_id, album_id).magenta()); return Ok(()); } + println!("Saving artist album: {}", format!("{} - {}", artist_id, album_id).magenta()); + sqlx::query(r#" INSERT INTO artist_albums ( artist_id, - album_id, + album_id ) VALUES ( $1, $2 ) @@ -333,14 +382,451 @@ pub async fn save_artist_album(tx: &mut sqlx::Transaction<'_, Postgres>, artist_ } -pub async fn save_user_artist(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: ArtistRecord) -> Result<(), Error> { +pub async fn save_user_artist(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: ArtistRecord, uri: &str) -> Result<(), Error> { + let hash = sha256::digest(record.name.to_lowercase()); + + let mut artists: Vec = sqlx::query_as("SELECT * FROM artists WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + let users: Vec = sqlx::query_as("SELECT * FROM users WHERE xata_id = $1") + .bind(user_id) + .fetch_all(&mut **tx) + .await?; + + let artist_id: &str; + + match artists.is_empty() { + true => { + println!("Saving artist: {}", record.name.magenta()); + let did = users[0].did.clone(); + sqlx::query(r#" + INSERT INTO artists ( + name, + sha256, + uri, + picture + ) VALUES ( + $1, $2, $3, $4 + ) + "#) + .bind(record.name) + .bind(&hash) + .bind(uri) + .bind(record.picture.map(|x| format!("https://cdn.bsky.app/img/avatar/plain/{}/{}@{}", did, x.r#ref.link, x.mime_type.split('/').last().unwrap_or("jpeg")))) + .execute(&mut **tx).await?; + + artists = sqlx::query_as("SELECT * FROM artists WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + artist_id = &artists[0].xata_id; + }, + false => { + artist_id = &artists[0].xata_id; + } + }; + + let user_artists: Vec = sqlx::query_as("SELECT * FROM user_artists WHERE user_id = $1 AND artist_id = $2") + .bind(user_id) + .bind(artist_id) + .fetch_all(&mut **tx) + .await?; + + if !user_artists.is_empty() { + println!("User artist already exists: {}", format!("{} - {}", user_id, artist_id).magenta()); + sqlx::query(r#" + UPDATE user_artists + SET scrobbles = scrobbles + 1, + uri = $3 + WHERE user_id = $1 AND artist_id = $2 + "#) + .bind(user_id) + .bind(artist_id) + .bind(uri) + .execute(&mut **tx).await?; + return Ok(()); + } + + println!("Saving user artist: {}", format!("{} - {}", user_id, artist_id).magenta()); + + sqlx::query(r#" + INSERT INTO user_artists ( + user_id, + artist_id, + uri, + scrobbles + ) VALUES ( + $1, $2, $3, $4 + ) + "#) + .bind(user_id) + .bind(artist_id) + .bind(uri) + .bind(1) + .execute(&mut **tx).await?; Ok(()) } -pub async fn save_user_album(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: AlbumRecord) -> Result<(), Error> { +pub async fn save_user_album(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: AlbumRecord, uri: &str) -> Result<(), Error> { + let users: Vec = sqlx::query_as("SELECT * FROM users WHERE xata_id = $1") + .bind(user_id) + .fetch_all(&mut **tx) + .await?; + + let hash = sha256::digest(format!( + "{} - {}", + record.title, + record.artist + ) + .to_lowercase() + ); + let mut albums: Vec = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + let album_id: &str; + + match albums.is_empty() { + true => { + println!("Saving album: {}", record.title.magenta()); + let did = users[0].did.clone(); + sqlx::query(r#" + INSERT INTO albums ( + title, + artist, + album_art, + year, + release_date, + sha256, + uri + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7 + ) + "#) + .bind(record.title) + .bind(record.artist) + .bind(record.album_art.map(|x| format!("https://cdn.bsky.app/img/feed_thumbnail/plain/{}/{}@{}", did, x.r#ref.link, x.mime_type.split('/').last().unwrap_or("jpeg")))) + .bind(record.year) + .bind(record.release_date) + .bind(&hash) + .bind(uri) + .execute(&mut **tx).await?; + + albums = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + album_id = &albums[0].xata_id; + }, + false => { + album_id = &albums[0].xata_id; + } + }; + + let user_albums: Vec = sqlx::query_as("SELECT * FROM user_albums WHERE user_id = $1 AND album_id = $2") + .bind(user_id) + .bind(album_id) + .fetch_all(&mut **tx) + .await?; + + if !user_albums.is_empty() { + println!("User album already exists: {}", format!("{} - {}", user_id, album_id).magenta()); + sqlx::query(r#" + UPDATE user_albums + SET scrobbles = scrobbles + 1, + uri = $3 + WHERE user_id = $1 AND album_id = $2 + "#) + .bind(user_id) + .bind(album_id) + .bind(uri) + .execute(&mut **tx).await?; + return Ok(()); + } + + println!("Saving user album: {}", format!("{} - {}", user_id, album_id).magenta()); + + sqlx::query(r#" + INSERT INTO user_albums ( + user_id, + album_id, + uri, + scrobbles + ) VALUES ( + $1, $2, $3, $4 + ) + "#) + .bind(user_id) + .bind(album_id) + .bind(uri) + .bind(1) + .execute(&mut **tx).await?; Ok(()) } -pub async fn save_user_track(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: SongRecord) -> Result<(), Error> { +pub async fn save_user_track(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: SongRecord, uri: &str) -> Result<(), Error> { + let hash = sha256::digest(format!( + "{} - {} - {}", + record.title, + record.artist, + record.album + ) + .to_lowercase() + ); + + let mut tracks: Vec = sqlx::query_as("SELECT * FROM tracks WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + let users: Vec = sqlx::query_as("SELECT * FROM users WHERE xata_id = $1") + .bind(user_id) + .fetch_all(&mut **tx) + .await?; + + let track_id: &str; + + match tracks.is_empty() { + true => { + println!("Saving track: {}", record.title.magenta()); + let did = users[0].did.clone(); + sqlx::query(r#" + INSERT INTO tracks ( + title, + artist, + album, + album_art, + album_artist, + track_number, + duration, + mb_id, + composer, + lyrics, + disc_number, + sha256, + copyright_message, + uri, + spotify_link, + label + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16 + ) + "#) + .bind(record.title) + .bind(record.artist) + .bind(record.album) + .bind(record.album_art.map(|x| format!("https://cdn.bsky.app/img/feed_thumbnail/plain/{}/{}@{}", did, x.r#ref.link, x.mime_type.split('/').last().unwrap_or("jpeg")))) + .bind(record.album_artist) + .bind(record.track_number) + .bind(record.duration) + .bind(record.mbid) + .bind(record.composer) + .bind(record.lyrics) + .bind(record.disc_number) + .bind(&hash) + .bind(record.copyright_message) + .bind(uri) + .bind(record.spotify_link) + .bind(record.label) + .execute(&mut **tx).await?; + + tracks = sqlx::query_as("SELECT * FROM tracks WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + track_id = &tracks[0].xata_id; + }, + false => { + track_id = &tracks[0].xata_id; + } + } + + let user_tracks: Vec = sqlx::query_as("SELECT * FROM user_tracks WHERE user_id = $1 AND track_id = $2") + .bind(user_id) + .bind(track_id) + .fetch_all(&mut **tx) + .await?; + + if !user_tracks.is_empty() { + println!("User track already exists: {}", format!("{} - {}", user_id, track_id).magenta()); + sqlx::query(r#" + UPDATE user_tracks + SET scrobbles = scrobbles + 1, + uri = $3 + WHERE user_id = $1 AND track_id = $2 + "#) + .bind(user_id) + .bind(track_id) + .bind(uri) + .execute(&mut **tx).await?; + return Ok(()); + } + + println!("Saving user track: {}", format!("{} - {}", user_id, track_id).magenta()); + + sqlx::query(r#" + INSERT INTO user_tracks ( + user_id, + track_id, + uri, + scrobbles + ) VALUES ( + $1, $2, $3, $4 + ) + "#) + .bind(user_id) + .bind(track_id) + .bind(uri) + .bind(1) + .execute(&mut **tx).await?; + + Ok(()) +} + +pub async fn update_artist_uri(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: ArtistRecord, uri: &str) -> Result<(), Error> { + let hash = sha256::digest(record.name.to_lowercase()); + let artists: Vec = sqlx::query_as("SELECT * FROM artists WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + if artists.is_empty() { + println!("Artist not found: {}", record.name.magenta()); + return Ok(()); + } + + let artist_id = &artists[0].xata_id; + + sqlx::query(r#" + UPDATE user_artists + SET uri = $3 + WHERE user_id = $1 AND artist_id = $2 + "#) + .bind(user_id) + .bind(artist_id) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE tracks + SET artist_uri = $2 + WHERE artist_uri IS NULL AND album_artist = $1 + "#) + .bind(&record.name) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE artists + SET uri = $2 + WHERE sha256 = $1 + "#) + .bind(&hash) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE albums + SET artist_uri = $2 + WHERE artist_uri IS NULL AND artist = $1 + "#) + .bind(&record.name) + .bind(uri) + .execute(&mut **tx).await?; + Ok(()) +} + +pub async fn update_album_uri(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: AlbumRecord, uri: &str) -> Result<(), Error> { + let hash = sha256::digest(format!( + "{} - {}", + record.title, + record.artist + ) + .to_lowercase() + ); + let albums: Vec = sqlx::query_as("SELECT * FROM albums WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + if albums.is_empty() { + println!("Album not found: {}", record.title.magenta()); + return Ok(()); + } + let album_id = &albums[0].xata_id; + sqlx::query(r#" + UPDATE user_albums + SET uri = $3 + WHERE user_id = $1 AND album_id = $2 + "#) + .bind(user_id) + .bind(album_id) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE tracks + SET album_uri = $2 + WHERE album_uri IS NULL AND album = $1 + "#) + .bind(record.title) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE albums + SET uri = $2 + WHERE sha256 = $1 + "#) + .bind(&hash) + .bind(uri) + .execute(&mut **tx).await?; + Ok(()) } + +pub async fn update_track_uri(tx: &mut sqlx::Transaction<'_, Postgres>, user_id: &str, record: SongRecord, uri: &str) -> Result<(), Error> { + let hash = sha256::digest(format!( + "{} - {} - {}", + record.title, + record.artist, + record.album + ) + .to_lowercase() + ); + let tracks: Vec = sqlx::query_as("SELECT * FROM tracks WHERE sha256 = $1") + .bind(&hash) + .fetch_all(&mut **tx) + .await?; + + if tracks.is_empty() { + println!("Track not found: {}", record.title.magenta()); + return Ok(()); + } + + let track_id = &tracks[0].xata_id; + sqlx::query(r#" + UPDATE user_tracks + SET uri = $3 + WHERE user_id = $1 AND track_id = $2 + "#) + .bind(user_id) + .bind(track_id) + .bind(uri) + .execute(&mut **tx).await?; + + sqlx::query(r#" + UPDATE tracks + SET uri = $2 + WHERE sha256 = $1 AND uri IS NULL + "#) + .bind(&hash) + .bind(uri) + .execute(&mut **tx).await?; + + Ok(()) +} + diff --git a/crates/jetstream/src/subscriber.rs b/crates/jetstream/src/subscriber.rs index dc2d8189..953bad71 100644 --- a/crates/jetstream/src/subscriber.rs +++ b/crates/jetstream/src/subscriber.rs @@ -1,13 +1,14 @@ -use std::env; +use std::{env, sync::Arc}; use anyhow::{Error, Context}; use futures_util::StreamExt; use owo_colors::OwoColorize; use sqlx::postgres::PgPoolOptions; +use tokio::sync::Mutex; use tokio_tungstenite::{connect_async, tungstenite::Message}; -use tokio::sync::mpsc; -use crate::{repo::save_scrobble, types::{Commit, Root}}; + +use crate::{repo::save_scrobble, types::Root}; pub const SCROBBLE_NSID: &str = "app.rocksky.scrobble"; pub const ARTIST_NSID: &str = "app.rocksky.artist"; @@ -34,17 +35,9 @@ impl ScrobbleSubscriber { let db_url = env::var("XATA_POSTGRES_URL") .context("Failed to get XATA_POSTGRES_URL environment variable")?; - let (tx, rx) = mpsc::channel::<(String, Commit)>(100); - - let tx_clone = tx.clone(); - - // Start the processor task - let processor = tokio::spawn(async move { - let pool = PgPoolOptions::new().max_connections(5) - .connect(&db_url).await?; - - process_scrobble_events(rx, &pool).await - }); + let pool = PgPoolOptions::new().max_connections(5) + .connect(&db_url).await?; + let pool = Arc::new(Mutex::new(pool)); let (mut ws_stream, _) = connect_async(&self.service_url).await?; println!("Connected to jetstream at {}", self.service_url.bright_green()); @@ -52,7 +45,7 @@ impl ScrobbleSubscriber { while let Some(msg) = ws_stream.next().await { match msg { Ok(msg) => { - if let Err(e) = self.handle_message(msg, &tx_clone).await { + if let Err(e) = handle_message(pool.clone(), msg).await { eprintln!("Error handling message: {}", e); } } @@ -63,48 +56,37 @@ impl ScrobbleSubscriber { } } - drop(tx); - - // Wait for the processor task to complete - match processor.await { - Ok(result) => { - if let Err(e) = result { - eprintln!("Processor task had an error: {}", e); - } - } - Err(e) => { - eprintln!("Processor task panicked: {}", e); - } - } Ok(()) } +} - async fn handle_message( - &self, - msg: Message, - tx: &mpsc::Sender<(String, Commit)>, - ) -> Result<(), Error> { +async fn handle_message( + pool: Arc>, + msg: Message, +) -> Result<(), Error> { + tokio::spawn(async move { if let Message::Text(text) = msg { let message: Root = serde_json::from_str(&text)?; + + if message.kind != "commit" { + return Ok::<(), Error>(()); + } + println!("Received message: {:#?}", message); if let Some(commit) = message.commit { - tx.send((message.did, commit)).await.map_err(|e| { - Error::msg(format!("Failed to send message to channel: {}", e)) - })?; + match save_scrobble(pool, &message.did, commit).await { + Ok(_) => { + println!("Scrobble saved successfully"); + } + Err(e) => { + eprintln!("Error saving scrobble: {}", e); + } + } } } - Ok(()) - } -} + }); -async fn process_scrobble_events( - mut rx: mpsc::Receiver<(String, Commit)>, - pool: &sqlx::Pool, -) -> Result<(), Error> { - while let Some((did, record)) = rx.recv().await { - save_scrobble(pool, &did, record).await?; - } Ok(()) -} \ No newline at end of file +} diff --git a/crates/jetstream/src/types.rs b/crates/jetstream/src/types.rs index ed6b7d9c..29ff051f 100644 --- a/crates/jetstream/src/types.rs +++ b/crates/jetstream/src/types.rs @@ -21,7 +21,7 @@ pub struct Commit { #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "camelCase")] -pub struct AlbumArt { +pub struct Blob { #[serde(rename = "$type")] pub r#type: String, pub r#ref: Ref, @@ -64,7 +64,7 @@ pub struct ScrobbleRecord { #[serde(skip_serializing_if = "Option::is_none")] pub wiki: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub album_art: Option, + pub album_art: Option, #[serde(skip_serializing_if = "Option::is_none")] pub youtube_link: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -94,28 +94,25 @@ pub struct Profile { pub r#type: String, pub avatar: Option, pub banner: Option, - pub created_at: String, - #[serde(rename = "pinnedPost")] + pub created_at: Option, pub pinned_post: Option, pub description: Option, - #[serde(rename = "displayName")] pub display_name: Option, pub handle: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "camelCase")] -pub struct Blob { +pub struct ImageBlob { #[serde(rename = "$type")] pub r#type: String, #[serde(rename = "ref")] pub r#ref: BlobRef, - #[serde(rename = "mimeType")] pub mime_type: String, pub size: u64, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, Clone)] pub struct BlobRef { #[serde(rename = "$link")] pub link: String, @@ -129,12 +126,92 @@ pub struct PinnedPost { #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "camelCase")] -pub struct ArtistRecord {} +pub struct ArtistRecord { + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub bio: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub picture: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tags: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub born: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub died: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub born_in: Option, + pub created_at: String, +} #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "camelCase")] -pub struct AlbumRecord {} +pub struct AlbumRecord { + pub title: String, + pub artist: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub release_date: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub year: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub genre: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub album_art: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tags: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub youtube_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub spotify_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tidal_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub apple_music_link: Option, + pub created_at: String, +} #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "camelCase")] -pub struct SongRecord {} +pub struct SongRecord { + pub title: String, + pub artist: String, + pub album: String, + pub album_artist: String, + pub duration: i32, + pub created_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub track_number: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub disc_number: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub genre: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub release_date: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub year: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tags: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub composer: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lyrics: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub copyright_message: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub wiki: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub album_art: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub youtube_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub spotify_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tidal_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub apple_music_link: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub label: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub mbid: Option, +} diff --git a/crates/jetstream/src/xata/mod.rs b/crates/jetstream/src/xata/mod.rs index 65ddb904..fe938aa6 100644 --- a/crates/jetstream/src/xata/mod.rs +++ b/crates/jetstream/src/xata/mod.rs @@ -9,4 +9,5 @@ pub mod track; pub mod user; pub mod user_album; pub mod user_artist; +pub mod user_playlist; pub mod user_track; diff --git a/crates/jetstream/src/xata/user_album.rs b/crates/jetstream/src/xata/user_album.rs index 67bad286..a1748431 100644 --- a/crates/jetstream/src/xata/user_album.rs +++ b/crates/jetstream/src/xata/user_album.rs @@ -5,6 +5,7 @@ pub struct UserAlbum { pub xata_id: String, pub user_id: String, pub album_id: String, + pub uri: Option, #[serde(with = "chrono::serde::ts_seconds")] pub xata_createdat: chrono::DateTime, } diff --git a/crates/jetstream/src/xata/user_artist.rs b/crates/jetstream/src/xata/user_artist.rs index 469e6a4d..e2de6e1b 100644 --- a/crates/jetstream/src/xata/user_artist.rs +++ b/crates/jetstream/src/xata/user_artist.rs @@ -5,6 +5,7 @@ pub struct UserArtist { pub xata_id: String, pub user_id: String, pub artist_id: String, + pub uri: Option, #[serde(with = "chrono::serde::ts_seconds")] pub xata_createdat: chrono::DateTime, } diff --git a/crates/jetstream/src/xata/user_playlist.rs b/crates/jetstream/src/xata/user_playlist.rs new file mode 100644 index 00000000..8d6eaf5f --- /dev/null +++ b/crates/jetstream/src/xata/user_playlist.rs @@ -0,0 +1,11 @@ +use serde::Deserialize; + +#[derive(Debug, sqlx::FromRow, Deserialize, Clone)] +pub struct UserPlaylist { + pub xata_id: String, + pub user_id: String, + pub playlist_id: String, + pub uri: Option, + #[serde(with = "chrono::serde::ts_seconds")] + pub xata_createdat: chrono::DateTime, +} diff --git a/crates/jetstream/src/xata/user_track.rs b/crates/jetstream/src/xata/user_track.rs index af8f705d..75157db2 100644 --- a/crates/jetstream/src/xata/user_track.rs +++ b/crates/jetstream/src/xata/user_track.rs @@ -1,10 +1,11 @@ use serde::Deserialize; #[derive(Debug, sqlx::FromRow, Deserialize, Clone)] -pub struct UserPlaylist { +pub struct UserTrack { pub xata_id: String, pub user_id: String, - pub playlist_id: String, + pub track_id: String, + pub uri: Option, #[serde(with = "chrono::serde::ts_seconds")] pub xata_createdat: chrono::DateTime, } diff --git a/rockskyapi/rocksky-auth/src/index.ts b/rockskyapi/rocksky-auth/src/index.ts index 8ee15dd0..e9d59852 100644 --- a/rockskyapi/rocksky-auth/src/index.ts +++ b/rockskyapi/rocksky-auth/src/index.ts @@ -112,7 +112,7 @@ app.post("/now-playing", async (c) => { return c.text("Unauthorized"); } - await scrobbleTrack(ctx, track, user, agent); + await scrobbleTrack(ctx, track, agent); return c.json({ status: "ok" }); }); diff --git a/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts b/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts index 4ac4a70a..1ef9169a 100644 --- a/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts +++ b/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts @@ -1,6 +1,7 @@ import { Agent, BlobRef } from "@atproto/api"; import { TID } from "@atproto/common"; -import { equals, SelectedPick } from "@xata.io/client"; +import { equals } from "@xata.io/client"; +import chalk from "chalk"; import { Context } from "context"; import { createHash } from "crypto"; import dayjs from "dayjs"; @@ -10,7 +11,6 @@ import * as Scrobble from "lexicon/types/app/rocksky/scrobble"; import * as Song from "lexicon/types/app/rocksky/song"; import downloadImage, { getContentType } from "lib/downloadImage"; import { Track } from "types/track"; -import { ScrobblesRecord } from "xata"; export async function putArtistRecord( track: Track, @@ -254,106 +254,6 @@ async function putScrobbleRecord( } } -export async function updateUserLibrary( - ctx: Context, - user, - track: Track, - agent: Agent, - track_id: string, - album_id: string, - artist_id: string, - trackUri: string, - albumUri: string, - artistUri: string -): Promise { - const existingUserTrack = await ctx.client.db.user_tracks - .filter("user_id", equals(user.xata_id)) - .filter("track_id", equals(track_id)) - .getFirst(); - - if (!trackUri.includes(user.did)) { - trackUri = await putSongRecord(track, agent); - } - - if (!existingUserTrack) { - await ctx.client.db.user_tracks.create({ - user_id: user.xata_id, - track_id, - uri: trackUri, - scrobbles: 1, - }); - } else { - await ctx.client.db.user_tracks.update({ - xata_id: existingUserTrack.xata_id, - uri: trackUri, - scrobbles: existingUserTrack.scrobbles - ? existingUserTrack.scrobbles + 1 - : 1, - }); - } - - const existingUserArtist = await ctx.client.db.user_artists - .filter("user_id", equals(user.xata_id)) - .filter({ - $any: [ - { - artist_id, - }, - { - uri: artistUri, - }, - ], - }) - .getFirst(); - - if (!artistUri.includes(user.did)) { - artistUri = await putArtistRecord(track, agent); - } - - if (!existingUserArtist) { - await ctx.client.db.user_artists.create({ - user_id: user.xata_id, - artist_id, - uri: artistUri, - scrobbles: 1, - }); - } else { - await ctx.client.db.user_artists.update({ - xata_id: existingUserArtist.xata_id, - uri: artistUri, - scrobbles: existingUserArtist.scrobbles - ? existingUserArtist.scrobbles + 1 - : 1, - }); - } - - const existingUserAlbum = await ctx.client.db.user_albums - .filter("user_id", equals(user.xata_id)) - .filter("album_id", equals(album_id)) - .getFirst(); - - if (!albumUri.includes(user.did)) { - albumUri = await putAlbumRecord(track, agent); - } - - if (!existingUserAlbum) { - await ctx.client.db.user_albums.create({ - user_id: user.xata_id, - album_id, - uri: albumUri, - scrobbles: 1, - }); - } else { - await ctx.client.db.user_albums.update({ - xata_id: existingUserAlbum.xata_id, - uri: albumUri, - scrobbles: existingUserAlbum.scrobbles - ? existingUserAlbum.scrobbles + 1 - : 1, - }); - } -} - export async function publishScrobble(ctx: Context, id: string) { const scrobble = await ctx.client.db.scrobbles .select(["*", "track_id.*", "album_id.*", "artist_id.*", "user_id.*"]) @@ -411,9 +311,8 @@ export async function publishScrobble(ctx: Context, id: string) { export async function scrobbleTrack( ctx: Context, track: Track, - user, agent: Agent -): Promise>> { +): Promise { const existingTrack = await ctx.client.db.tracks .filter( "sha256", @@ -427,38 +326,10 @@ export async function scrobbleTrack( ) .getFirst(); - let trackUri = existingTrack?.uri; if (!existingTrack?.uri) { - trackUri = await putSongRecord(track, agent); + await putSongRecord(track, agent); } - const { xata_id: track_id } = await ctx.client.db.tracks.createOrUpdate( - existingTrack?.xata_id, - { - title: track.title, - artist: track.artist, - album: track.album, - album_art: track.albumArt, - album_artist: track.albumArtist, - track_number: track.trackNumber, - duration: track.duration, - mb_id: track.mbId, - composer: track.composer, - lyrics: track.lyrics, - disc_number: track.discNumber, - // compute sha256 (lowercase(title + artist + album)) - sha256: createHash("sha256") - .update( - `${track.title} - ${track.artist} - ${track.album}`.toLowerCase() - ) - .digest("hex"), - copyright_message: track.copyrightMessage, - uri: trackUri ? trackUri : undefined, - spotify_link: track.spotifyLink ? track.spotifyLink : undefined, - label: track.label ? track.label : undefined, - } - ); - const existingArtist = await ctx.client.db.artists .filter( "sha256", @@ -470,22 +341,10 @@ export async function scrobbleTrack( ) .getFirst(); - let artistUri = existingArtist?.uri; if (!existingArtist?.uri) { - artistUri = await putArtistRecord(track, agent); + await putArtistRecord(track, agent); } - const { xata_id: artist_id, uri: new_artist_uri } = - await ctx.client.db.artists.createOrUpdate(existingArtist?.xata_id, { - name: track.albumArtist, - // compute sha256 (lowercase(name)) - sha256: createHash("sha256") - .update(track.albumArtist.toLowerCase()) - .digest("hex"), - uri: artistUri ? artistUri : undefined, - picture: track.artistPicture ? track.artistPicture : undefined, - }); - const existingAlbum = await ctx.client.db.albums .filter( "sha256", @@ -497,97 +356,32 @@ export async function scrobbleTrack( ) .getFirst(); - let albumUri = existingAlbum?.uri; if (!existingAlbum?.uri) { - albumUri = await putAlbumRecord(track, agent); + await putAlbumRecord(track, agent); } - const { xata_id: album_id, uri: new_album_uri } = - await ctx.client.db.albums.createOrUpdate(existingAlbum?.xata_id, { - title: track.album, - artist: track.albumArtist, - album_art: track.albumArt, - year: track.year, - release_date: track.releaseDate - ? track.releaseDate.toISOString() - : undefined, - // compute sha256 (lowercase(title + artist)) - sha256: createHash("sha256") - .update(`${track.album} - ${track.albumArtist}`.toLowerCase()) - .digest("hex"), - uri: albumUri ? albumUri : undefined, - artist_uri: new_artist_uri, - }); - - const existingAlbumTrack = await ctx.client.db.album_tracks - .filter("album_id", equals(album_id)) - .filter("track_id", equals(track_id)) - .getFirst(); - - await ctx.client.db.album_tracks.createOrUpdate(existingAlbumTrack?.xata_id, { - album_id, - track_id, - }); - - const existingArtistTrack = await ctx.client.db.artist_tracks - .filter("artist_id", equals(artist_id)) - .filter("track_id", equals(track_id)) - .getFirst(); - - await ctx.client.db.artist_tracks.createOrUpdate( - existingArtistTrack?.xata_id, - { - artist_id, - track_id, - } - ); - - const existingArtistAlbum = await ctx.client.db.artist_albums - .filter("artist_id", equals(artist_id)) - .filter("album_id", equals(album_id)) - .getFirst(); - - await ctx.client.db.artist_albums.createOrUpdate( - existingArtistAlbum?.xata_id, - { - artist_id, - album_id, - } - ); - const scrobbleUri = await putScrobbleRecord(track, agent); - await updateUserLibrary( - ctx, - user, - track, - agent, - track_id, - album_id, - artist_id, - trackUri, - albumUri, - artistUri - ); - - await ctx.client.db.tracks.update({ - xata_id: track_id, - artist_uri: new_artist_uri, - album_uri: new_album_uri, - }); - - const scrobble = await ctx.client.db.scrobbles.create({ - user_id: user.xata_id, - track_id, - album_id, - artist_id, - uri: scrobbleUri, - timestamp: track.timestamp - ? dayjs.unix(track.timestamp).toDate() - : new Date(), - }); - - await publishScrobble(ctx, scrobble.xata_id); + // loop while scrobble is null, try 5 times, sleep 1 second between tries + let tries = 0, + scrobble = null; + while (!scrobble && tries < 5) { + scrobble = await ctx.client.db.scrobbles + .select(["*", "track_id.*", "album_id.*", "artist_id.*", "user_id.*"]) + .filter("uri", equals(scrobbleUri)) + .getFirst(); + + if (scrobble) { + await publishScrobble(ctx, scrobble.xata_id); + console.log("Scrobble published"); + break; + } + tries += 1; + console.log("Scrobble not found, trying again: ", chalk.magenta(tries)); + await new Promise((resolve) => setTimeout(resolve, 1000)); + } - return scrobble; + if (tries === 5 && !scrobble) { + console.log(`Scrobble not found after ${chalk.magenta("5 tries")}`); + } } -- 2.51.2