diff --git a/apps/api/src/providers/plex.ts b/apps/api/src/providers/plex.ts index 6956642..1838566 100644 --- a/apps/api/src/providers/plex.ts +++ b/apps/api/src/providers/plex.ts @@ -61,8 +61,11 @@ interface PlexServer { name: string; owned: boolean; tokenEnc: string; - /** candidate base URLs, best first */ + /** candidate base URLs, last known good first */ uris: string[]; + /** Backoff for shared (friends') servers that keep failing, so they don't slow every poll. */ + failures?: number; + retryAt?: number; } interface PlexMeta { @@ -116,26 +119,48 @@ interface PlexSessionTrack { Player?: { state: 'playing' | 'paused' | 'buffering' }; } -/** Try each candidate URI until one answers; moves the working one to the front. */ +const REQUEST_TIMEOUT_MS = 5000; +const MAX_BACKOFF_MS = 30 * 60_000; + +/** Human-readable reason, e.g. "ECONNREFUSED" rather than undici's generic "fetch failed". */ +function describe(err: unknown): string { + if (err instanceof UnsafeUrlError) return 'only reachable on a private network'; + if (err instanceof HttpError) return `HTTP ${err.status}`; + const e = err as { name?: string; message?: string; cause?: { code?: string; message?: string } }; + if (e?.name === 'TimeoutError') return 'timed out'; + return e?.cause?.code ?? e?.cause?.message ?? e?.message ?? String(err); +} + +async function sessionsAt(uri: string, token: string): Promise { + const base = await assertSafeServerUrl(uri); + const res = await fetchJson<{ MediaContainer: { Metadata?: PlexSessionTrack[] } }>( + `${base}/status/sessions`, + { headers: plexHeaders(token), ...serverFetchOptions() }, + REQUEST_TIMEOUT_MS, + ); + return res?.MediaContainer.Metadata ?? []; +} + +/** + * Servers advertise many addresses (public, relay, LAN, Docker bridges). Try the + * last one that worked; if it fails, race the rest and remember the winner. + */ async function fetchSessions(server: PlexServer): Promise { const token = decrypt(server.tokenEnc); - let lastError: unknown; - for (const [i, uri] of server.uris.entries()) { + const [first, ...rest] = server.uris; + if (!first) throw new Error('no addresses'); + try { + return await sessionsAt(first, token); + } catch (err) { + if (rest.length === 0 || (err instanceof HttpError && err.status === 401)) throw err; try { - const base = await assertSafeServerUrl(uri); - const res = await fetchJson<{ MediaContainer: { Metadata?: PlexSessionTrack[] } }>( - `${base}/status/sessions`, - { headers: plexHeaders(token), ...serverFetchOptions() }, - 5000, - ); - if (i > 0) server.uris = [uri, ...server.uris.filter((u) => u !== uri)]; - return res?.MediaContainer.Metadata ?? []; - } catch (e) { - lastError = e; - if (e instanceof HttpError && e.status === 401) break; // token no longer valid for this server + const winner = await Promise.any(rest.map((uri) => sessionsAt(uri, token).then((sessions) => ({ uri, sessions })))); + server.uris = [winner.uri, ...server.uris.filter((u) => u !== winner.uri)]; + return winner.sessions; + } catch { + throw err; // report why the preferred address failed } } - throw lastError instanceof Error ? lastError : new Error(`${server.name}: unreachable`); } function isUsersSession(s: PlexSessionTrack, accountId: string, server: PlexServer): boolean { @@ -154,27 +179,46 @@ export const plexProvider: PollingProvider = { meta.discoveredAt = Date.now(); } - const results = await Promise.allSettled(servers.map((s) => fetchSessions(s))); - await db - .update(schema.connections) - .set({ meta: { ...meta, servers } }) - .where(eq(schema.connections.id, conn.id)); - - if (servers.length > 0 && results.every((r) => r.status === 'rejected')) { - // force rediscovery next time; server addresses may have changed - await db - .update(schema.connections) - .set({ meta: { ...meta, servers, discoveredAt: 0 } }) - .where(eq(schema.connections.id, conn.id)); - const reason = (results[0] as PromiseRejectedResult).reason; - if (reason instanceof UnsafeUrlError) throw new Error('Your Plex server is only reachable on a private network'); - throw new Error(`Couldn't reach your Plex server${servers.length > 1 ? 's' : ''}: ${reason?.message ?? reason}`); - } + const now = Date.now(); + const due = servers.filter((s) => s.owned || !s.retryAt || s.retryAt <= now); + const attempted = due.length > 0 ? due : servers; + const results = await Promise.all( + attempted.map(async (server) => { + try { + const sessions = await fetchSessions(server); + delete server.failures; + delete server.retryAt; + return { server, sessions }; + } catch (err) { + if (!server.owned) { + server.failures = (server.failures ?? 0) + 1; + server.retryAt = now + Math.min(MAX_BACKOFF_MS, 60_000 * 2 ** (server.failures - 1)); + } + return { server, error: describe(err) }; + } + }), + ); const mine = results - .flatMap((r, i) => (r.status === 'fulfilled' ? r.value.map((s) => ({ s, server: servers[i]! })) : [])) + .flatMap(({ server, sessions }) => (sessions ?? []).map((s) => ({ s, server }))) .filter(({ s, server }) => s.type === 'track' && isUsersSession(s, conn.externalUserId, server)) .map(({ s }) => s); + + const failed = results.filter((r) => r.error); + const ownedAttempted = results.filter((r) => r.server.owned); + const allFailed = failed.length === results.length; + const ownServersDown = ownedAttempted.length > 0 && ownedAttempted.every((r) => r.error); + await db + .update(schema.connections) + // force rediscovery next time if nothing answered; server addresses may have changed + .set({ meta: { ...meta, servers, ...(allFailed ? { discoveredAt: 0 } : {}) } }) + .where(eq(schema.connections.id, conn.id)); + + if (mine.length === 0 && (allFailed || ownServersDown)) { + const details = (ownServersDown && !allFailed ? ownedAttempted : failed).map((r) => `${r.server.name}: ${r.error}`); + throw new Error(`Couldn't reach your Plex server${details.length > 1 ? 's' : ''} (${details.join('; ')})`); + } + const session = mine.find((s) => s.Player?.state === 'playing') ?? mine[0]; if (!session) return null; diff --git a/apps/api/test/connections.test.ts b/apps/api/test/connections.test.ts index 887c992..4b0039e 100644 --- a/apps/api/test/connections.test.ts +++ b/apps/api/test/connections.test.ts @@ -160,7 +160,7 @@ describe('OAuth return URLs', () => { }); describe('Plex polling', () => { - async function plexConnection(uris: string[]) { + async function plexConnection(uris: string[], extraServers: object[] = []) { const { user } = await createUser(app(), 'alice'); return upsertConnection({ userId: user.id, @@ -168,9 +168,14 @@ describe('Plex polling', () => { externalUserId: '12345', displayName: null, accessTokenEnc: encrypt('account-token'), - meta: { discoveredAt: Date.now(), servers: [{ name: 'Home', owned: true, tokenEnc: encrypt('server-token'), uris }] }, + meta: { + discoveredAt: Date.now(), + servers: [{ name: 'Home', owned: true, tokenEnc: encrypt('server-token'), uris }, ...extraServers], + }, }); } + const friendServer = (name: string, uri: string) => ({ name, owned: false, tokenEnc: encrypt('t'), uris: [uri] }); + const refused = () => Promise.reject(Object.assign(new TypeError('fetch failed'), { cause: { code: 'ECONNREFUSED' } })); const sessions = { MediaContainer: { @@ -208,4 +213,32 @@ describe('Plex polling', () => { await expect(plexProvider.poll(conn)).rejects.toThrow(/private network/); expect(fetchMock).not.toHaveBeenCalled(); }); + + it("backs off a friend's server that keeps failing, so it stops slowing every poll", async () => { + let friendCalls = 0; + vi.stubGlobal( + 'fetch', + routeFetch([ + [/^https:\/\/8\.8\.8\.8/, () => json(sessions)], + [/^https:\/\/9\.9\.9\.9/, () => (friendCalls++, refused())], + ]), + ); + const conn = await plexConnection(['https://8.8.8.8:32400'], [friendServer('Friend', 'https://9.9.9.9:32400')]); + expect((await plexProvider.poll(conn))?.track.title).toBe('Digital Love'); + const [row] = await db.select().from(schema.connections).where(eq(schema.connections.id, conn.id)); + await plexProvider.poll(row!); + expect(friendCalls).toBe(1); // skipped on the second poll + }); + + it("reports the owner's server being down even when friends' servers answer", async () => { + vi.stubGlobal( + 'fetch', + routeFetch([ + [/^https:\/\/8\.8\.8\.8/, () => refused()], + [/^https:\/\/9\.9\.9\.9/, () => json({ MediaContainer: {} })], + ]), + ); + const conn = await plexConnection(['https://8.8.8.8:32400'], [friendServer('Friend', 'https://9.9.9.9:32400')]); + await expect(plexProvider.poll(conn)).rejects.toThrow("Couldn't reach your Plex server (Home: ECONNREFUSED)"); + }); });