From a30892db4e6676accab67aaf865a52cd84211aaa Mon Sep 17 00:00:00 2001
From: Tsiry Sandratraina <tsiry.sndr@fluentci.io>
Date: Sun, 2 Mar 2025 23:55:31 +0300
Subject: [PATCH] rocksky: fix artist album not published to nats

---
 crates/analytics/src/core.rs                  | 26 ++++-----
 crates/analytics/src/subscriber/mod.rs        | 18 ++++++
 crates/analytics/src/subscriber/types.rs      | 11 ++++
 .../src/nowplaying/nowplaying.service.ts      | 58 +++++++++++--------
 4 files changed, 77 insertions(+), 36 deletions(-)

diff --git a/crates/analytics/src/core.rs b/crates/analytics/src/core.rs
index 9870d62d..bb5d7062 100644
--- a/crates/analytics/src/core.rs
+++ b/crates/analytics/src/core.rs
@@ -580,19 +580,19 @@ pub async fn load_artist_albums(conn: Arc<Mutex<Connection>>, pool: &Pool<Postgr
     .await?;
 
     for (i, artist_album) in artist_albums.clone().into_iter().enumerate() {
-    println!("artist_albums {} - {} - {}", i, artist_album.artist_id.bright_green(), artist_album.album_id);
-    match conn.execute(
-        "INSERT INTO artist_albums (id, artist_id, album_id, created_at) VALUES (?, ?, ?, ?)",
-            params![
-            artist_album.xata_id,
-            artist_album.artist_id,
-            artist_album.album_id,
-            artist_album.xata_createdat,
-            ],
-    ) {
-        Ok(_) => (),
-        Err(e) => println!("error: {}", e),
-    }
+        println!("artist_albums {} - {} - {}", i, artist_album.artist_id.bright_green(), artist_album.album_id);
+        match conn.execute(
+            "INSERT INTO artist_albums (id, artist_id, album_id, created_at) VALUES (?, ?, ?, ?)",
+                params![
+                artist_album.xata_id,
+                artist_album.artist_id,
+                artist_album.album_id,
+                artist_album.xata_createdat,
+                ],
+        ) {
+            Ok(_) => (),
+            Err(e) => println!("error: {}", e),
+        }
     }
 
     println!("artist_albums: {:?}", artist_albums.len());
diff --git a/crates/analytics/src/subscriber/mod.rs b/crates/analytics/src/subscriber/mod.rs
index 80757181..483a5440 100644
--- a/crates/analytics/src/subscriber/mod.rs
+++ b/crates/analytics/src/subscriber/mod.rs
@@ -336,6 +336,24 @@ pub async fn save_scrobble(conn: Arc<Mutex<Connection>>, payload: ScrobblePayloa
         }
     }
 
+    match conn.execute(
+      "INSERT INTO artist_albums (id, artist_id, album_id, created_at) VALUES (?, ?, ?, ?)",
+          params![
+            payload.artist_album.xata_id,
+            payload.artist_album.artist_id.xata_id,
+            payload.artist_album.album_id.xata_id,
+            payload.artist_album.xata_createdat,
+          ],
+  ) {
+      Ok(_) => (),
+      Err(e) => {
+        if !e.to_string().contains("violates primary key constraint") {
+          println!("[artist_albums] error: {}", e);
+          return Err(e.into());
+        }
+      }
+  }
+
     match conn.execute(
       "INSERT INTO user_albums (id, user_id, album_id, created_at) VALUES (?, ?, ?, ?)",
        params![
diff --git a/crates/analytics/src/subscriber/types.rs b/crates/analytics/src/subscriber/types.rs
index e7611d97..d5add7f0 100644
--- a/crates/analytics/src/subscriber/types.rs
+++ b/crates/analytics/src/subscriber/types.rs
@@ -33,6 +33,7 @@ pub struct ScrobblePayload {
     pub user_track: UserTrack,
     pub album_track: AlbumTrack,
     pub artist_track: ArtistTrack,
+    pub artist_album: ArtistAlbum,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -210,6 +211,16 @@ pub struct ArtistTrack {
     pub xata_version: i32,
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ArtistAlbum {
+    pub album_id: Ref,
+    pub artist_id: Ref,
+    pub xata_createdat: DateTime<Utc>,
+    pub xata_id: String,
+    pub xata_updatedat: DateTime<Utc>,
+    pub xata_version: i32,
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct Ref {
     pub xata_id: String,
diff --git a/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts b/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts
index f9c4f079..45e6b06c 100644
--- a/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts
+++ b/rockskyapi/rocksky-auth/src/nowplaying/nowplaying.service.ts
@@ -323,29 +323,40 @@ export async function publishScrobble(ctx: Context, id: string) {
     .filter("xata_id", equals(id))
     .getFirst();
 
-  const [user_album, user_artist, user_track, album_track, artist_track] =
-    await Promise.all([
-      ctx.client.db.user_albums
-        .select(["*"])
-        .filter("album_id.xata_id", equals(scrobble.album_id.xata_id))
-        .getFirst(),
-      ctx.client.db.user_artists
-        .select(["*"])
-        .filter("artist_id.xata_id", equals(scrobble.artist_id.xata_id))
-        .getFirst(),
-      ctx.client.db.user_tracks
-        .select(["*"])
-        .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
-        .getFirst(),
-      ctx.client.db.album_tracks
-        .select(["*"])
-        .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
-        .getFirst(),
-      ctx.client.db.artist_tracks
-        .select(["*"])
-        .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
-        .getFirst(),
-    ]);
+  const [
+    user_album,
+    user_artist,
+    user_track,
+    album_track,
+    artist_track,
+    artist_album,
+  ] = await Promise.all([
+    ctx.client.db.user_albums
+      .select(["*"])
+      .filter("album_id.xata_id", equals(scrobble.album_id.xata_id))
+      .getFirst(),
+    ctx.client.db.user_artists
+      .select(["*"])
+      .filter("artist_id.xata_id", equals(scrobble.artist_id.xata_id))
+      .getFirst(),
+    ctx.client.db.user_tracks
+      .select(["*"])
+      .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
+      .getFirst(),
+    ctx.client.db.album_tracks
+      .select(["*"])
+      .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
+      .getFirst(),
+    ctx.client.db.artist_tracks
+      .select(["*"])
+      .filter("track_id.xata_id", equals(scrobble.track_id.xata_id))
+      .getFirst(),
+    ctx.client.db.artist_albums
+      .select(["*"])
+      .filter("album_id.xata_id", equals(scrobble.album_id.xata_id))
+      .filter("artist_id.xata_id", equals(scrobble.artist_id.xata_id))
+      .getFirst(),
+  ]);
 
   const message = JSON.stringify({
     scrobble,
@@ -354,6 +365,7 @@ export async function publishScrobble(ctx: Context, id: string) {
     user_track,
     album_track,
     artist_track,
+    artist_album,
   });
 
   ctx.nc.publish("rocksky.scrobble", Buffer.from(message));
-- 
2.51.2