Add API and worker: accounts, follows, now-playing, cross-service links

- Fastify API with cookie (web) and bearer (mobile) sessions, follows,
  pinned friends, featured songs, account deletion, Redis-backed rate limits
- Providers: Spotify (OAuth), Last.fm, ListenBrainz, Plex (PIN flow and
  server discovery), Jellyfin (token auth, revoked on disconnect)
- Worker polls on an adaptive schedule in a Redis sorted set, so polling
  scales horizontally and API usage tracks whether anyone is watching
- Track resolver links songs across services via ISRC (Deezer, Spotify),
  iTunes search and song.link
- Realtime updates over WebSocket fan out through Redis pub/sub
- User-supplied server URLs are restricted to public addresses, checked
  both up front and at connect time to defeat DNS rebinding

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MzFywujuH9TiuXbr9vjoX8

apps/api/src/services/connections.ts

@@ -0,0 +1,52 @@
+import { and, eq, inArray } from 'drizzle-orm';
+import type { ConnectionDTO } from '@musicshare/shared';
+import { db, schema } from '../db';
+import type { Connection } from '../db/schema';
+import { env } from '../env';
+import { randomToken } from '../lib/crypto';
+import { POLLED_PROVIDERS } from '../providers';
+import { markWatched, pollSoon, schedulePoll } from './scheduler';
+
+export const toConnectionDTO = (c: Connection): ConnectionDTO => ({
+  provider: c.provider,
+  externalUserId: c.externalUserId,
+  displayName: c.displayName,
+  status: c.status,
+  lastError: c.lastError,
+  lastPolledAt: c.lastPolledAt?.toISOString() ?? null,
+  webhookUrl: c.webhookSecret ? `${env.API_PUBLIC_URL}/webhooks/${c.provider}/${c.webhookSecret}` : null,
+});
+
+/** Providers that can notify us via webhook (which triggers an immediate poll). */
+const WEBHOOK_PROVIDERS = new Set(['plex', 'jellyfin']);
+
+export async function upsertConnection(
+  values: Omit<typeof schema.connections.$inferInsert, 'id' | 'createdAt' | 'status' | 'lastError' | 'webhookSecret'>,
+) {
+  const set = { ...values, status: 'active' as const, lastError: null };
+  const [row] = await db
+    .insert(schema.connections)
+    // on reconnect the existing webhook secret is kept, so URLs users pasted into Plex/Jellyfin keep working
+    .values({ ...set, webhookSecret: WEBHOOK_PROVIDERS.has(values.provider) ? randomToken(24) : null })
+    .onConflictDoUpdate({ target: [schema.connections.userId, schema.connections.provider], set })
+    .returning();
+  await schedulePoll(row!.id);
+  return row!;
+}
+
+/** Someone started looking at these users: poll them now instead of waiting for the idle interval. */
+export async function watchUsers(userIds: string[]) {
+  const fresh = await markWatched(userIds);
+  if (fresh.length === 0) return;
+  const rows = await db
+    .select({ id: schema.connections.id })
+    .from(schema.connections)
+    .where(
+      and(
+        inArray(schema.connections.userId, fresh),
+        eq(schema.connections.status, 'active'),
+        inArray(schema.connections.provider, POLLED_PROVIDERS),
+      ),
+    );
+  await pollSoon(rows.map((r) => r.id));
+}

apps/api/src/services/nowPlaying.ts

