musicshare/apps/api/src/services/nowPlaying.ts

import { and, desc, eq, inArray, lt } from 'drizzle-orm';
import type { NowPlayingDTO, NowPlayingProvider, RecentPlayDTO, UserSummaryDTO } from '@musicshare/shared';
import { db, schema } from '../db';
import type { User } from '../db/schema';
import { publicAvatarUrl } from '../lib/avatars';
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: publicAvatarUrl(u.avatarUrl),
});

/** Phones report changes and a heartbeat every few minutes; if they go quiet (app killed, no signal), stop showing "listening now". */
export const DEVICE_STALE_MS = 9 * 60_000;
const PUSH_SOURCES = new Set<NowPlayingProvider>(['android_media_session']);

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 && !(PUSH_SOURCES.has(np.source) && Date.now() - np.updatedAt.getTime() > DEVICE_STALE_MS),
      source: np.source,
      sourceApp: np.sourceApp,
      sourceUrl: np.sourceUrl,
      progressMs: np.progressMs,
      updatedAt: np.updatedAt.toISOString(),
    }))
    .sort((a, b) => Number(b.isPlaying) - Number(a.isPlaying) || b.updatedAt.localeCompare(a.updatedAt));
}

/** Plays kept per user. Profiles show the latest few; the rest leaves room for a future history page. */
export const LISTEN_HISTORY_LIMIT = 50;
/** Coming back to the same track after this long counts as a new play. */
const REPLAY_GAP_MS = 10 * 60_000;

async function recordListen(userId: string, trackId: string, source: NowPlayingProvider, at: Date) {
  await db.insert(schema.listens).values({ userId, trackId, source, startedAt: at });
  // Trim to the newest LISTEN_HISTORY_LIMIT rows for this user.
  const [cutoff] = await db
    .select({ startedAt: schema.listens.startedAt })
    .from(schema.listens)
    .where(eq(schema.listens.userId, userId))
    .orderBy(desc(schema.listens.startedAt))
    .offset(LISTEN_HISTORY_LIMIT - 1)
    .limit(1);
  if (cutoff) {
    await db.delete(schema.listens).where(and(eq(schema.listens.userId, userId), lt(schema.listens.startedAt, cutoff.startedAt)));
  }
}

export async function getRecentPlays(userId: string, limit = 10): Promise<RecentPlayDTO[]> {
  const rows = await db
    .select()
    .from(schema.listens)
    .where(eq(schema.listens.userId, userId))
    .orderBy(desc(schema.listens.startedAt))
    .limit(limit + 1);
  // Don't repeat the track that's playing right now.
  const [current] = await db.select().from(schema.nowPlaying).where(eq(schema.nowPlaying.userId, userId));
  const plays = current?.isPlaying && rows[0]?.trackId === current.trackId ? rows.slice(1) : rows.slice(0, limit);
  const tracks = await getTrackDTOs(plays.map((p) => p.trackId));
  return plays.map((p) => ({ track: tracks.get(p.trackId)!, source: p.source, playedAt: p.startedAt.toISOString() }));
}

/** Paused by the user, or suspended by a moderator: nothing new is shared either way. */
export const isSharingPaused = (user: Pick<User, 'sharingPausedUntil'> & Partial<Pick<User, 'suspendedAt'>>) =>
  Boolean(user.suspendedAt) || Boolean(user.sharingPausedUntil && user.sharingPausedUntil > new Date());

/** Starting a pause: whatever was playing becomes "played … ago" for friends. */
export async function pauseSharing(userId: string) {
  const updated = await db
    .update(schema.nowPlaying)
    .set({ isPlaying: false, updatedAt: new Date() })
    .where(and(eq(schema.nowPlaying.userId, userId), eq(schema.nowPlaying.isPlaying, true)))
    .returning({ userId: schema.nowPlaying.userId });
  if (updated.length) await publishNowPlaying(userId);
}

export async function publishNowPlaying(userId: string) {
  const [dto] = await getNowPlaying([userId]);
  if (dto) await redis.publish(nowPlayingChannel(userId), JSON.stringify(dto));
}

const SEEK_THRESHOLD_MS = 5_000;

/** 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,
  opts: { sourceApp?: string | null; sourceUrl?: string | 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;
  // Same song, but not where it should be by now: a seek (or a restart). Tell
  // watchers, or their progress bars would drift.
  const jumped =
    !changed &&
    current.progressMs != null &&
    result.progressMs != null &&
    Math.abs(result.progressMs - (current.progressMs + (now.getTime() - current.updatedAt.getTime()))) > SEEK_THRESHOLD_MS;
  const newPlay =
    !current || current.trackId !== track.id || (!current.isPlaying && now.getTime() - current.updatedAt.getTime() > REPLAY_GAP_MS);
  const values = {
    trackId: track.id,
    source,
    sourceApp: opts.sourceApp ?? null,
    sourceUrl: opts.sourceUrl ?? null,
    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 (newPlay) await recordListen(userId, track.id, source, now);
  if (jumped) await publishNowPlaying(userId);
  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;
}