Something went wrong. Try again.
A decentralized music tracking and discovery platform built on AT Protocol 🎵 rocksky.app
spotify atproto lastfm musicbrainz scrobbling listenbrainz
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411import { MatchTrackResult } from "lib/matchTrack";import { logger } from "logger";import dayjs from "dayjs";import { createAgent } from "lib/agent";import { getDidAndHandle } from "lib/getDidAndHandle";import { ctx } from "context";import schema from "schema";import { and, eq, gte, lte, or, sql } from "drizzle-orm";import os from "node:os";import path from "node:path";import fs from "node:fs";import { c } from "theme";import * as Album from "lexicon/types/app/rocksky/album";import * as Artist from "lexicon/types/app/rocksky/artist";import * as Scrobble from "lexicon/types/app/rocksky/scrobble";import * as Song from "lexicon/types/app/rocksky/song";import { TID } from "@atproto/common";import { Agent } from "@atproto/api";import { createUser, subscribeToJetstream, sync } from "cmd/sync";import _ from "lodash";
export async function publishScrobble( track: MatchTrackResult, timestamp?: number, dryRun?: boolean,) { const [did, handle] = await getDidAndHandle(); const agent: Agent = await createAgent(did, handle); const recentScrobble = await getRecentScrobble(did, track, timestamp); const user = await createUser(agent, did, handle); await subscribeToJetstream(user);
const lockFilePath = path.join(os.tmpdir(), `rocksky-${did}.lock`);
if (fs.existsSync(lockFilePath)) { logger.error( `${c.primary(handle)} Scrobble publishing failed: lock file exists, maybe rocksky-cli is still syncing?\nPlease wait for rocksky to finish syncing before publishing scrobbles or delete the lock file manually ${c.primary(lockFilePath)}`, ); return false; }
if (recentScrobble) { logger.info`${handle} Skipping scrobble for ${track.title} by ${track.artist} at ${timestamp ? dayjs.unix(timestamp).format("YYYY-MM-DD HH:mm:ss") : dayjs().format("YYYY-MM-DD HH:mm:ss")} (already scrobbled)`; return true; }
const totalScrobbles = await countScrobbles(did); if (totalScrobbles === 0) { logger.warn`${handle} No scrobbles found for this user. Are you sure you have successfully synced your scrobbles locally?\nIf not, please run ${"rocksky sync"} to sync your scrobbles before publishing scrobbles.`; }
logger.info`${handle} Publishing scrobble for ${track.title} by ${track.artist} at ${timestamp ? dayjs.unix(timestamp).format("YYYY-MM-DD HH:mm:ss") : dayjs().format("YYYY-MM-DD HH:mm:ss")}`;
if (await shouldSync(agent)) { logger.info`${handle} Syncing scrobbles before publishing`; await sync(); } else { logger.info`${handle} Local scrobbles are up-to-date, skipping sync`; }
if (dryRun) { logger.info`${handle} Dry run: Skipping publishing scrobble for ${track.title} by ${track.artist} at ${timestamp ? dayjs.unix(timestamp).format("YYYY-MM-DD HH:mm:ss") : dayjs().format("YYYY-MM-DD HH:mm:ss")}`; return true; }
const existingTrack = await ctx.db .select() .from(schema.tracks) .where( or( and( sql`LOWER(${schema.tracks.title}) = LOWER(${track.title})`, sql`LOWER(${schema.tracks.artist}) = LOWER(${track.artist})`, ), and( sql`LOWER(${schema.tracks.title}) = LOWER(${track.title})`, sql`LOWER(${schema.tracks.albumArtist}) = LOWER(${track.artist})`, ), and( sql`LOWER(${schema.tracks.title}) = LOWER(${track.title})`, sql`LOWER(${schema.tracks.albumArtist}) = LOWER(${track.albumArtist})`, ), ), ) .limit(1) .execute() .then((rows) => rows[0]);
if (!existingTrack) { await putSongRecord(agent, track); }
const existingArtist = await ctx.db .select() .from(schema.artists) .where( or( sql`LOWER(${schema.artists.name}) = LOWER(${track.artist})`, sql`LOWER(${schema.artists.name}) = LOWER(${track.albumArtist})`, ), ) .limit(1) .execute() .then((rows) => rows[0]);
if (!existingArtist) { await putArtistRecord(agent, track); }
const existingAlbum = await ctx.db .select() .from(schema.albums) .where( and( sql`LOWER(${schema.albums.title}) = LOWER(${track.album})`, sql`LOWER(${schema.albums.artist}) = LOWER(${track.albumArtist})`, ), ) .limit(1) .execute() .then((rows) => rows[0]);
if (!existingAlbum) { await putAlbumRecord(agent, track); }
const scrobbleUri = await putScrobbleRecord(agent, track, timestamp);
// wait for the scrobble to be published if (scrobbleUri) { const MAX_ATTEMPTS = 40; let attempts = 0; do { const count = await ctx.db .select({ count: sql`COUNT(*)`, }) .from(schema.scrobbles) .where(eq(schema.scrobbles.uri, scrobbleUri)) .execute() .then((rows) => _.get(rows, "[0].count", 0) as number);
if (count > 0 || attempts >= MAX_ATTEMPTS) { if (attempts == MAX_ATTEMPTS) { logger.error`Failed to detect published scrobble after ${MAX_ATTEMPTS} attempts`; } break; }
await new Promise((resolve) => setTimeout(resolve, 600)); attempts += 1; } while (true); }
return true;}
async function getRecentScrobble( did: string, track: MatchTrackResult, timestamp?: number,) { const scrobbleTime = dayjs.unix(timestamp || dayjs().unix()); return ctx.db .select({ scrobble: schema.scrobbles, user: schema.users, track: schema.tracks, }) .from(schema.scrobbles) .innerJoin(schema.users, eq(schema.scrobbles.userId, schema.users.id)) .innerJoin(schema.tracks, eq(schema.scrobbles.trackId, schema.tracks.id)) .where( and( eq(schema.users.did, did), sql`LOWER(${schema.tracks.title}) = LOWER(${track.title})`, sql`LOWER(${schema.tracks.artist}) = LOWER(${track.artist})`, gte( schema.scrobbles.timestamp, scrobbleTime.subtract(60, "seconds").toDate(), ), lte( schema.scrobbles.timestamp, scrobbleTime.add(60, "seconds").toDate(), ), ), ) .limit(1) .then((rows) => rows[0]);}
async function countScrobbles(did: string): Promise<number> { return ctx.db .select({ count: sql<number>`count(*)` }) .from(schema.scrobbles) .innerJoin(schema.users, eq(schema.scrobbles.userId, schema.users.id)) .where(eq(schema.users.did, did)) .then((rows) => rows[0].count);}
async function putSongRecord(agent: Agent, track: MatchTrackResult) { const rkey = TID.nextStr();
const record: Song.Record = { $type: "app.rocksky.song", title: track.title, artist: track.artist, artists: track.mbArtists === null ? undefined : track.mbArtists, album: track.album, albumArtist: track.albumArtist, duration: track.duration, releaseDate: track.releaseDate ? new Date(track.releaseDate).toISOString() : undefined, year: track.year === null ? undefined : track.year, albumArtUrl: track.albumArt, composer: track.composer ? track.composer : undefined, lyrics: track.lyrics ? track.lyrics : undefined, trackNumber: track.trackNumber, discNumber: track.discNumber === 0 ? 1 : track.discNumber, copyrightMessage: track.copyrightMessage ? track.copyrightMessage : undefined, createdAt: new Date().toISOString(), spotifyLink: track.spotifyLink ? track.spotifyLink : undefined, tags: track.genres || [], mbid: track.mbId, };
if (!Song.validateRecord(record).success) { logger.info`${Song.validateRecord(record)}`; logger.info`${record}`; throw new Error("Invalid Song record"); }
try { const res = await agent.com.atproto.repo.putRecord({ repo: agent.assertDid, collection: "app.rocksky.song", rkey, record, validate: false, }); const uri = res.data.uri; logger.info`Song record created at ${uri}`; return uri; } catch (e) { logger.error`Error creating song record: ${e}`; return null; }}
async function putArtistRecord(agent: Agent, track: MatchTrackResult) { const rkey = TID.nextStr(); const record: Artist.Record = { $type: "app.rocksky.artist", name: track.albumArtist, createdAt: new Date().toISOString(), pictureUrl: track.artistPicture || undefined, tags: track.genres || [], };
if (!Artist.validateRecord(record).success) { logger.info`${Artist.validateRecord(record)}`; logger.info`${record}`; throw new Error("Invalid Artist record"); }
try { const res = await agent.com.atproto.repo.putRecord({ repo: agent.assertDid, collection: "app.rocksky.artist", rkey, record, validate: false, }); const uri = res.data.uri; logger.info`Artist record created at ${uri}`; return uri; } catch (e) { logger.error`Error creating artist record: ${e}`; return null; }}
async function putAlbumRecord(agent: Agent, track: MatchTrackResult) { const rkey = TID.nextStr();
const record = { $type: "app.rocksky.album", title: track.album, artist: track.albumArtist, year: track.year === null ? undefined : track.year, releaseDate: track.releaseDate ? new Date(track.releaseDate).toISOString() : undefined, createdAt: new Date().toISOString(), albumArtUrl: track.albumArt, };
if (!Album.validateRecord(record).success) { logger.info`${Album.validateRecord(record)}`; logger.info`${record}`; throw new Error("Invalid Album record"); }
try { const res = await agent.com.atproto.repo.putRecord({ repo: agent.assertDid, collection: "app.rocksky.album", rkey, record, validate: false, }); const uri = res.data.uri; logger.info`Album record created at ${uri}`; return uri; } catch (e) { logger.error`Error creating album record: ${e}`; return null; }}
async function putScrobbleRecord( agent: Agent, track: MatchTrackResult, timestamp?: number,) { const rkey = TID.nextStr();
const record: Scrobble.Record = { $type: "app.rocksky.scrobble", title: track.title, albumArtist: track.albumArtist, albumArtUrl: track.albumArt, artist: track.artist, artists: track.mbArtists === null ? undefined : track.mbArtists, album: track.album, duration: track.duration, trackNumber: track.trackNumber, discNumber: track.discNumber === 0 ? 1 : track.discNumber, releaseDate: track.releaseDate ? new Date(track.releaseDate).toISOString() : undefined, year: track.year === null ? undefined : track.year, composer: track.composer ? track.composer : undefined, lyrics: track.lyrics ? track.lyrics : undefined, copyrightMessage: track.copyrightMessage ? track.copyrightMessage : undefined, createdAt: timestamp ? dayjs.unix(timestamp).toISOString() : new Date().toISOString(), spotifyLink: track.spotifyLink ? track.spotifyLink : undefined, tags: track.genres || [], mbid: track.mbId, };
if (!Scrobble.validateRecord(record).success) { logger.info`${Scrobble.validateRecord(record)}`; logger.info`${record}`; throw new Error("Invalid Scrobble record"); }
try { const res = await agent.com.atproto.repo.putRecord({ repo: agent.assertDid, collection: "app.rocksky.scrobble", rkey, record, validate: false, }); const uri = res.data.uri; logger.info`Scrobble record created at ${uri}`; return uri; } catch (e) { logger.error`Error creating scrobble record: ${e}`; return null; }}
async function shouldSync(agent: Agent): Promise<boolean> { const res = await agent.com.atproto.repo.listRecords({ repo: agent.assertDid, collection: "app.rocksky.scrobble", limit: 1, });
const records = res.data.records as Array<{ uri: string; cid: string; value: Scrobble.Record; }>;
if (!records.length) { logger.info`No scrobble records found`; return true; }
const { count } = await ctx.db .select({ count: sql<number>`count(*)`, }) .from(schema.scrobbles) .where(eq(schema.scrobbles.cid, records[0].cid)) .execute() .then((result) => result[0]);
return count === 0;}