@@ -0,0 +1,88 @@
+import { eq, inArray } from 'drizzle-orm';
+import type { NowPlayingDTO, NowPlayingProvider, UserSummaryDTO } from '@musicshare/shared';
+import { db, schema } from '../db';
+import type { User } from '../db/schema';
+import { nowPlayingChannel, redis } from '../lib/redis';
+import type { PollResult } from '../providers/types';
+import { enqueueResolve } from './queues';
+import { getTrackDTOs, upsertTrack } from './tracks';
+
+export const toUserSummary = (u: Pick<User, 'id' | 'username' | 'displayName' | 'avatarUrl'>): UserSummaryDTO => ({
+  id: u.id,
+  username: u.username,
+  displayName: u.displayName,
+  avatarUrl: u.avatarUrl,
+});
+
+export async function getNowPlaying(userIds: string[]): Promise<NowPlayingDTO[]> {
+  if (userIds.length === 0) return [];
+  const rows = await db
+    .select({ np: schema.nowPlaying, user: schema.users })
+    .from(schema.nowPlaying)
+    .innerJoin(schema.users, eq(schema.users.id, schema.nowPlaying.userId))
+    .where(inArray(schema.nowPlaying.userId, userIds));
+  const tracks = await getTrackDTOs(rows.map((r) => r.np.trackId));
+  return rows
+    .map(({ np, user }) => ({
+      user: toUserSummary(user),
+      track: tracks.get(np.trackId)!,
+      isPlaying: np.isPlaying,
+      source: np.source,
+      progressMs: np.progressMs,
+      updatedAt: np.updatedAt.toISOString(),
+    }))
+    .sort((a, b) => Number(b.isPlaying) - Number(a.isPlaying) || b.updatedAt.localeCompare(a.updatedAt));
+}
+
+export async function publishNowPlaying(userId: string) {
+  const [dto] = await getNowPlaying([userId]);
+  if (dto) await redis.publish(nowPlayingChannel(userId), JSON.stringify(dto));
+}
+
+/** Apply one provider's poll result to the user's now-playing state. Returns whether they're playing. */
+export async function applyPollResult(
+  userId: string,
+  source: NowPlayingProvider,
+  result: PollResult | null,
+): Promise<boolean> {
+  const [current] = await db.select().from(schema.nowPlaying).where(eq(schema.nowPlaying.userId, userId));
+  const now = new Date();
+
+  if (!result || !result.isPlaying) {
+    // Only the source that reported playback may mark it stopped; otherwise an
+    // idle Last.fm connection would clobber an active Spotify session.
+    if (current?.isPlaying && current.source === source) {
+      await db
+        .update(schema.nowPlaying)
+        .set({ isPlaying: false, updatedAt: now })
+        .where(eq(schema.nowPlaying.userId, userId));
+      await publishNowPlaying(userId);
+    }
+    return false;
+  }
+
+  const track = await upsertTrack(result.track);
+  const changed = !current || current.trackId !== track.id || !current.isPlaying;
+  const values = {
+    trackId: track.id,
+    source,
+    isPlaying: true,
+    progressMs: result.progressMs,
+    updatedAt: now,
+    ...(changed ? { startedAt: now } : {}),
+  };
+  await db
+    .insert(schema.nowPlaying)
+    .values({ userId, ...values })
+    .onConflictDoUpdate({ target: schema.nowPlaying.userId, set: values });
+
+  if (changed) {
+    await publishNowPlaying(userId);
+    const [t] = await db
+      .select({ resolved: schema.tracks.linksResolvedAt })
+      .from(schema.tracks)
+      .where(eq(schema.tracks.id, track.id));
+    if (!t?.resolved) await enqueueResolve(track.id);
+  }
+  return true;
+}

apps/api/src/services/queues.ts

@@ -0,0 +1,24 @@
+import { Queue } from 'bullmq';
+import { createRedis } from '../lib/redis';
+
+const connection = createRedis();
+
+export const POLL_QUEUE = 'poll';
+export const RESOLVE_QUEUE = 'resolve-track';
+
+export const pollQueue = new Queue<{ connectionId: string }>(POLL_QUEUE, { connection });
+export const resolveQueue = new Queue<{ trackId: string }>(RESOLVE_QUEUE, { connection });
+
+export function enqueueResolve(trackId: string) {
+  return resolveQueue.add(
+    'resolve',
+    { trackId },
+    {
+      jobId: `resolve-${trackId}`,
+      attempts: 3,
+      backoff: { type: 'exponential', delay: 30_000 },
+      removeOnComplete: true,
+      removeOnFail: 1000,
+    },
+  );
+}

