diff --git a/apps/cli/drizzle/0000_amazing_redwing.sql b/apps/cli/drizzle/0000_parched_robbie_robertson.sql similarity index 81% rename from apps/cli/drizzle/0000_amazing_redwing.sql rename to apps/cli/drizzle/0000_parched_robbie_robertson.sql index 673baa60..e759dbc6 100644 --- a/apps/cli/drizzle/0000_amazing_redwing.sql +++ b/apps/cli/drizzle/0000_parched_robbie_robertson.sql @@ -16,22 +16,22 @@ CREATE TABLE `albums` ( `year` integer, `album_art` text, `uri` text, + `cid` text NOT NULL, `artist_uri` text, `apple_music_link` text, `spotify_link` text, `tidal_link` text, `youtube_link` text, - `sha256` text NOT NULL, `created_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, `updated_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL ); --> statement-breakpoint CREATE UNIQUE INDEX `albums_uri_unique` ON `albums` (`uri`);--> statement-breakpoint +CREATE UNIQUE INDEX `albums_cid_unique` ON `albums` (`cid`);--> statement-breakpoint CREATE UNIQUE INDEX `albums_apple_music_link_unique` ON `albums` (`apple_music_link`);--> statement-breakpoint CREATE UNIQUE INDEX `albums_spotify_link_unique` ON `albums` (`spotify_link`);--> statement-breakpoint CREATE UNIQUE INDEX `albums_tidal_link_unique` ON `albums` (`tidal_link`);--> statement-breakpoint CREATE UNIQUE INDEX `albums_youtube_link_unique` ON `albums` (`youtube_link`);--> statement-breakpoint -CREATE UNIQUE INDEX `albums_sha256_unique` ON `albums` (`sha256`);--> statement-breakpoint CREATE TABLE `artist_albums` ( `id` text PRIMARY KEY NOT NULL, `artist_id` text NOT NULL, @@ -60,8 +60,8 @@ CREATE TABLE `artists` ( `born_in` text, `died` integer, `picture` text, - `sha256` text NOT NULL, `uri` text, + `cid` text NOT NULL, `apple_music_link` text, `spotify_link` text, `tidal_link` text, @@ -71,8 +71,15 @@ CREATE TABLE `artists` ( `updated_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL ); --> statement-breakpoint -CREATE UNIQUE INDEX `artists_sha256_unique` ON `artists` (`sha256`);--> statement-breakpoint CREATE UNIQUE INDEX `artists_uri_unique` ON `artists` (`uri`);--> statement-breakpoint +CREATE UNIQUE INDEX `artists_cid_unique` ON `artists` (`cid`);--> statement-breakpoint +CREATE TABLE `auth_sessions` ( + `key` text PRIMARY KEY NOT NULL, + `session` text NOT NULL, + `created_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, + `updated_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL +); +--> statement-breakpoint CREATE TABLE `loved_tracks` ( `id` text PRIMARY KEY NOT NULL, `user_id` text NOT NULL, @@ -84,6 +91,25 @@ CREATE TABLE `loved_tracks` ( ); --> statement-breakpoint CREATE UNIQUE INDEX `loved_tracks_uri_unique` ON `loved_tracks` (`uri`);--> statement-breakpoint +CREATE TABLE `scrobbles` ( + `xata_id` text PRIMARY KEY NOT NULL, + `user_id` text, + `track_id` text, + `album_id` text, + `artist_id` text, + `uri` text, + `cid` text, + `created_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, + `updated_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, + `timestamp` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, + FOREIGN KEY (`user_id`) REFERENCES `users`(`id`) ON UPDATE no action ON DELETE no action, + FOREIGN KEY (`track_id`) REFERENCES `tracks`(`id`) ON UPDATE no action ON DELETE no action, + FOREIGN KEY (`album_id`) REFERENCES `albums`(`id`) ON UPDATE no action ON DELETE no action, + FOREIGN KEY (`artist_id`) REFERENCES `artists`(`id`) ON UPDATE no action ON DELETE no action +); +--> statement-breakpoint +CREATE UNIQUE INDEX `scrobbles_uri_unique` ON `scrobbles` (`uri`);--> statement-breakpoint +CREATE UNIQUE INDEX `scrobbles_cid_unique` ON `scrobbles` (`cid`);--> statement-breakpoint CREATE TABLE `tracks` ( `id` text PRIMARY KEY NOT NULL, `title` text NOT NULL, @@ -98,7 +124,6 @@ CREATE TABLE `tracks` ( `spotify_link` text, `apple_music_link` text, `tidal_link` text, - `sha256` text NOT NULL, `disc_number` integer, `lyrics` text, `composer` text, @@ -106,6 +131,7 @@ CREATE TABLE `tracks` ( `label` text, `copyright_message` text, `uri` text, + `cid` text NOT NULL, `album_uri` text, `artist_uri` text, `created_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL, @@ -117,8 +143,8 @@ CREATE UNIQUE INDEX `tracks_youtube_link_unique` ON `tracks` (`youtube_link`);-- CREATE UNIQUE INDEX `tracks_spotify_link_unique` ON `tracks` (`spotify_link`);--> statement-breakpoint CREATE UNIQUE INDEX `tracks_apple_music_link_unique` ON `tracks` (`apple_music_link`);--> statement-breakpoint CREATE UNIQUE INDEX `tracks_tidal_link_unique` ON `tracks` (`tidal_link`);--> statement-breakpoint -CREATE UNIQUE INDEX `tracks_sha256_unique` ON `tracks` (`sha256`);--> statement-breakpoint CREATE UNIQUE INDEX `tracks_uri_unique` ON `tracks` (`uri`);--> statement-breakpoint +CREATE UNIQUE INDEX `tracks_cid_unique` ON `tracks` (`cid`);--> statement-breakpoint CREATE TABLE `user_albums` ( `id` text PRIMARY KEY NOT NULL, `user_id` text NOT NULL, diff --git a/apps/cli/drizzle/meta/0000_snapshot.json b/apps/cli/drizzle/meta/0000_snapshot.json index 3dae2e64..bacb6ec3 100644 --- a/apps/cli/drizzle/meta/0000_snapshot.json +++ b/apps/cli/drizzle/meta/0000_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "a549f070-9d4d-4d38-bd6a-c04bc9a4889b", + "id": "571a287d-ea60-4ac4-847a-307da78c375c", "prevId": "00000000-0000-0000-0000-000000000000", "tables": { "album_tracks": { @@ -130,6 +130,13 @@ "notNull": false, "autoincrement": false }, + "cid": { + "name": "cid", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, "artist_uri": { "name": "artist_uri", "type": "text", @@ -165,13 +172,6 @@ "notNull": false, "autoincrement": false }, - "sha256": { - "name": "sha256", - "type": "text", - "primaryKey": false, - "notNull": true, - "autoincrement": false - }, "created_at": { "name": "created_at", "type": "integer", @@ -197,6 +197,13 @@ ], "isUnique": true }, + "albums_cid_unique": { + "name": "albums_cid_unique", + "columns": [ + "cid" + ], + "isUnique": true + }, "albums_apple_music_link_unique": { "name": "albums_apple_music_link_unique", "columns": [ @@ -224,13 +231,6 @@ "youtube_link" ], "isUnique": true - }, - "albums_sha256_unique": { - "name": "albums_sha256_unique", - "columns": [ - "sha256" - ], - "isUnique": true } }, "foreignKeys": {}, @@ -438,18 +438,18 @@ "notNull": false, "autoincrement": false }, - "sha256": { - "name": "sha256", + "uri": { + "name": "uri", "type": "text", "primaryKey": false, - "notNull": true, + "notNull": false, "autoincrement": false }, - "uri": { - "name": "uri", + "cid": { + "name": "cid", "type": "text", "primaryKey": false, - "notNull": false, + "notNull": true, "autoincrement": false }, "apple_music_link": { @@ -505,17 +505,17 @@ } }, "indexes": { - "artists_sha256_unique": { - "name": "artists_sha256_unique", + "artists_uri_unique": { + "name": "artists_uri_unique", "columns": [ - "sha256" + "uri" ], "isUnique": true }, - "artists_uri_unique": { - "name": "artists_uri_unique", + "artists_cid_unique": { + "name": "artists_cid_unique", "columns": [ - "uri" + "cid" ], "isUnique": true } @@ -525,6 +525,46 @@ "uniqueConstraints": {}, "checkConstraints": {} }, + "auth_sessions": { + "name": "auth_sessions", + "columns": { + "key": { + "name": "key", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "session": { + "name": "session", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, "loved_tracks": { "name": "loved_tracks", "columns": { @@ -606,6 +646,157 @@ "uniqueConstraints": {}, "checkConstraints": {} }, + "scrobbles": { + "name": "scrobbles", + "columns": { + "xata_id": { + "name": "xata_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "track_id": { + "name": "track_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "album_id": { + "name": "album_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "artist_id": { + "name": "artist_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "uri": { + "name": "uri", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cid": { + "name": "cid", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP" + }, + "timestamp": { + "name": "timestamp", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "CURRENT_TIMESTAMP" + } + }, + "indexes": { + "scrobbles_uri_unique": { + "name": "scrobbles_uri_unique", + "columns": [ + "uri" + ], + "isUnique": true + }, + "scrobbles_cid_unique": { + "name": "scrobbles_cid_unique", + "columns": [ + "cid" + ], + "isUnique": true + } + }, + "foreignKeys": { + "scrobbles_user_id_users_id_fk": { + "name": "scrobbles_user_id_users_id_fk", + "tableFrom": "scrobbles", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "scrobbles_track_id_tracks_id_fk": { + "name": "scrobbles_track_id_tracks_id_fk", + "tableFrom": "scrobbles", + "tableTo": "tracks", + "columnsFrom": [ + "track_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "scrobbles_album_id_albums_id_fk": { + "name": "scrobbles_album_id_albums_id_fk", + "tableFrom": "scrobbles", + "tableTo": "albums", + "columnsFrom": [ + "album_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "scrobbles_artist_id_artists_id_fk": { + "name": "scrobbles_artist_id_artists_id_fk", + "tableFrom": "scrobbles", + "tableTo": "artists", + "columnsFrom": [ + "artist_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, "tracks": { "name": "tracks", "columns": { @@ -700,13 +891,6 @@ "notNull": false, "autoincrement": false }, - "sha256": { - "name": "sha256", - "type": "text", - "primaryKey": false, - "notNull": true, - "autoincrement": false - }, "disc_number": { "name": "disc_number", "type": "integer", @@ -756,6 +940,13 @@ "notNull": false, "autoincrement": false }, + "cid": { + "name": "cid", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, "album_uri": { "name": "album_uri", "type": "text", @@ -823,17 +1014,17 @@ ], "isUnique": true }, - "tracks_sha256_unique": { - "name": "tracks_sha256_unique", + "tracks_uri_unique": { + "name": "tracks_uri_unique", "columns": [ - "sha256" + "uri" ], "isUnique": true }, - "tracks_uri_unique": { - "name": "tracks_uri_unique", + "tracks_cid_unique": { + "name": "tracks_cid_unique", "columns": [ - "uri" + "cid" ], "isUnique": true } diff --git a/apps/cli/drizzle/meta/_journal.json b/apps/cli/drizzle/meta/_journal.json index 56e1195f..d13f04ed 100644 --- a/apps/cli/drizzle/meta/_journal.json +++ b/apps/cli/drizzle/meta/_journal.json @@ -5,8 +5,8 @@ { "idx": 0, "version": "6", - "when": 1767904137315, - "tag": "0000_amazing_redwing", + "when": 1768034891092, + "tag": "0000_parched_robbie_robertson", "breakpoints": true } ] diff --git a/apps/cli/package.json b/apps/cli/package.json index 0836316c..1130d351 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -41,14 +41,19 @@ "commander": "^13.1.0", "cors": "^2.8.5", "dayjs": "^1.11.13", + "dotenv": "^16.4.7", "drizzle-kit": "^0.31.1", "drizzle-orm": "^0.45.1", "effect": "^3.19.14", "env-paths": "^3.0.0", + "envalid": "^8.0.0", "express": "^5.1.0", + "kysely": "^0.27.5", + "lodash": "^4.17.21", "md5": "^2.3.0", "open": "^10.1.0", "table": "^6.9.0", + "unstorage": "^1.14.4", "zod": "^3.24.3" }, "devDependencies": { diff --git a/apps/cli/src/cmd/sync.ts b/apps/cli/src/cmd/sync.ts index ef2f9aaa..7260c5d6 100644 --- a/apps/cli/src/cmd/sync.ts +++ b/apps/cli/src/cmd/sync.ts @@ -1,11 +1,55 @@ import { JetStreamClient, JetStreamEvent } from "jetstream"; -import chalk from "chalk"; import { logger } from "logger"; +import { ctx } from "context"; +import { isValidHandle } from "@atproto/syntax"; +import { Agent } from "@atproto/api"; +import { env } from "lib/env"; +import { createAgent } from "lib/agent"; +import chalk from "chalk"; +import * as Artist from "lexicon/types/app/rocksky/artist"; +import * as Album from "lexicon/types/app/rocksky/album"; +import * as Song from "lexicon/types/app/rocksky/song"; +import * as Scrobble from "lexicon/types/app/rocksky/scrobble"; +import { SelectUser } from "schema/users"; +import schema from "schema"; +import { createId } from "@paralleldrive/cuid2"; +import _ from "lodash"; + +type Artists = { value: Artist.Record; uri: string; cid: string }[]; +type Albums = { value: Album.Record; uri: string; cid: string }[]; +type Songs = { value: Song.Record; uri: string; cid: string }[]; +type Scrobbles = { value: Scrobble.Record; uri: string; cid: string }[]; + +export async function sync() { + const [did, handle] = await getDidAndHandle(); + const agent: Agent = await createAgent(did, handle); + + const user = await createUser(agent, did, handle); + subscribeToJetstream(did); + + logger.info` DID: ${did}`; + logger.info` Handle: ${handle}`; + + const [artists, albums, songs, scrobbles] = await Promise.all([ + getRockskyUserArtists(agent), + getRockskyUserAlbums(agent), + getRockskyUserSongs(agent), + getRockskyUserScrobbles(agent), + ]); + + logger.info` Artists: ${artists.length}`; + logger.info` Albums: ${albums.length}`; + logger.info` Songs: ${songs.length}`; + logger.info` Scrobbles: ${scrobbles.length}`; + + await createArtists(artists, user); + await createAlbums(albums, user); + await createSongs(songs, user); + await createScrobbles(scrobbles, user); +} const getEndpoint = () => { - const endpoint = process.env.JETSTREAM_SERVER - ? process.env.JETSTREAM_SERVER - : "wss://jetstream1.us-west.bsky.network/subscribe"; + const endpoint = env.JETSTREAM_SERVER; if (endpoint?.endsWith("/subscribe")) { return endpoint; @@ -14,7 +58,183 @@ const getEndpoint = () => { return `${endpoint}/subscribe`; }; -export function sync() { +const getDidAndHandle = async (): Promise<[string, string]> => { + let handle = env.ROCKSKY_HANDLE || env.ROCKSKY_IDENTIFIER; + let did = env.ROCKSKY_HANDLE || env.ROCKSKY_IDENTIFIER; + + if (handle.startsWith("did:plc:") || handle.startsWith("did:web:")) { + handle = await ctx.resolver.resolveDidToHandle(handle); + } + + if (!isValidHandle(handle)) { + logger.error`❌ Invalid handle: ${handle}`; + process.exit(1); + } + + if (!did.startsWith("did:plc:") && !did.startsWith("did:web:")) { + did = await ctx.baseIdResolver.handle.resolve(did); + } + + return [did, handle]; +}; + +const createUser = async ( + agent: Agent, + did: string, + handle: string, +): Promise => { + const { data: profileRecord } = await agent.com.atproto.repo.getRecord({ + repo: agent.assertDid, + collection: "app.bsky.actor.profile", + rkey: "self", + }); + + const displayName = _.get(profileRecord, "value.displayName") as + | string + | undefined; + const avatar = `https://cdn.bsky.app/img/avatar/plain/${did}/${_.get(profileRecord, "value.avatar.ref", "").toString()}@jpeg`; + + const [user] = await ctx.db + .insert(schema.users) + .values({ + id: createId(), + did, + handle, + displayName, + avatar, + }) + .onConflictDoUpdate({ + target: schema.users.did, + set: { + handle, + displayName, + avatar, + updatedAt: new Date(), + }, + }) + .returning() + .execute(); + + return user; +}; + +const createArtists = async (artists: Artists, _user: SelectUser) => { + if (artists.length === 0) return; + + await ctx.db + .insert(schema.artists) + .values( + artists.map((artist) => ({ + id: createId(), + name: artist.value.name, + cid: artist.cid, + uri: artist.uri, + biography: artist.value.bio, + born: artist.value.born ? new Date(artist.value.born) : null, + bornIn: artist.value.bornIn, + died: artist.value.died ? new Date(artist.value.died) : null, + picture: artist.value.pictureUrl, + sha256: artist.value.sha256 as string, + genres: (artist.value.genres as string[]).join(", "), + })), + ) + .onConflictDoNothing({ + target: schema.artists.cid, + }) + .returning() + .execute(); +}; + +const createAlbums = async (albums: Albums, user: SelectUser) => { + if (albums.length === 0) return; + + await ctx.db + .insert(schema.albums) + .values( + albums.map((album) => ({ + id: createId(), + cid: album.cid, + title: "", + artist: "", + sha256: "", + uri: album.uri, + mbid: "", + description: "", + imageUrl: "", + spotifyId: "", + appleMusicId: "", + genres: "", + releaseDate: "", + year: undefined, + })), + ) + .onConflictDoNothing({ + target: schema.albums.cid, + }) + .returning() + .execute(); +}; + +const createSongs = async (songs: Songs, user: SelectUser) => { + if (songs.length === 0) return; + + await ctx.db + .insert(schema.tracks) + .values( + songs.map((song) => ({ + id: createId(), + cid: song.cid, + uri: song.uri, + title: song.value.title, + artist: song.value.artist, + albumArtist: song.value.albumArtist, + albumArt: song.value.albumArtUrl, + album: song.value.album, + trackNumber: song.value.trackNumber, + duration: song.value.duration, + mbId: song.value.mbid, + youtubeLink: song.value.youtubeLink, + spotifyLink: song.value.spotifyLink, + appleMusicLink: song.value.appleMusicLink, + tidalLink: song.value.tidalLink, + discNumber: song.value.discNumber, + lyrics: song.value.lyrics, + composer: song.value.composer, + genre: song.value.genre, + label: song.value.label, + copyrightMessage: song.value.copyrightMessage, + albumUri: "", + artistUri: "", + })), + ) + .onConflictDoNothing({ + target: schema.tracks.cid, + }) + .returning() + .execute(); +}; + +const createScrobbles = async (scrobbles: Scrobbles, user: SelectUser) => { + if (!scrobbles.length) return; + + await ctx.db + .insert(schema.scrobbles) + .values( + scrobbles.map((scrobble) => ({ + id: createId(), + trackId: "", + userId: user.id, + timestamp: new Date(), + })), + ) + .onConflictDoNothing({ + target: schema.scrobbles.cid, + }) + .returning() + .execute(); +}; + +const subscribeToJetstream = (_did: string) => { const client = new JetStreamClient({ wantedCollections: [ "app.rocksky.scrobble", @@ -25,7 +245,7 @@ export function sync() { endpoint: getEndpoint(), // Optional: filter by specific DIDs - // wantedDids: ["did:plc:example123"], + // wantedDids: [did], // Reconnection settings maxReconnectAttempts: 10, @@ -72,4 +292,129 @@ export function sync() { }); client.connect(); -} +}; + +const getRockskyUserSongs = async (agent: Agent): Promise => { + let results: { + value: Song.Record; + uri: string; + cid: string; + }[] = []; + let cursor: string | undefined; + let i = 1; + do { + const res = await agent.com.atproto.repo.listRecords({ + repo: agent.assertDid, + collection: "app.rocksky.song", + limit: 100, + cursor, + }); + const records = res.data.records as Array<{ + uri: string; + cid: string; + value: Song.Record; + }>; + results = results.concat(records); + cursor = res.data.cursor; + logger.info(`${chalk.greenBright(i)} songs`); + i += 100; + } while (cursor); + + return results; +}; + +const getRockskyUserAlbums = async (agent: Agent): Promise => { + let results: { + value: Album.Record; + uri: string; + cid: string; + }[] = []; + let cursor: string | undefined; + let i = 1; + do { + const res = await agent.com.atproto.repo.listRecords({ + repo: agent.assertDid, + collection: "app.rocksky.album", + limit: 100, + cursor, + }); + + const records = res.data.records as Array<{ + uri: string; + cid: string; + value: Album.Record; + }>; + + results = results.concat(records); + + cursor = res.data.cursor; + logger.info(`${chalk.greenBright(i)} albums`); + i += 100; + } while (cursor); + + return results; +}; + +const getRockskyUserArtists = async (agent: Agent): Promise => { + let results: { + value: Artist.Record; + uri: string; + cid: string; + }[] = []; + let cursor: string | undefined; + let i = 1; + do { + const res = await agent.com.atproto.repo.listRecords({ + repo: agent.assertDid, + collection: "app.rocksky.artist", + limit: 100, + cursor, + }); + + const records = res.data.records as Array<{ + uri: string; + cid: string; + value: Artist.Record; + }>; + + results = results.concat(records); + + cursor = res.data.cursor; + logger.info(`${chalk.greenBright(i)} artists`); + i += 100; + } while (cursor); + + return results; +}; + +const getRockskyUserScrobbles = async (agent: Agent): Promise => { + let results: { + value: Scrobble.Record; + uri: string; + cid: string; + }[] = []; + let cursor: string | undefined; + let i = 1; + do { + const res = await agent.com.atproto.repo.listRecords({ + repo: agent.assertDid, + collection: "app.rocksky.scrobble", + limit: 100, + cursor, + }); + + const records = res.data.records as Array<{ + uri: string; + cid: string; + value: Scrobble.Record; + }>; + + results = results.concat(records); + + cursor = res.data.cursor; + logger.info(`${chalk.greenBright(i)} scrobbles`); + i += 100; + } while (cursor); + + return results; +}; diff --git a/apps/cli/src/context.ts b/apps/cli/src/context.ts index c0b1693d..39e81d3e 100644 --- a/apps/cli/src/context.ts +++ b/apps/cli/src/context.ts @@ -1,8 +1,19 @@ import { logger } from "logger"; import drizzle from "./drizzle"; +import sqliteKv from "sqliteKv"; +import { createBidirectionalResolver, createIdResolver } from "lib/idResolver"; +import { createStorage } from "unstorage"; + +const kv = createStorage({ + driver: sqliteKv({ location: ":memory:", table: "kv" }), +}); + +const baseIdResolver = createIdResolver(kv); export const ctx = { db: drizzle.db, + resolver: createBidirectionalResolver(baseIdResolver), + baseIdResolver, logger, }; diff --git a/apps/cli/src/lib/agent.ts b/apps/cli/src/lib/agent.ts new file mode 100644 index 00000000..70a81a09 --- /dev/null +++ b/apps/cli/src/lib/agent.ts @@ -0,0 +1,55 @@ +import { Agent, AtpAgent } from "@atproto/api"; +import { ctx } from "context"; +import { eq } from "drizzle-orm"; +import authSessions from "schema/auth-session"; +import extractPdsFromDid from "./extractPdsFromDid"; +import { env } from "./env"; + +export async function createAgent(did: string, handle: string): Promise { + const pds = await extractPdsFromDid(did); + const agent = new AtpAgent({ + service: new URL(pds), + }); + + try { + const [data] = await ctx.db + .select() + .from(authSessions) + .where(eq(authSessions.key, did)) + .execute(); + + if (!data) { + throw new Error("No session found"); + } + + await agent.resumeSession(JSON.parse(data.session)); + return agent; + } catch (e) { + ctx.logger.error`resuming session ${did}`; + ctx.logger.error(e); + + await ctx.db + .delete(authSessions) + .where(eq(authSessions.key, did)) + .execute(); + + await agent.login({ + identifier: handle, + password: env.ROCKSKY_PASSWORD, + }); + + await ctx.db + .insert(authSessions) + .values({ + key: did, + session: JSON.stringify(agent.session), + }) + .onConflictDoUpdate({ + target: authSessions.key, + set: { session: JSON.stringify(agent.session) }, + }) + .execute(); + + return agent; + } +} diff --git a/apps/cli/src/lib/didUnstorageCache.ts b/apps/cli/src/lib/didUnstorageCache.ts new file mode 100644 index 00000000..b708713d --- /dev/null +++ b/apps/cli/src/lib/didUnstorageCache.ts @@ -0,0 +1,72 @@ +import type { CacheResult, DidCache, DidDocument } from "@atproto/identity"; +import type { Storage } from "unstorage"; + +const HOUR = 60e3 * 60; +const DAY = HOUR * 24; + +type CacheVal = { + doc: DidDocument; + updatedAt: number; +}; + +/** + * An unstorage based DidCache with staleness and max TTL + */ +export class StorageCache implements DidCache { + public staleTTL: number; + public maxTTL: number; + public cache: Storage; + private prefix: string; + constructor({ + store, + prefix, + staleTTL, + maxTTL, + }: { + store: Storage; + prefix: string; + staleTTL?: number; + maxTTL?: number; + }) { + this.cache = store as Storage; + this.prefix = prefix; + this.staleTTL = staleTTL ?? HOUR; + this.maxTTL = maxTTL ?? DAY; + } + + async cacheDid(did: string, doc: DidDocument): Promise { + await this.cache.set(this.prefix + did, { doc, updatedAt: Date.now() }); + } + + async refreshCache( + did: string, + getDoc: () => Promise, + ): Promise { + const doc = await getDoc(); + if (doc) { + await this.cacheDid(did, doc); + } + } + + async checkCache(did: string): Promise { + const val = await this.cache.get(this.prefix + did); + if (!val) return null; + const now = Date.now(); + const expired = now > val.updatedAt + this.maxTTL; + const stale = now > val.updatedAt + this.staleTTL; + return { + ...val, + did, + stale, + expired, + }; + } + + async clearEntry(did: string): Promise { + await this.cache.remove(this.prefix + did); + } + + async clear(): Promise { + await this.cache.clear(this.prefix); + } +} diff --git a/apps/cli/src/lib/env.ts b/apps/cli/src/lib/env.ts new file mode 100644 index 00000000..1ddf8cbb --- /dev/null +++ b/apps/cli/src/lib/env.ts @@ -0,0 +1,13 @@ +import dotenv from "dotenv"; +import { cleanEnv, str } from "envalid"; + +dotenv.config(); + +export const env = cleanEnv(process.env, { + ROCKSKY_IDENTIFIER: str({}), + ROCKSKY_HANDLE: str({ default: "" }), + ROCKSKY_PASSWORD: str({}), + JETSTREAM_SERVER: str({ + default: "wss://jetstream1.us-west.bsky.network/subscribe", + }), +}); diff --git a/apps/cli/src/lib/extractPdsFromDid.ts b/apps/cli/src/lib/extractPdsFromDid.ts new file mode 100644 index 00000000..3227f88a --- /dev/null +++ b/apps/cli/src/lib/extractPdsFromDid.ts @@ -0,0 +1,33 @@ +export default async function extractPdsFromDid( + did: string, +): Promise { + let didDocUrl: string; + + if (did.startsWith("did:plc:")) { + didDocUrl = `https://plc.directory/${did}`; + } else if (did.startsWith("did:web:")) { + const domain = did.substring("did:web:".length); + didDocUrl = `https://${domain}/.well-known/did.json`; + } else { + throw new Error("Unsupported DID method"); + } + + const response = await fetch(didDocUrl); + if (!response.ok) throw new Error("Failed to fetch DID doc"); + + const doc: { + service?: Array<{ + type: string; + id: string; + serviceEndpoint: string; + }>; + } = await response.json(); + + // Find the atproto PDS service + const pdsService = doc.service?.find( + (s: any) => + s.type === "AtprotoPersonalDataServer" && s.id.endsWith("#atproto_pds"), + ); + + return pdsService?.serviceEndpoint ?? null; +} diff --git a/apps/cli/src/lib/idResolver.ts b/apps/cli/src/lib/idResolver.ts new file mode 100644 index 00000000..51ed79d1 --- /dev/null +++ b/apps/cli/src/lib/idResolver.ts @@ -0,0 +1,52 @@ +import { IdResolver } from "@atproto/identity"; +import type { Storage } from "unstorage"; +import { StorageCache } from "./didUnstorageCache"; + +const HOUR = 60e3 * 60; +const DAY = HOUR * 24; +const WEEK = HOUR * 7; + +export function createIdResolver(kv: Storage) { + return new IdResolver({ + didCache: new StorageCache({ + store: kv, + prefix: "didCache:", + staleTTL: DAY, + maxTTL: WEEK, + }), + }); +} + +export interface BidirectionalResolver { + resolveDidToHandle(did: string): Promise; + resolveDidsToHandles(dids: string[]): Promise>; +} + +export function createBidirectionalResolver(resolver: IdResolver) { + return { + async resolveDidToHandle(did: string): Promise { + const didDoc = await resolver.did.resolveAtprotoData(did); + + // asynchronously double check that the handle resolves back + resolver.handle.resolve(didDoc.handle).then((resolvedHandle) => { + if (resolvedHandle !== did) { + resolver.did.ensureResolve(did, true); + } + }); + return didDoc?.handle ?? did; + }, + + async resolveDidsToHandles( + dids: string[], + ): Promise> { + const didHandleMap: Record = {}; + const resolves = await Promise.all( + dids.map((did) => this.resolveDidToHandle(did).catch((_) => did)), + ); + for (let i = 0; i < dids.length; i++) { + didHandleMap[dids[i]] = resolves[i]; + } + return didHandleMap; + }, + }; +} diff --git a/apps/cli/src/schema/album-tracks.ts b/apps/cli/src/schema/album-tracks.ts index 74d3fbca..8e0ee192 100644 --- a/apps/cli/src/schema/album-tracks.ts +++ b/apps/cli/src/schema/album-tracks.ts @@ -1,7 +1,7 @@ import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; -import albums from "./albums.js"; -import tracks from "./tracks.js"; +import albums from "./albums"; +import tracks from "./tracks"; const albumTracks = sqliteTable("album_tracks", { id: text("id").primaryKey().notNull(), diff --git a/apps/cli/src/schema/albums.ts b/apps/cli/src/schema/albums.ts index 2e6a7d5d..d8dd6f61 100644 --- a/apps/cli/src/schema/albums.ts +++ b/apps/cli/src/schema/albums.ts @@ -9,12 +9,12 @@ const albums = sqliteTable("albums", { year: integer("year"), albumArt: text("album_art"), uri: text("uri").unique(), + cid: text("cid").unique().notNull(), artistUri: text("artist_uri"), appleMusicLink: text("apple_music_link").unique(), spotifyLink: text("spotify_link").unique(), tidalLink: text("tidal_link").unique(), youtubeLink: text("youtube_link").unique(), - sha256: text("sha256").unique().notNull(), createdAt: integer("created_at", { mode: "timestamp" }) .notNull() .default(sql`CURRENT_TIMESTAMP`), diff --git a/apps/cli/src/schema/artist-albums.ts b/apps/cli/src/schema/artist-albums.ts index 2edf86d8..31b97008 100644 --- a/apps/cli/src/schema/artist-albums.ts +++ b/apps/cli/src/schema/artist-albums.ts @@ -1,7 +1,7 @@ import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; -import albums from "./albums.js"; -import artists from "./artists.js"; +import albums from "./albums"; +import artists from "./artists"; const artistAlbums = sqliteTable("artist_albums", { id: text("id").primaryKey().notNull(), diff --git a/apps/cli/src/schema/artist-tracks.ts b/apps/cli/src/schema/artist-tracks.ts index c38d33e2..1614d5f7 100644 --- a/apps/cli/src/schema/artist-tracks.ts +++ b/apps/cli/src/schema/artist-tracks.ts @@ -1,7 +1,7 @@ import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; -import artists from "./artists.js"; -import tracks from "./tracks.js"; +import artists from "./artists"; +import tracks from "./tracks"; const artistTracks = sqliteTable("artist_tracks", { id: text("id").primaryKey().notNull(), diff --git a/apps/cli/src/schema/artists.ts b/apps/cli/src/schema/artists.ts index 0c1b7123..e26c499b 100644 --- a/apps/cli/src/schema/artists.ts +++ b/apps/cli/src/schema/artists.ts @@ -9,8 +9,8 @@ const artists = sqliteTable("artists", { bornIn: text("born_in"), died: integer("died", { mode: "timestamp" }), picture: text("picture"), - sha256: text("sha256").unique().notNull(), uri: text("uri").unique(), + cid: text("cid").unique().notNull(), appleMusicLink: text("apple_music_link"), spotifyLink: text("spotify_link"), tidalLink: text("tidal_link"), diff --git a/apps/cli/src/schema/auth-session.ts b/apps/cli/src/schema/auth-session.ts new file mode 100644 index 00000000..b9e84000 --- /dev/null +++ b/apps/cli/src/schema/auth-session.ts @@ -0,0 +1,18 @@ +import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; +import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; + +const authSessions = sqliteTable("auth_sessions", { + key: text("key").primaryKey().notNull(), + session: text("session").notNull(), + createdAt: integer("created_at", { mode: "timestamp" }) + .notNull() + .default(sql`CURRENT_TIMESTAMP`), + updatedAt: integer("updated_at", { mode: "timestamp" }) + .notNull() + .default(sql`CURRENT_TIMESTAMP`), +}); + +export type SelectAuthSession = InferSelectModel; +export type InsertAuthSession = InferInsertModel; + +export default authSessions; diff --git a/apps/cli/src/schema/index.ts b/apps/cli/src/schema/index.ts index 8a6c80be..91505b86 100644 --- a/apps/cli/src/schema/index.ts +++ b/apps/cli/src/schema/index.ts @@ -1,7 +1,15 @@ import albumTracks from "./album-tracks"; import albums from "./albums"; +import artistAlbums from "./artist-albums"; +import artistTracks from "./artist-tracks"; import artists from "./artists"; +import authSessions from "./auth-session"; +import lovedTracks from "./loved-tracks"; +import scrobbles from "./scrobbles"; import tracks from "./tracks"; +import userAlbums from "./user-albums"; +import userArtists from "./user-artists"; +import userTracks from "./user-tracks"; import users from "./users"; export default { @@ -10,4 +18,12 @@ export default { artists, albums, albumTracks, + authSessions, + artistAlbums, + artistTracks, + lovedTracks, + scrobbles, + userAlbums, + userArtists, + userTracks, }; diff --git a/apps/cli/src/schema/scrobbles.ts b/apps/cli/src/schema/scrobbles.ts new file mode 100644 index 00000000..6732c6ca --- /dev/null +++ b/apps/cli/src/schema/scrobbles.ts @@ -0,0 +1,30 @@ +import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; +import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; +import albums from "./albums"; +import artists from "./artists"; +import tracks from "./tracks"; +import users from "./users"; + +const scrobbles = sqliteTable("scrobbles", { + id: text("xata_id").primaryKey().notNull(), + userId: text("user_id").references(() => users.id), + trackId: text("track_id").references(() => tracks.id), + albumId: text("album_id").references(() => albums.id), + artistId: text("artist_id").references(() => artists.id), + uri: text("uri").unique(), + cid: text("cid").unique(), + createdAt: integer("created_at", { mode: "timestamp" }) + .notNull() + .default(sql`CURRENT_TIMESTAMP`), + updatedAt: integer("updated_at", { mode: "timestamp" }) + .notNull() + .default(sql`CURRENT_TIMESTAMP`), + timestamp: integer("timestamp", { mode: "timestamp" }) + .notNull() + .default(sql`CURRENT_TIMESTAMP`), +}); + +export type SelectScrobble = InferSelectModel; +export type InsertScrobble = InferInsertModel; + +export default scrobbles; diff --git a/apps/cli/src/schema/tracks.ts b/apps/cli/src/schema/tracks.ts index 3e779343..70553fbd 100644 --- a/apps/cli/src/schema/tracks.ts +++ b/apps/cli/src/schema/tracks.ts @@ -15,7 +15,6 @@ const tracks = sqliteTable("tracks", { spotifyLink: text("spotify_link").unique(), appleMusicLink: text("apple_music_link").unique(), tidalLink: text("tidal_link").unique(), - sha256: text("sha256").unique().notNull(), discNumber: integer("disc_number"), lyrics: text("lyrics"), composer: text("composer"), @@ -23,6 +22,7 @@ const tracks = sqliteTable("tracks", { label: text("label"), copyrightMessage: text("copyright_message"), uri: text("uri").unique(), + cid: text("cid").unique().notNull(), albumUri: text("album_uri"), artistUri: text("artist_uri"), createdAt: integer("created_at", { mode: "timestamp" }) diff --git a/apps/cli/src/sqliteKv.ts b/apps/cli/src/sqliteKv.ts new file mode 100644 index 00000000..fab414df --- /dev/null +++ b/apps/cli/src/sqliteKv.ts @@ -0,0 +1,173 @@ +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect } from "kysely"; +import { defineDriver } from "unstorage"; + +interface TableSchema { + [k: string]: { + id: string; + value: string; + created_at: string; + updated_at: string; + }; +} + +export type KvDb = Kysely; + +const DRIVER_NAME = "sqlite"; + +export default defineDriver< + { + location?: string; + table: string; + getDb?: () => KvDb; + }, + KvDb +>( + ({ + location, + table = "kv", + getDb = (): KvDb => { + let _db: KvDb | null = null; + + return (() => { + if (_db) { + return _db; + } + + if (!location) { + throw new Error("SQLite location is required"); + } + + const sqlite = new Database(location, { fileMustExist: false }); + + // Enable WAL mode + sqlite.pragma("journal_mode = WAL"); + + _db = new Kysely({ + dialect: new SqliteDialect({ + database: sqlite, + }), + }); + + // Create table if not exists + _db.schema + .createTable(table) + .ifNotExists() + .addColumn("id", "text", (col) => col.primaryKey()) + .addColumn("value", "text", (col) => col.notNull()) + .addColumn("created_at", "text", (col) => col.notNull()) + .addColumn("updated_at", "text", (col) => col.notNull()) + .execute(); + + return _db; + })(); + }, + }) => { + return { + name: DRIVER_NAME, + options: { location, table }, + getInstance: getDb, + + async hasItem(key) { + const result = await getDb() + .selectFrom(table) + .select(["id"]) + .where("id", "=", key) + .executeTakeFirst(); + return !!result; + }, + + async getItem(key) { + const result = await getDb() + .selectFrom(table) + .select(["value"]) + .where("id", "=", key) + .executeTakeFirst(); + return result?.value ?? null; + }, + + async setItem(key: string, value: string) { + const now = new Date().toISOString(); + await getDb() + .insertInto(table) + .values({ + id: key, + value, + created_at: now, + updated_at: now, + }) + .onConflict((oc) => + oc.column("id").doUpdateSet({ + value, + updated_at: now, + }), + ) + .execute(); + }, + + async setItems(items) { + const now = new Date().toISOString(); + + await getDb() + .transaction() + .execute(async (trx) => { + await Promise.all( + items.map(({ key, value }) => { + return trx + .insertInto(table) + .values({ + id: key, + value, + created_at: now, + updated_at: now, + }) + .onConflict((oc) => + oc.column("id").doUpdateSet({ + value, + updated_at: now, + }), + ) + .execute(); + }), + ); + }); + }, + + async removeItem(key: string) { + await getDb().deleteFrom(table).where("id", "=", key).execute(); + }, + + async getMeta(key: string) { + const result = await getDb() + .selectFrom(table) + .select(["created_at", "updated_at"]) + .where("id", "=", key) + .executeTakeFirst(); + if (!result) { + return null; + } + return { + birthtime: new Date(result.created_at), + mtime: new Date(result.updated_at), + }; + }, + + async getKeys(base = "") { + const results = await getDb() + .selectFrom(table) + .select(["id"]) + .where("id", "like", `${base}%`) + .execute(); + return results.map((r) => r.id); + }, + + async clear() { + await getDb().deleteFrom(table).execute(); + }, + + async dispose() { + await getDb().destroy(); + }, + }; + }, +); diff --git a/bun.lock b/bun.lock index d27d7a0a..386c59a6 100644 --- a/bun.lock +++ b/bun.lock @@ -125,14 +125,19 @@ "commander": "^13.1.0", "cors": "^2.8.5", "dayjs": "^1.11.13", + "dotenv": "^16.4.7", "drizzle-kit": "^0.31.1", "drizzle-orm": "^0.45.1", "effect": "^3.19.14", "env-paths": "^3.0.0", + "envalid": "^8.0.0", "express": "^5.1.0", + "kysely": "^0.27.5", + "lodash": "^4.17.21", "md5": "^2.3.0", "open": "^10.1.0", "table": "^6.9.0", + "unstorage": "^1.14.4", "zod": "^3.24.3", }, "devDependencies": { diff --git a/crates/analytics/src/handlers/artists.rs b/crates/analytics/src/handlers/artists.rs index be06d0de..4ab0c2d5 100644 --- a/crates/analytics/src/handlers/artists.rs +++ b/crates/analytics/src/handlers/artists.rs @@ -345,7 +345,7 @@ pub async fn get_artist_tracks( LEFT JOIN tracks t ON at.track_id = t.id LEFT JOIN artists a ON at.artist_id = a.id LEFT JOIN scrobbles s ON s.track_id = t.id - WHERE at.artist_id = ? OR a.uri = ? + WHERE (a.id = ? OR a.uri = ?) AND (t.artist_uri = ?) GROUP BY t.id, t.title, t.artist, t.album_artist, t.album, t.uri, t.album_art, t.duration, t.disc_number, t.track_number, t.artist_uri, t.album_uri, t.sha256, t.copyright_message, t.label, t.created_at ORDER BY play_count DESC @@ -355,6 +355,7 @@ pub async fn get_artist_tracks( let tracks = stmt.query_map( [ + ¶ms.artist_id, ¶ms.artist_id, ¶ms.artist_id, &limit.to_string(),