Add Rocksky sync, ID resolver, sqlite KV driver

Introduce Rocksky sync flow and Agent support. Replace sha256 columns
with cid across schemas, add scrobbles and auth_sessions tables, and add
an unstorage SQLite driver and DID cache. Add env parsing, ID
resolution/extraction utilities, update context, and adjust deps and
migration metadata

apps/cli/drizzle/0000_amazing_redwing.sql apps/cli/drizzle/0000_parched_robbie_robertson.sql

···
 	`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,
···
 	`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,
···
 	`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,
···
 );
 --> 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,
···
 	`spotify_link` text,
 	`apple_music_link` text,
 	`tidal_link` text,
-	`sha256` text NOT NULL,
 	`disc_number` integer,
 	`lyrics` text,
 	`composer` text,
···
 	`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,
···
 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,

apps/cli/drizzle/meta/0000_snapshot.json

···
 {
   "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": {
···
           "notNull": false,
           "autoincrement": false
         },
+        "cid": {
+          "name": "cid",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
         "artist_uri": {
           "name": "artist_uri",
           "type": "text",
···
           "notNull": false,
           "autoincrement": false
         },
-        "sha256": {
-          "name": "sha256",
-          "type": "text",
-          "primaryKey": false,
-          "notNull": true,
-          "autoincrement": false
-        },
         "created_at": {
           "name": "created_at",
           "type": "integer",
···
           ],
           "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": [
···
           "name": "albums_youtube_link_unique",
           "columns": [
             "youtube_link"
-          ],
-          "isUnique": true
-        },
-        "albums_sha256_unique": {
-          "name": "albums_sha256_unique",
-          "columns": [
-            "sha256"
           ],
           "isUnique": true
         }
···
           "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": {
···
         }
       },
       "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
         }
···
       "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": {
···
       "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": {
···
           "notNull": false,
           "autoincrement": false
         },
-        "sha256": {
-          "name": "sha256",
-          "type": "text",
-          "primaryKey": false,
-          "notNull": true,
-          "autoincrement": false
-        },
         "disc_number": {
           "name": "disc_number",
           "type": "integer",
···
           "notNull": false,
           "autoincrement": false
         },
+        "cid": {
+          "name": "cid",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
         "album_uri": {
           "name": "album_uri",
           "type": "text",
···
           ],
           "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
         }

apps/cli/drizzle/meta/_journal.json

···
     {
       "idx": 0,
       "version": "6",
-      "when": 1767904137315,
-      "tag": "0000_amazing_redwing",
+      "when": 1768034891092,
+      "tag": "0000_parched_robbie_robertson",
       "breakpoints": true
     }
   ]

apps/cli/package.json

···
     "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": {

apps/cli/src/cmd/sync.ts

···
 import { JetStreamClient, JetStreamEvent } from "jetstream";
+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 { logger } from "logger";
+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;
···
   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<SelectUser> => {
+  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",
···
     endpoint: getEndpoint(),
 
     // Optional: filter by specific DIDs
-    // wantedDids: ["did:plc:example123"],
+    // wantedDids: [did],
 
     // Reconnection settings
     maxReconnectAttempts: 10,
···
   });
 
   client.connect();
-}
+};
+
+const getRockskyUserSongs = async (agent: Agent): Promise<Songs> => {
+  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<Albums> => {
+  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<Artists> => {
+  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<Scrobbles> => {
+  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;
+};

apps/cli/src/context.ts

···
 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,
 };
 

apps/cli/src/lib/agent.ts

···
+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<Agent> {
+  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;
+  }
+}

apps/cli/src/lib/didUnstorageCache.ts

···
+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<CacheVal>;
+  private prefix: string;
+  constructor({
+    store,
+    prefix,
+    staleTTL,
+    maxTTL,
+  }: {
+    store: Storage;
+    prefix: string;
+    staleTTL?: number;
+    maxTTL?: number;
+  }) {
+    this.cache = store as Storage<CacheVal>;
+    this.prefix = prefix;
+    this.staleTTL = staleTTL ?? HOUR;
+    this.maxTTL = maxTTL ?? DAY;
+  }
+
+  async cacheDid(did: string, doc: DidDocument): Promise<void> {
+    await this.cache.set(this.prefix + did, { doc, updatedAt: Date.now() });
+  }
+
+  async refreshCache(
+    did: string,
+    getDoc: () => Promise<DidDocument | null>,
+  ): Promise<void> {
+    const doc = await getDoc();
+    if (doc) {
+      await this.cacheDid(did, doc);
+    }
+  }
+
+  async checkCache(did: string): Promise<CacheResult | null> {
+    const val = await this.cache.get<CacheVal>(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<void> {
+    await this.cache.remove(this.prefix + did);
+  }
+
+  async clear(): Promise<void> {
+    await this.cache.clear(this.prefix);
+  }
+}

apps/cli/src/lib/env.ts

···
+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",
+  }),
+});

apps/cli/src/lib/extractPdsFromDid.ts

···
+export default async function extractPdsFromDid(
+  did: string,
+): Promise<string | null> {
+  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;
+}

apps/cli/src/lib/idResolver.ts

···
+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<string>;
+  resolveDidsToHandles(dids: string[]): Promise<Record<string, string>>;
+}
+
+export function createBidirectionalResolver(resolver: IdResolver) {
+  return {
+    async resolveDidToHandle(did: string): Promise<string> {
+      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<Record<string, string>> {
+      const didHandleMap: Record<string, string> = {};
+      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;
+    },
+  };
+}

apps/cli/src/schema/album-tracks.ts

···
 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(),

apps/cli/src/schema/albums.ts

···
   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`),

apps/cli/src/schema/artist-albums.ts

···
 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(),

apps/cli/src/schema/artist-tracks.ts

···
 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(),

apps/cli/src/schema/artists.ts

···
   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"),

apps/cli/src/schema/auth-session.ts

···
+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<typeof authSessions>;
+export type InsertAuthSession = InferInsertModel<typeof authSessions>;
+
+export default authSessions;

apps/cli/src/schema/index.ts

···
 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 {
···
   artists,
   albums,
   albumTracks,
+  authSessions,
+  artistAlbums,
+  artistTracks,
+  lovedTracks,
+  scrobbles,
+  userAlbums,
+  userArtists,
+  userTracks,
 };

apps/cli/src/schema/scrobbles.ts

···
+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<typeof scrobbles>;
+export type InsertScrobble = InferInsertModel<typeof scrobbles>;
+
+export default scrobbles;

apps/cli/src/schema/tracks.ts

···
   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"),
···
   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" })

apps/cli/src/sqliteKv.ts

···
+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<TableSchema>;
+
+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<TableSchema>({
+          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();
+      },
+    };
+  },
+);

bun.lock

···
         "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": {

crates/analytics/src/handlers/artists.rs

···
         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
···
 
     let tracks = stmt.query_map(
         [
+            &params.artist_id,
             &params.artist_id,
             &params.artist_id,
             &limit.to_string(),