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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181import chalk from "chalk";import { consola } from "consola";import { ctx } from "context";import { desc, eq, or } from "drizzle-orm";import { createHash } from "node:crypto";import { publishScrobble } from "nowplaying/nowplaying.service";import albums from "../schema/albums";import artists from "../schema/artists";import scrobbles from "../schema/scrobbles";import tracks from "../schema/tracks";import users from "../schema/users";
const args = process.argv.slice(2);
export async function updateUris(did: string) { // Get scrobbles with track and user data const records = await ctx.db .select({ track: tracks, user: users, }) .from(scrobbles) .innerJoin(tracks, eq(scrobbles.trackId, tracks.id)) .innerJoin(users, eq(scrobbles.userId, users.id)) .where(or(eq(users.did, did), eq(users.handle, did))) .orderBy(desc(scrobbles.createdAt)) .limit(process.env.SYNC_SIZE ? parseInt(process.env.SYNC_SIZE, 10) : 20);
for (const { track } of records) { const trackHash = createHash("sha256") .update(`${track.title} - ${track.artist} - ${track.album}`.toLowerCase()) .digest("hex");
const existingTrack = await ctx.db .select() .from(tracks) .where(eq(tracks.sha256, trackHash)) .limit(1) .then((rows) => rows[0]);
if (existingTrack && !existingTrack.albumUri) { consola.info(`Updating album uri for ${chalk.cyan(track.id)} ...`);
const albumHash = createHash("sha256") .update(`${track.album} - ${track.albumArtist}`.toLowerCase()) .digest("hex");
const album = await ctx.db .select() .from(albums) .where(eq(albums.sha256, albumHash)) .limit(1) .then((rows) => rows[0]);
if (album) { await ctx.db .update(tracks) .set({ albumUri: album.uri }) .where(eq(tracks.id, existingTrack.id)); } }
if (existingTrack && !existingTrack.artistUri) { consola.info(`Updating artist uri for ${chalk.cyan(track.id)} ...`);
const artistHash = createHash("sha256") .update(track.albumArtist.toLowerCase()) .digest("hex");
const artist = await ctx.db .select() .from(artists) .where(eq(artists.sha256, artistHash)) .limit(1) .then((rows) => rows[0]);
if (artist) { await ctx.db .update(tracks) .set({ artistUri: artist.uri }) .where(eq(tracks.id, existingTrack.id)); } }
const albumHash = createHash("sha256") .update(`${track.album} - ${track.albumArtist}`.toLowerCase()) .digest("hex");
const album = await ctx.db .select() .from(albums) .where(eq(albums.sha256, albumHash)) .limit(1) .then((rows) => rows[0]);
if (existingTrack && album && !album.artistUri) { consola.info(`Updating artist uri for ${chalk.cyan(album.id)} ...`);
const artistHash = createHash("sha256") .update(track.albumArtist.toLowerCase()) .digest("hex");
const artist = await ctx.db .select() .from(artists) .where(eq(artists.sha256, artistHash)) .limit(1) .then((rows) => rows[0]);
if (artist) { await ctx.db .update(albums) .set({ artistUri: artist.uri }) .where(eq(albums.id, album.id)); } } }}
if (args.includes("--background")) { consola.info("Wait for new scrobbles to sync ..."); const sub = ctx.nc.subscribe("rocksky.user.scrobble.sync"); for await (const m of sub) { const did = new TextDecoder().decode(m.data); // wait for 15 seconds to ensure the scrobble is fully created await new Promise((resolve) => setTimeout(resolve, 15000)); consola.info(`Syncing scrobbles ${chalk.magenta(did)} ...`); await updateUris(did);
const records = await ctx.db .select({ scrobble: scrobbles, }) .from(scrobbles) .innerJoin(users, eq(scrobbles.userId, users.id)) .where(or(eq(users.did, did), eq(users.handle, did))) .orderBy(desc(scrobbles.createdAt)) .limit(5);
for (const { scrobble } of records) { consola.info(`Syncing scrobble ${chalk.cyan(scrobble.id)} ...`); try { await publishScrobble(ctx, scrobble.id); } catch (err) { consola.error( `Failed to sync scrobble ${chalk.cyan(scrobble.id)}:`, err, ); } } } process.exit(0);}
for (const arg of args) { consola.info(`Syncing scrobbles ${chalk.magenta(arg)} ...`); await updateUris(arg);
const records = await ctx.db .select({ scrobble: scrobbles, }) .from(scrobbles) .innerJoin(users, eq(scrobbles.userId, users.id)) .where(or(eq(users.did, arg), eq(users.handle, arg))) .orderBy(desc(scrobbles.createdAt)) .limit(process.env.SYNC_SIZE ? parseInt(process.env.SYNC_SIZE, 10) : 20);
for (const { scrobble } of records) { consola.info(`Syncing scrobble ${chalk.cyan(scrobble.id)} ...`); try { await publishScrobble(ctx, scrobble.id); } catch (err) { consola.error(`Failed to sync scrobble ${chalk.cyan(scrobble.id)}:`, err); } } consola.info(`Synced ${chalk.greenBright(records.length)} scrobbles`);}
process.exit(0);