apps/api/src/services/resolver.ts

@@ -0,0 +1,76 @@
+import { eq } from 'drizzle-orm';
+import type { StreamingService } from '@musicshare/shared';
+import { db, schema } from '../db';
+import { spotifyConfigured } from '../env';
+import { appleMusicFind, deezerByIsrc, deezerFind, odesliLinks, type DeezerTrack } from '../providers/catalog';
+import { searchSpotify } from '../providers/spotify';
+import { publishNowPlaying } from './nowPlaying';
+import { addLinks } from './tracks';
+
+/**
+ * Fill in links to every streaming service for a track.
+ * ISRC is the backbone: Deezer gives us one for free, and Spotify can be
+ * searched by it exactly. Apple Music (without MusicKit) and YouTube Music
+ * need fuzzy matching or song.link.
+ */
+export async function resolveTrackLinks(trackId: string) {
+  const [track] = await db.select().from(schema.tracks).where(eq(schema.tracks.id, trackId));
+  if (!track) return;
+  const existing = await db.select().from(schema.trackLinks).where(eq(schema.trackLinks.trackId, trackId));
+  const links = new Map<StreamingService, string>(existing.map((l) => [l.service, l.url]));
+  let isrc = track.isrc;
+
+  // Deezer: canonical ISRC + artwork source
+  if (!links.has('deezer') || !isrc) {
+    let dz: DeezerTrack | null = isrc ? await deezerByIsrc(isrc) : null;
+    const matchType = dz ? 'isrc' : 'search';
+    dz ??= await deezerFind(track);
+    if (dz) {
+      await addLinks(trackId, { deezer: { id: String(dz.id), url: dz.link } }, matchType);
+      links.set('deezer', dz.link);
+      const patch: Partial<typeof track> = {};
+      if (!isrc && dz.isrc) isrc = patch.isrc = dz.isrc;
+      if (!track.artworkUrl) patch.artworkUrl = dz.album.cover_xl ?? dz.album.cover_big ?? null;
+      if (!track.durationMs) patch.durationMs = dz.duration * 1000;
+      if (!track.album) patch.album = dz.album.title;
+      if (Object.keys(patch).length) await db.update(schema.tracks).set(patch).where(eq(schema.tracks.id, trackId));
+    }
+  }
+
+  if (!links.has('spotify') && spotifyConfigured) {
+    const sp = await searchSpotify(track, isrc);
+    const link = sp?.track.links?.spotify;
+    if (sp && link) {
+      await addLinks(trackId, { spotify: link }, sp.matchType);
+      links.set('spotify', link.url);
+    }
+  }
+
+  if (!links.has('apple_music')) {
+    const am = await appleMusicFind(track);
+    if (am) {
+      await addLinks(trackId, { apple_music: am }, 'search');
+      links.set('apple_music', am.url);
+    }
+  }
+
+  // song.link covers YouTube Music and TIDAL, which have no keyless search
+  if (!links.has('youtube_music') || !links.has('tidal')) {
+    const source = links.get('spotify') ?? links.get('deezer') ?? links.get('apple_music');
+    // best-effort: a song.link outage shouldn't block the links we already found
+    const found = source ? await odesliLinks(source).catch((e) => (console.warn('odesli failed', e.message), null)) : null;
+    if (found) {
+      const missing = Object.entries(found).filter(([s]) => !links.has(s as StreamingService));
+      await addLinks(trackId, Object.fromEntries(missing.map(([s, url]) => [s, { id: null, url }])), 'odesli');
+    }
+  }
+
+  await db.update(schema.tracks).set({ linksResolvedAt: new Date() }).where(eq(schema.tracks.id, trackId));
+
+  // push the new links to anyone currently looking at this track
+  const listeners = await db
+    .select({ userId: schema.nowPlaying.userId })
+    .from(schema.nowPlaying)
+    .where(eq(schema.nowPlaying.trackId, trackId));
+  await Promise.all(listeners.map((l) => publishNowPlaying(l.userId)));
+}

