A decentralized music tracking and discovery platform built on AT Protocol 🎵
rocksky.app
spotify
atproto
lastfm
musicbrainz
scrobbling
listenbrainz
[scrobbler] add support for listebrainz API
+303
-6
+1
Cargo.lock
+1
Cargo.lock
+1
crates/scrobbler/Cargo.toml
+1
crates/scrobbler/Cargo.toml
+3
-3
crates/scrobbler/src/auth.rs
+3
-3
crates/scrobbler/src/auth.rs
+44
-1
crates/scrobbler/src/handlers/mod.rs
+44
-1
crates/scrobbler/src/handlers/mod.rs
···
1
-
use actix_web::{get, post, web, HttpResponse, Responder};
1
+
use actix_web::http::header;
2
+
use actix_web::{get, post, web, HttpRequest, HttpResponse, Responder};
2
3
use anyhow::Error;
3
4
use scrobble::handle_scrobble;
4
5
use sqlx::{Pool, Postgres};
···
7
8
use v1::submission::submission;
8
9
use std::collections::BTreeMap;
9
10
use std::sync::Arc;
11
+
use tokio_stream::StreamExt;
12
+
10
13
use crate::cache::Cache;
14
+
use crate::listenbrainz::submit::submit_listens;
15
+
use crate::listenbrainz::types::SubmitListensRequest;
11
16
use crate::BANNER;
12
17
13
18
pub mod scrobble;
14
19
pub mod v1;
20
+
21
+
#[macro_export]
22
+
macro_rules! read_payload {
23
+
($payload:expr) => {{
24
+
let mut body = Vec::new();
25
+
while let Some(chunk) = $payload.next().await {
26
+
match chunk {
27
+
Ok(bytes) => body.extend_from_slice(&bytes),
28
+
Err(err) => return Err(err.into()),
29
+
}
30
+
}
31
+
body
32
+
}};
33
+
}
15
34
16
35
17
36
#[get("/")]
···
79
98
let method = form.get("method").unwrap_or(&"".to_string()).to_string();
80
99
call_method(&method, conn, cache, form.into_inner()).await
81
100
.map_err(actix_web::error::ErrorInternalServerError)
101
+
}
102
+
103
+
#[post("/listenbrainz/1/submit-listens")]
104
+
pub async fn handle_submit_listens(
105
+
req: HttpRequest,
106
+
data: web::Data<Arc<Pool<Postgres>>>,
107
+
cache: web::Data<Cache>,
108
+
mut payload: web::Payload,
109
+
) -> impl Responder {
110
+
let token = match req.headers().get("Authorization") {
111
+
Some(header) => header.to_str().map_err(actix_web::error::ErrorBadRequest)?,
112
+
None => return Ok(HttpResponse::Unauthorized().finish()),
113
+
};
114
+
let token = token.trim_start_matches("Token ");
115
+
let token = token.trim_start_matches("Bearer ");
116
+
117
+
let payload = read_payload!(payload);
118
+
let body = String::from_utf8_lossy(&payload);
119
+
let req = serde_json::from_str::<SubmitListensRequest>(&body)
120
+
.map_err(actix_web::error::ErrorBadRequest)?;
121
+
122
+
submit_listens(req, cache.get_ref(), data.get_ref(), token)
123
+
.await
124
+
.map_err(actix_web::error::ErrorInternalServerError)
82
125
}
83
126
84
127
pub async fn call_method(
+37
crates/scrobbler/src/listenbrainz/submit.rs
+37
crates/scrobbler/src/listenbrainz/submit.rs
···
1
+
use std::sync::Arc;
2
+
use anyhow::Error;
3
+
use actix_web::HttpResponse;
4
+
use serde_json::json;
5
+
6
+
use crate::{cache::Cache, scrobbler::scrobble_listenbrainz};
7
+
8
+
use super::types::SubmitListensRequest;
9
+
10
+
pub async fn submit_listens(
11
+
payload: SubmitListensRequest,
12
+
cache: &Cache,
13
+
pool: &Arc<sqlx::Pool<sqlx::Postgres>>,
14
+
token: &str
15
+
) -> Result<HttpResponse, Error> {
16
+
match scrobble_listenbrainz(pool, cache, payload, token)
17
+
.await {
18
+
Ok(_) => Ok(HttpResponse::Ok().json(
19
+
json!({
20
+
"status": "ok",
21
+
"payload": {
22
+
"submitted_listens": 1,
23
+
"ignored_listens": 0
24
+
},
25
+
})
26
+
)),
27
+
Err(e) => {
28
+
println!("Error submitting listens: {}", e);
29
+
Ok(HttpResponse::BadRequest().json(
30
+
serde_json::json!({
31
+
"error": 4,
32
+
"message": format!("Failed to parse listens: {}", e)
33
+
})
34
+
))
35
+
}
36
+
}
37
+
}
+34
crates/scrobbler/src/listenbrainz/types.rs
+34
crates/scrobbler/src/listenbrainz/types.rs
···
1
+
use serde::Deserialize;
2
+
3
+
#[derive(Deserialize, Debug, Clone)]
4
+
pub struct AdditionalInfo {
5
+
pub release_name: Option<String>,
6
+
pub musicbrainz_artist_id: Option<String>,
7
+
pub musicbrainz_track_id: Option<String>,
8
+
pub duration_ms: Option<u64>,
9
+
pub media_player: Option<String>,
10
+
pub submission_client: Option<String>,
11
+
}
12
+
13
+
#[derive(Deserialize, Debug, Clone)]
14
+
pub struct TrackMetadata {
15
+
pub artist_name: String,
16
+
pub track_name: String,
17
+
pub release_name: Option<String>,
18
+
pub recording_mbid: Option<String>,
19
+
pub artist_mbid: Option<String>,
20
+
pub release_mbid: Option<String>,
21
+
pub additional_info: Option<AdditionalInfo>,
22
+
}
23
+
24
+
#[derive(Deserialize, Debug, Clone)]
25
+
pub struct ListenPayload {
26
+
pub track_metadata: TrackMetadata,
27
+
pub listened_at: u64,
28
+
}
29
+
30
+
#[derive(Deserialize, Debug, Clone)]
31
+
pub struct SubmitListensRequest {
32
+
pub listen_type: String,
33
+
pub payload: Vec<ListenPayload>,
34
+
}
+2
crates/scrobbler/src/main.rs
+2
crates/scrobbler/src/main.rs
···
12
12
pub mod rocksky;
13
13
pub mod repo;
14
14
pub mod types;
15
+
pub mod listenbrainz;
15
16
16
17
use actix_session::SessionExt as _;
17
18
use std::{env, sync::Arc, time::Duration};
···
75
76
.service(handlers::handle_methods)
76
77
.service(handlers::handle_nowplaying)
77
78
.service(handlers::handle_submission)
79
+
.service(handlers::handle_submit_listens)
78
80
.service(handlers::index)
79
81
.service(handlers::handle_get)
80
82
})
+17
-1
crates/scrobbler/src/repo/user.rs
+17
-1
crates/scrobbler/src/repo/user.rs
···
14
14
.fetch_all(pool)
15
15
.await?;
16
16
17
-
if results.len() == 0 {
17
+
if results.is_empty(){
18
18
return Ok(None);
19
19
}
20
20
21
21
Ok(Some(results[0].clone()))
22
22
}
23
23
24
+
25
+
pub async fn get_user_by_did(pool: &Pool<Postgres>, did: &str) -> Result<Option<User>, Error> {
26
+
let results: Vec<User> = sqlx::query_as(r#"
27
+
SELECT * FROM users
28
+
WHERE did = $1
29
+
"#)
30
+
.bind(did)
31
+
.fetch_all(pool)
32
+
.await?;
33
+
34
+
if results.is_empty() {
35
+
return Ok(None);
36
+
}
37
+
38
+
Ok(Some(results[0].clone()))
39
+
}
+162
-1
crates/scrobbler/src/scrobbler.rs
+162
-1
crates/scrobbler/src/scrobbler.rs
···
6
6
use sqlx::{Pool, Postgres};
7
7
8
8
use crate::{
9
-
auth::extract_did, cache::Cache, crypto::decrypt_aes_256_ctr, musicbrainz::client::MusicbrainzClient, repo, rocksky, spotify::{
9
+
auth::{decode_token, extract_did}, cache::Cache, crypto::decrypt_aes_256_ctr, listenbrainz::types::SubmitListensRequest, musicbrainz::client::MusicbrainzClient, repo, rocksky, spotify::{
10
10
client::SpotifyClient,
11
11
refresh_token
12
12
}, types::{Scrobble, Track}, xata::user::User
···
231
231
232
232
let user = user.unwrap();
233
233
let user = serde_json::from_str::<User>(&user)?;
234
+
235
+
let spofity_tokens = repo::spotify_token::get_spotify_tokens(pool, 100).await?;
236
+
237
+
if spofity_tokens.is_empty() {
238
+
return Err(Error::msg("No Spotify tokens found"));
239
+
}
240
+
241
+
let mb_client = MusicbrainzClient::new();
242
+
243
+
let mut scrobble = Scrobble {
244
+
artist: artist.trim().to_string(),
245
+
track: track.trim().to_string(),
246
+
timestamp: timestamp.parse::<u64>()?,
247
+
album: None,
248
+
context: None,
249
+
stream_id: None,
250
+
chosen_by_user: None,
251
+
track_number: None,
252
+
mbid: None,
253
+
album_artist: None,
254
+
duration: None,
255
+
ignored: None,
256
+
};
257
+
258
+
let did = user.did.clone();
259
+
260
+
/*
261
+
0. check if scrobble is cached
262
+
1. if mbid is present, check if it exists in the database
263
+
2. if it exists, scrobble
264
+
3. if it doesn't exist, check if it exists in Musicbrainz (using mbid)
265
+
4. if it exists, get album art from spotify and scrobble
266
+
5. if it doesn't exist, check if it exists in Spotify
267
+
6. if it exists, scrobble
268
+
7. if it doesn't exist, check if it exists in Musicbrainz (using track and artist)
269
+
8. if it exists, scrobble
270
+
9. if it doesn't exist, skip unknown track
271
+
*/
272
+
let key = format!("{} - {}", scrobble.artist.to_lowercase(), scrobble.track.to_lowercase());
273
+
let cached = cache.get(&key)?;
274
+
if cached.is_some() {
275
+
println!("{}", format!("Cached: {}", key).yellow());
276
+
let track = serde_json::from_str::<Track>(&cached.unwrap())?;
277
+
scrobble.album = Some(track.album.clone());
278
+
rocksky::scrobble(cache, &did, track, scrobble.timestamp).await?;
279
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
280
+
return Ok(());
281
+
}
282
+
283
+
if let Some(mbid) = &scrobble.mbid {
284
+
// let result = repo::track::get_track_by_mbid(pool, mbid).await?;
285
+
let result = mb_client.get_recording(mbid).await?;
286
+
println!("{}", "Musicbrainz (mbid)".yellow());
287
+
scrobble.album = Some(Track::from(result.clone()).album);
288
+
rocksky::scrobble(cache, &did, result.into(), scrobble.timestamp).await?;
289
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
290
+
return Ok(());
291
+
}
292
+
293
+
let result = repo::track::get_track(pool, &scrobble.track, &scrobble.artist).await?;
294
+
295
+
if let Some(track) = result {
296
+
println!("{}", "Xata (track)".yellow());
297
+
scrobble.album = Some(track.album.clone());
298
+
let album = repo::album::get_album_by_track_id(pool, &track.xata_id).await?;
299
+
let artist = repo::artist::get_artist_by_track_id(pool, &track.xata_id).await?;
300
+
let mut track: Track = track.into();
301
+
track.year = match album.year {
302
+
Some(year) => Some(year as u32),
303
+
None => match album.release_date.clone() {
304
+
Some(release_date) => {
305
+
let year = release_date.split("-").next();
306
+
year.and_then(|x| x.parse::<u32>().ok())
307
+
}
308
+
None => None,
309
+
},
310
+
};
311
+
track.release_date = album.release_date.map(|x| x.split("T").next().unwrap().to_string());
312
+
track.artist_picture = artist.picture.clone();
313
+
314
+
rocksky::scrobble(cache, &did, track, scrobble.timestamp).await?;
315
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
316
+
return Ok(());
317
+
}
318
+
319
+
// we need to pick a random token to avoid Spotify rate limiting
320
+
// and to avoid using the same token for all scrobbles
321
+
// this is a simple way to do it, but we can improve it later
322
+
// by using a more sophisticated algorithm
323
+
// or by using a token pool
324
+
let mut rng = rand::rng();
325
+
let random_index = rng.random_range(0..spofity_tokens.len());
326
+
let spotify_token = &spofity_tokens[random_index];
327
+
328
+
let spotify_token = decrypt_aes_256_ctr(
329
+
&spotify_token.refresh_token,
330
+
&hex::decode(env::var("SPOTIFY_ENCRYPTION_KEY")?)?
331
+
)?;
332
+
333
+
let spotify_token = refresh_token(&spotify_token).await?;
334
+
let spotify_client = SpotifyClient::new(&spotify_token.access_token);
335
+
336
+
let result = spotify_client.search(&format!(r#"track:"{}" artist:"{}""#, scrobble.track, scrobble.artist)).await?;
337
+
338
+
if let Some(track) = result.tracks.items.first() {
339
+
println!("{}", "Spotify (track)".yellow());
340
+
scrobble.album = Some(track.album.name.clone());
341
+
let mut track = track.clone();
342
+
343
+
if let Some(album) = spotify_client.get_album(&track.album.id).await? {
344
+
track.album = album;
345
+
}
346
+
347
+
if let Some(artist) = spotify_client.get_artist(&track.album.artists[0].id).await? {
348
+
track.album.artists[0] = artist;
349
+
}
350
+
351
+
rocksky::scrobble(cache, &did, track.into(), scrobble.timestamp).await?;
352
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
353
+
return Ok(());
354
+
}
355
+
356
+
let query = format!(
357
+
r#"recording:"{}" AND artist:"{}""#,
358
+
scrobble.track, scrobble.artist
359
+
);
360
+
let result = mb_client.search(&query).await?;
361
+
362
+
if let Some(recording) = result.recordings.first() {
363
+
let result = mb_client.get_recording(&recording.id).await?;
364
+
println!("{}", "Musicbrainz (recording)".yellow());
365
+
scrobble.album = Some(Track::from(result.clone()).album);
366
+
rocksky::scrobble(cache, &did, result.into(), scrobble.timestamp).await?;
367
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
368
+
return Ok(());
369
+
}
370
+
371
+
println!("{} {} - {}, skipping", "Track not found: ".yellow(), artist, track);
372
+
373
+
Ok(())
374
+
}
375
+
376
+
pub async fn scrobble_listenbrainz(pool: &Pool<Postgres>, cache: &Cache, req: SubmitListensRequest, token: &str) -> Result<(), Error> {
377
+
if req.payload.is_empty() {
378
+
return Err(Error::msg("No payload found"));
379
+
}
380
+
381
+
let artist = req.payload[0].track_metadata.artist_name.clone();
382
+
let track = req.payload[0].track_metadata.track_name.clone();
383
+
let timestamp = req.payload[0].listened_at.to_string();
384
+
385
+
let claims = decode_token(token)?;
386
+
let did = claims.did.clone();
387
+
let user = repo::user::get_user_by_did(pool, &did)
388
+
.await?;
389
+
390
+
if user.is_none() {
391
+
return Err(Error::msg("User not found"));
392
+
}
393
+
394
+
let user = user.unwrap();
234
395
235
396
let spofity_tokens = repo::spotify_token::get_spotify_tokens(pool, 100).await?;
236
397