Add phone playback reporting and pause sharing to the API
Phones that read system media sessions report playback themselves:
- POST /me/devices issues a per-phone token that can only call
POST /devices/now-playing; revocable, hashed at rest, 10 per user
- Reports name the app ("via Spotify"); unknown apps use the OS label
- Push-reported playback expires after 9 minutes of silence, since
phones send a heartbeat every 4 minutes while playing
Users can pause sharing (PUT /me/sharing: 1h, 8h, indefinitely). While
paused, connections aren't polled, phone reports are dropped, nothing is
added to history, and the current track switches to "played … ago".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MzFywujuH9TiuXbr9vjoX8
This commit is contained in:
@@ -11,6 +11,7 @@ import { HttpError } from './lib/http';
|
||||
import { redis } from './lib/redis';
|
||||
import { authRoutes } from './routes/auth';
|
||||
import { connectionRoutes } from './routes/connections';
|
||||
import { deviceRoutes } from './routes/devices';
|
||||
import { meRoutes } from './routes/me';
|
||||
import { realtimeRoutes } from './routes/realtime';
|
||||
import { trackRoutes } from './routes/tracks';
|
||||
@@ -56,6 +57,7 @@ export async function buildApp({ logger = true, rateLimits = true }: { logger?:
|
||||
await app.register(userRoutes);
|
||||
await app.register(trackRoutes);
|
||||
await app.register(connectionRoutes);
|
||||
await app.register(deviceRoutes);
|
||||
await app.register(realtimeRoutes);
|
||||
// registered as its own plugin so its permissive body parsers stay scoped to webhook routes
|
||||
await app.register(webhookRoutes);
|
||||
|
||||
@@ -26,6 +26,8 @@ export const users = pgTable('users', {
|
||||
avatarUrl: varchar('avatar_url', { length: 500 }),
|
||||
preferredService: text('preferred_service').$type<StreamingService>(),
|
||||
/** Billing plan. Kept in sync by the billing webhook (RevenueCat/Stripe) — see lib/entitlements.ts */
|
||||
/** While in the future, nothing new is shared: services aren't polled and phone reports are dropped. */
|
||||
sharingPausedUntil: timestamp('sharing_paused_until', { withTimezone: true }),
|
||||
plan: text('plan').$type<Plan>().notNull().default('free'),
|
||||
planExpiresAt: timestamp('plan_expires_at', { withTimezone: true }),
|
||||
createdAt: createdAt(),
|
||||
@@ -134,6 +136,8 @@ export const nowPlaying = pgTable(
|
||||
.notNull()
|
||||
.references(() => tracks.id),
|
||||
source: text('source').$type<NowPlayingProvider>().notNull(),
|
||||
/** App name when a phone reported it, e.g. "Spotify". */
|
||||
sourceApp: text('source_app'),
|
||||
isPlaying: boolean('is_playing').notNull(),
|
||||
progressMs: integer('progress_ms'),
|
||||
startedAt: timestamp('started_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
@@ -142,6 +146,24 @@ export const nowPlaying = pgTable(
|
||||
(t) => [index('now_playing_track_idx').on(t.trackId)],
|
||||
);
|
||||
|
||||
/** Phones that report playback directly (Android media sessions). Each has its own revocable token. */
|
||||
export const devices = pgTable(
|
||||
'devices',
|
||||
{
|
||||
id: uuid('id').primaryKey().defaultRandom(),
|
||||
userId: uuid('user_id')
|
||||
.notNull()
|
||||
.references(() => users.id, { onDelete: 'cascade' }),
|
||||
/** sha256 of the device token; the token can only report playback. */
|
||||
tokenHash: text('token_hash').notNull().unique(),
|
||||
platform: text('platform').$type<'android' | 'ios'>().notNull(),
|
||||
name: varchar('name', { length: 100 }).notNull(),
|
||||
lastSeenAt: timestamp('last_seen_at', { withTimezone: true }),
|
||||
createdAt: createdAt(),
|
||||
},
|
||||
(t) => [index('devices_user_idx').on(t.userId)],
|
||||
);
|
||||
|
||||
/** Recent plays per user, for "Recently played" on profiles. Trimmed to the latest LISTEN_HISTORY_LIMIT. */
|
||||
export const listens = pgTable(
|
||||
'listens',
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
/**
|
||||
* Phones that report playback themselves (the Android app reads system media
|
||||
* sessions). Each phone registers once and gets a device token that can only
|
||||
* call POST /devices/now-playing, so a leaked token can't read or change
|
||||
* anything else, and the user can revoke a single phone.
|
||||
*/
|
||||
import { and, desc, eq, notInArray } from 'drizzle-orm';
|
||||
import type { FastifyInstance, FastifyRequest } from 'fastify';
|
||||
import { ANDROID_MUSIC_APPS, devicePlaybackSchema, registerDeviceSchema, type DeviceDTO } from '@musicshare/shared';
|
||||
import { db, schema } from '../db';
|
||||
import { requireUser } from '../lib/auth';
|
||||
import { randomToken, sha256 } from '../lib/crypto';
|
||||
import { HttpError } from '../lib/http';
|
||||
import { applyPollResult, isSharingPaused } from '../services/nowPlaying';
|
||||
|
||||
const MAX_DEVICES_PER_USER = 10;
|
||||
export const DEVICE_TOKEN_PREFIX = 'dev_';
|
||||
|
||||
type Device = typeof schema.devices.$inferSelect;
|
||||
|
||||
const toDeviceDTO = (d: Device): DeviceDTO => ({
|
||||
id: d.id,
|
||||
platform: d.platform,
|
||||
name: d.name,
|
||||
lastSeenAt: d.lastSeenAt?.toISOString() ?? null,
|
||||
createdAt: d.createdAt.toISOString(),
|
||||
});
|
||||
|
||||
function deviceToken(req: FastifyRequest): string | null {
|
||||
const header = req.headers.authorization;
|
||||
const token = header?.startsWith('Bearer ') ? header.slice(7) : null;
|
||||
return token?.startsWith(DEVICE_TOKEN_PREFIX) ? token : null;
|
||||
}
|
||||
|
||||
export async function deviceRoutes(app: FastifyInstance) {
|
||||
app.get('/me/devices', async (req) => {
|
||||
const user = requireUser(req);
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(schema.devices)
|
||||
.where(eq(schema.devices.userId, user.id))
|
||||
.orderBy(desc(schema.devices.createdAt));
|
||||
return rows.map(toDeviceDTO);
|
||||
});
|
||||
|
||||
app.post('/me/devices', async (req) => {
|
||||
const user = requireUser(req);
|
||||
const input = registerDeviceSchema.parse(req.body);
|
||||
const token = DEVICE_TOKEN_PREFIX + randomToken(32);
|
||||
const [device] = await db
|
||||
.insert(schema.devices)
|
||||
.values({ userId: user.id, tokenHash: sha256(token), platform: input.platform, name: input.name })
|
||||
.returning();
|
||||
|
||||
// Reinstalls register a new device; keep only the newest few.
|
||||
const keep = await db
|
||||
.select({ id: schema.devices.id })
|
||||
.from(schema.devices)
|
||||
.where(eq(schema.devices.userId, user.id))
|
||||
.orderBy(desc(schema.devices.createdAt))
|
||||
.limit(MAX_DEVICES_PER_USER);
|
||||
await db.delete(schema.devices).where(
|
||||
and(
|
||||
eq(schema.devices.userId, user.id),
|
||||
notInArray(
|
||||
schema.devices.id,
|
||||
keep.map((k) => k.id),
|
||||
),
|
||||
),
|
||||
);
|
||||
return { device: toDeviceDTO(device!), token };
|
||||
});
|
||||
|
||||
app.delete<{ Params: { id: string } }>('/me/devices/:id', async (req) => {
|
||||
const user = requireUser(req);
|
||||
if (!/^[0-9a-f-]{36}$/i.test(req.params.id)) throw new HttpError(404, 'Device not found');
|
||||
await db.delete(schema.devices).where(and(eq(schema.devices.id, req.params.id), eq(schema.devices.userId, user.id)));
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
app.post(
|
||||
'/devices/now-playing',
|
||||
{
|
||||
config: {
|
||||
rateLimit: {
|
||||
max: 120,
|
||||
timeWindow: '1 minute',
|
||||
keyGenerator: (req: FastifyRequest) => `device:${sha256(deviceToken(req) ?? req.ip)}`,
|
||||
},
|
||||
},
|
||||
},
|
||||
async (req, reply) => {
|
||||
const token = deviceToken(req);
|
||||
const [device] = token
|
||||
? await db.select().from(schema.devices).where(eq(schema.devices.tokenHash, sha256(token)))
|
||||
: [];
|
||||
if (!device) throw new HttpError(401, 'Unknown device');
|
||||
|
||||
const input = devicePlaybackSchema.parse(req.body);
|
||||
await db.update(schema.devices).set({ lastSeenAt: new Date() }).where(eq(schema.devices.id, device.id));
|
||||
const [owner] = await db.select().from(schema.users).where(eq(schema.users.id, device.userId));
|
||||
if (!owner || isSharingPaused(owner)) return reply.status(204).send(); // accepted, not shared
|
||||
const sourceApp = ANDROID_MUSIC_APPS[input.app] ?? input.appName ?? null;
|
||||
const track = input.track
|
||||
? {
|
||||
title: input.track.title,
|
||||
artist: input.track.artist || 'Unknown artist',
|
||||
album: input.track.album ?? null,
|
||||
durationMs: input.track.durationMs ?? null,
|
||||
}
|
||||
: null;
|
||||
|
||||
await applyPollResult(
|
||||
device.userId,
|
||||
'android_media_session',
|
||||
track ? { track, isPlaying: input.state === 'playing', progressMs: input.positionMs ?? null } : null,
|
||||
{ sourceApp },
|
||||
);
|
||||
return reply.status(204).send();
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import { eq } from 'drizzle-orm';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import { setFeaturedSchema, updateMeSchema, type MeDTO } from '@musicshare/shared';
|
||||
import { setFeaturedSchema, setSharingSchema, updateMeSchema, type MeDTO } from '@musicshare/shared';
|
||||
import { db, schema } from '../db';
|
||||
import type { User } from '../db/schema';
|
||||
import { SESSION_COOKIE, requireUser } from '../lib/auth';
|
||||
@@ -8,7 +8,9 @@ import { effectivePlan, entitlementsFor } from '../lib/entitlements';
|
||||
import { HttpError } from '../lib/http';
|
||||
import { deezerTrack } from '../providers/catalog';
|
||||
import { enqueueResolve } from '../services/queues';
|
||||
import { unschedulePoll } from '../services/scheduler';
|
||||
import { connectionIdsForUser } from '../services/connections';
|
||||
import { isSharingPaused, pauseSharing } from '../services/nowPlaying';
|
||||
import { pollSoon, unschedulePoll } from '../services/scheduler';
|
||||
import { upsertTrack } from '../services/tracks';
|
||||
|
||||
export const toMeDTO = (u: User): MeDTO => ({
|
||||
@@ -19,6 +21,7 @@ export const toMeDTO = (u: User): MeDTO => ({
|
||||
email: u.email,
|
||||
bio: u.bio,
|
||||
preferredService: u.preferredService,
|
||||
sharingPausedUntil: isSharingPaused(u) ? u.sharingPausedUntil!.toISOString() : null,
|
||||
plan: effectivePlan(u),
|
||||
entitlements: entitlementsFor(u),
|
||||
});
|
||||
@@ -46,6 +49,21 @@ export async function meRoutes(app: FastifyInstance) {
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
app.put('/me/sharing', async (req) => {
|
||||
const user = requireUser(req);
|
||||
const { pause } = setSharingSchema.parse(req.body);
|
||||
const hours = { '1h': 1, '8h': 8, indefinitely: 24 * 365 * 100 } as const;
|
||||
const until = pause ? new Date(Date.now() + hours[pause] * 3600_000) : null;
|
||||
const [updated] = await db
|
||||
.update(schema.users)
|
||||
.set({ sharingPausedUntil: until })
|
||||
.where(eq(schema.users.id, user.id))
|
||||
.returning();
|
||||
if (until) await pauseSharing(user.id);
|
||||
else await pollSoon(await connectionIdsForUser(user.id)); // pick up what's playing right away
|
||||
return toMeDTO(updated!);
|
||||
});
|
||||
|
||||
app.put('/me/featured', async (req) => {
|
||||
const user = requireUser(req);
|
||||
const input = setFeaturedSchema.parse(req.body);
|
||||
|
||||
@@ -50,3 +50,11 @@ export async function watchUsers(userIds: string[]) {
|
||||
);
|
||||
await pollSoon(rows.map((r) => r.id));
|
||||
}
|
||||
|
||||
export async function connectionIdsForUser(userId: string): Promise<string[]> {
|
||||
const rows = await db
|
||||
.select({ id: schema.connections.id })
|
||||
.from(schema.connections)
|
||||
.where(and(eq(schema.connections.userId, userId), eq(schema.connections.status, 'active')));
|
||||
return rows.map((r) => r.id);
|
||||
}
|
||||
|
||||
@@ -14,6 +14,10 @@ export const toUserSummary = (u: Pick<User, 'id' | 'username' | 'displayName' |
|
||||
avatarUrl: 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
|
||||
@@ -26,8 +30,9 @@ export async function getNowPlaying(userIds: string[]): Promise<NowPlayingDTO[]>
|
||||
.map(({ np, user }) => ({
|
||||
user: toUserSummary(user),
|
||||
track: tracks.get(np.trackId)!,
|
||||
isPlaying: np.isPlaying,
|
||||
isPlaying: np.isPlaying && !(PUSH_SOURCES.has(np.source) && Date.now() - np.updatedAt.getTime() > DEVICE_STALE_MS),
|
||||
source: np.source,
|
||||
sourceApp: np.sourceApp,
|
||||
progressMs: np.progressMs,
|
||||
updatedAt: np.updatedAt.toISOString(),
|
||||
}))
|
||||
@@ -68,6 +73,19 @@ export async function getRecentPlays(userId: string, limit = 10): Promise<Recent
|
||||
return plays.map((p) => ({ track: tracks.get(p.trackId)!, source: p.source, playedAt: p.startedAt.toISOString() }));
|
||||
}
|
||||
|
||||
export const isSharingPaused = (user: Pick<User, 'sharingPausedUntil'>) =>
|
||||
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));
|
||||
@@ -78,6 +96,7 @@ export async function applyPollResult(
|
||||
userId: string,
|
||||
source: NowPlayingProvider,
|
||||
result: PollResult | null,
|
||||
opts: { sourceApp?: string | null } = {},
|
||||
): Promise<boolean> {
|
||||
const [current] = await db.select().from(schema.nowPlaying).where(eq(schema.nowPlaying.userId, userId));
|
||||
const now = new Date();
|
||||
@@ -102,6 +121,7 @@ export async function applyPollResult(
|
||||
const values = {
|
||||
trackId: track.id,
|
||||
source,
|
||||
sourceApp: opts.sourceApp ?? null,
|
||||
isPlaying: true,
|
||||
progressMs: result.progressMs,
|
||||
updatedAt: now,
|
||||
|
||||
@@ -6,7 +6,7 @@ import { closeConnections } from './lib/lifecycle';
|
||||
import { createRedis } from './lib/redis';
|
||||
import { POLLED_PROVIDERS, pollingProviders } from './providers';
|
||||
import { ProviderAuthError, ProviderRateLimitError } from './providers/types';
|
||||
import { applyPollResult } from './services/nowPlaying';
|
||||
import { applyPollResult, isSharingPaused } from './services/nowPlaying';
|
||||
import { POLL_QUEUE, RESOLVE_QUEUE, pollQueue } from './services/queues';
|
||||
import { resolveTrackLinks } from './services/resolver';
|
||||
import {
|
||||
@@ -34,6 +34,13 @@ async function pollConnection(connectionId: string) {
|
||||
}
|
||||
const { conn, user } = row;
|
||||
|
||||
if (isSharingPaused(user)) {
|
||||
// Don't even ask the provider while paused; check back when the pause ends.
|
||||
const resumeIn = user.sharingPausedUntil!.getTime() - Date.now();
|
||||
await schedulePoll(conn.id, Math.min(resumeIn, IDLE_POLL_SECONDS * 1000));
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await provider.poll(conn);
|
||||
const playing = await applyPollResult(user.id, conn.provider, result);
|
||||
|
||||
Reference in New Issue
Block a user