apps/api/src/services/scheduler.ts

@@ -0,0 +1,70 @@
+/**
+ * Poll scheduling. Each connection has a "next poll at" score in a Redis
+ * sorted set. Any number of workers can run `claimDue` concurrently: the Lua
+ * script claims atomically, so no leader election is needed and polling
+ * scales horizontally. Intervals adapt to whether anyone is watching the user,
+ * which is what keeps third-party API usage proportional to real engagement.
+ */
+import { redis } from '../lib/redis';
+
+const DUE_KEY = 'poll:due';
+const WATCHED_TTL_SECONDS = 90;
+
+export const IDLE_POLL_SECONDS = 180;
+export const PLAYING_UNWATCHED_POLL_SECONDS = 60;
+
+export function schedulePoll(connectionId: string, delayMs = 0) {
+  return redis.zadd(DUE_KEY, Date.now() + delayMs, connectionId);
+}
+
+/** Pull forward (never push back) a connection's next poll. */
+export function pollSoon(connectionIds: string[]) {
+  if (connectionIds.length === 0) return;
+  const now = Date.now();
+  return redis.zadd(DUE_KEY, 'LT', ...connectionIds.flatMap((id) => [now, id]));
+}
+
+/** Add without overriding an existing schedule. Used to reconcile from the DB. */
+export function ensureScheduled(connectionIds: string[]) {
+  if (connectionIds.length === 0) return;
+  const now = Date.now();
+  return redis.zadd(DUE_KEY, 'NX', ...connectionIds.flatMap((id) => [now, id]));
+}
+
+export const unschedulePoll = (connectionId: string) => redis.zrem(DUE_KEY, connectionId);
+
+const CLAIM_SCRIPT = `
+local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, ARGV[2])
+if #ids > 0 then
+  -- park claimed items far in the future; the worker reschedules after polling
+  for _, id in ipairs(ids) do redis.call('ZADD', KEYS[1], ARGV[3], id) end
+end
+return ids`;
+
+/** Claims due connections. Claimed items get a safety reschedule in case the worker dies. */
+export async function claimDue(limit = 500): Promise<string[]> {
+  const now = Date.now();
+  return (await redis.eval(CLAIM_SCRIPT, 1, DUE_KEY, now, limit, now + 5 * 60_000)) as string[];
+}
+
+// ---- presence: "someone is looking at this user right now" ----
+
+const watchedKey = (userId: string) => `watched:${userId}`;
+
+/** Marks users as watched; returns the ones that weren't already watched. */
+export async function markWatched(userIds: string[]): Promise<string[]> {
+  if (userIds.length === 0) return [];
+  const pipeline = redis.pipeline();
+  for (const id of userIds) pipeline.set(watchedKey(id), '1', 'EX', WATCHED_TTL_SECONDS, 'NX');
+  const res = await pipeline.exec();
+  const fresh = userIds.filter((_, i) => res?.[i]?.[1] === 'OK');
+  // refresh TTL on the rest
+  const refresh = redis.pipeline();
+  for (const id of userIds) refresh.expire(watchedKey(id), WATCHED_TTL_SECONDS);
+  await refresh.exec();
+  return fresh;
+}
+
+export async function isWatched(userId: string): Promise<boolean> {
+  return (await redis.exists(watchedKey(userId))) === 1;
+}

apps/api/src/services/tracks.ts

@@ -0,0 +1,104 @@
+import { and, eq, inArray, or } from 'drizzle-orm';
+import type { StreamingService, TrackDTO } from '@musicshare/shared';
+import { db, schema } from '../db';
+import type { Track } from '../db/schema';
+import { matchKey } from '../lib/http';
+import type { IncomingTrack } from '../providers/types';
+
+type LinkRow = typeof schema.trackLinks.$inferSelect;
+
+export function toTrackDTO(track: Track, links: LinkRow[]): TrackDTO {
+  return {
+    id: track.id,
+    title: track.title,
+    artist: track.artist,
+    album: track.album,
+    artworkUrl: track.artworkUrl,
+    durationMs: track.durationMs,
+    isrc: track.isrc,
+    links: Object.fromEntries(links.map((l) => [l.service, l.url])) as Partial<Record<StreamingService, string>>,
+  };
+}
+
+export async function getTrackDTOs(ids: string[]): Promise<Map<string, TrackDTO>> {
+  const unique = [...new Set(ids)];
+  if (unique.length === 0) return new Map();
+  const [rows, links] = await Promise.all([
+    db.select().from(schema.tracks).where(inArray(schema.tracks.id, unique)),
+    db.select().from(schema.trackLinks).where(inArray(schema.trackLinks.trackId, unique)),
+  ]);
+  return new Map(rows.map((t) => [t.id, toTrackDTO(t, links.filter((l) => l.trackId === t.id))]));
+}
+
+export async function addLinks(
+  trackId: string,
+  links: Partial<Record<StreamingService, { id: string | null; url: string }>>,
+  matchType: LinkRow['matchType'],
+) {
+  const rows = Object.entries(links).map(([service, l]) => ({
+    trackId,
+    service: service as StreamingService,
+    externalId: l.id,
+    url: l.url,
+    matchType,
+  }));
+  if (rows.length) await db.insert(schema.trackLinks).values(rows).onConflictDoNothing();
+}
+
+/**
+ * Find-or-create the canonical track for something a provider reported.
+ * Identity, strongest first: ISRC → a known provider ID → normalized artist|title.
+ */
+export async function upsertTrack(incoming: IncomingTrack): Promise<{ id: string; created: boolean }> {
+  const key = matchKey(incoming.artist, incoming.title);
+  let existing: Track | undefined;
+
+  if (incoming.isrc) {
+    [existing] = await db.select().from(schema.tracks).where(eq(schema.tracks.isrc, incoming.isrc)).limit(1);
+  }
+  const knownIds = Object.entries(incoming.links ?? {}).filter(([, l]) => l.id);
+  if (!existing && knownIds.length) {
+    const [hit] = await db
+      .select({ track: schema.tracks })
+      .from(schema.trackLinks)
+      .innerJoin(schema.tracks, eq(schema.tracks.id, schema.trackLinks.trackId))
+      .where(
+        or(
+          ...knownIds.map(([service, l]) =>
+            and(eq(schema.trackLinks.service, service as StreamingService), eq(schema.trackLinks.externalId, l.id!)),
+          ),
+        ),
+      )
+      .limit(1);
+    existing = hit?.track;
+  }
+  if (!existing) {
+    [existing] = await db.select().from(schema.tracks).where(eq(schema.tracks.matchKey, key)).limit(1);
+  }
+
+  if (existing) {
+    const patch: Partial<Track> = {};
+    if (!existing.isrc && incoming.isrc) patch.isrc = incoming.isrc;
+    if (!existing.artworkUrl && incoming.artworkUrl) patch.artworkUrl = incoming.artworkUrl;
+    if (!existing.album && incoming.album) patch.album = incoming.album;
+    if (!existing.durationMs && incoming.durationMs) patch.durationMs = incoming.durationMs;
+    if (Object.keys(patch).length) await db.update(schema.tracks).set(patch).where(eq(schema.tracks.id, existing.id));
+    if (incoming.links) await addLinks(existing.id, incoming.links, 'provider');
+    return { id: existing.id, created: false };
+  }
+
+  const [created] = await db
+    .insert(schema.tracks)
+    .values({
+      matchKey: key,
+      title: incoming.title,
+      artist: incoming.artist,
+      album: incoming.album ?? null,
+      durationMs: incoming.durationMs ?? null,
+      artworkUrl: incoming.artworkUrl ?? null,
+      isrc: incoming.isrc ?? null,
+    })
+    .returning({ id: schema.tracks.id });
+  if (incoming.links) await addLinks(created!.id, incoming.links, 'provider');
+  return { id: created!.id, created: true };
+}