From 3eec2f218433ca9aa72621e052ba9baebce907eb Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 15:49:31 +0100 Subject: [PATCH] feat(plugins): listenbrainz and lastfm Co-Authored-By: Claude Sonnet 5.5 --- plugins/lastfm/index.ts | 19 ++ plugins/lastfm/src/lastfm.ts | 137 +++++++++++++++ plugins/lastfm/src/options.ts | 8 + plugins/lastfm/test/lastfm.test.ts | 92 ++++++++++ plugins/listenbrainz/index.ts | 19 ++ plugins/listenbrainz/src/listenbrainz.ts | 164 ++++++++++++++++++ plugins/listenbrainz/src/options.ts | 8 + .../listenbrainz/test/listenbrainz.test.ts | 98 +++++++++++ 8 files changed, 545 insertions(+) create mode 100644 plugins/lastfm/index.ts create mode 100644 plugins/lastfm/src/lastfm.ts create mode 100644 plugins/lastfm/src/options.ts create mode 100644 plugins/lastfm/test/lastfm.test.ts create mode 100644 plugins/listenbrainz/index.ts create mode 100644 plugins/listenbrainz/src/listenbrainz.ts create mode 100644 plugins/listenbrainz/src/options.ts create mode 100644 plugins/listenbrainz/test/listenbrainz.test.ts diff --git a/plugins/lastfm/index.ts b/plugins/lastfm/index.ts new file mode 100644 index 0000000..e22470e --- /dev/null +++ b/plugins/lastfm/index.ts @@ -0,0 +1,19 @@ +import { definePlugin } from "@sttracker/plugin-sdk/types"; +import { pull, pullChanges } from "./src/lastfm.ts"; + +export default definePlugin({ + manifest: { + id: "lastfm", + name: "Last.fm", + version: "0.1.0", + sdkVersion: 1, + mediaTypes: ["track", "album", "artist"], + capabilities: ["pull", "pullChanges"], + config: [ + { key: "username", label: "Username", type: "string", required: true }, + { key: "apiKey", label: "API key", type: "secret", required: true }, + ], + }, + pull, + pullChanges, +}); diff --git a/plugins/lastfm/src/lastfm.ts b/plugins/lastfm/src/lastfm.ts new file mode 100644 index 0000000..3641f6c --- /dev/null +++ b/plugins/lastfm/src/lastfm.ts @@ -0,0 +1,137 @@ +import { HttpError } from "@sttracker/plugin-sdk/http"; +import type { ListenFields, PluginContext, SourceRecord } from "@sttracker/plugin-sdk/types"; +import { sleep } from "./options.ts"; + +export const BASE = "https://ws.audioscrobbler.com/2.0/"; +export const MAX_ATTEMPTS = 3; + +export class LastfmError extends Error { + readonly code: number; + constructor(code: number, message: string) { + super(`Last.fm error ${code}: ${message}`); + this.name = "LastfmError"; + this.code = code; + } +} + +type Q = Record; +interface Text { "#text"?: string; mbid?: string; name?: string } +interface LfmTrack { + name?: string; + mbid?: string; + artist?: Text; + album?: Text; + date?: { uts?: string }; + "@attr"?: { nowplaying?: string }; +} + +function errorOf(body: unknown): { code: number; message: string } | null { + if (body && typeof body === "object" && "error" in body && typeof (body as { error: unknown }).error === "number") { + const b = body as { error: number; message?: string }; + return { code: b.error, message: b.message ?? "unknown error" }; + } + return null; +} + +/** Last.fm reports errors in the JSON body (HTTP 200 or 4xx); error 29 (rate limit) is retried here. */ +export async function call(ctx: PluginContext, query: Q): Promise { + const apiKey = (await ctx.credentials.get())?.apiKey ?? String(ctx.config.apiKey ?? ""); + for (let attempt = 1; ; attempt++) { + ctx.signal.throwIfAborted(); + let body: unknown; + try { + body = await ctx.http.json(BASE, { query: { ...query, api_key: apiKey, format: "json" } }); + } catch (e) { + if (!(e instanceof HttpError)) throw e; + let parsed: unknown; + try { parsed = JSON.parse(e.body); } catch { throw e; } + const err = errorOf(parsed); + if (!err) throw e; + body = parsed; + } + const err = errorOf(body); + if (!err) return body as T; + if (err.code === 29 && attempt < MAX_ATTEMPTS) { + await sleep(2 ** attempt * 1000); + continue; + } + throw new LastfmError(err.code, err.message); + } +} + +const arr = (v: T | T[] | undefined): T[] => (v === undefined ? [] : Array.isArray(v) ? v : [v]); +const nonEmpty = (s: string | undefined): string | undefined => (s && s.trim() !== "" ? s : undefined); + +function toListen(t: LfmTrack): SourceRecord | null { + const uts = t.date?.uts; + const artist = nonEmpty(t.artist?.["#text"] ?? t.artist?.name); + const track = nonEmpty(t.name); + if (!uts || !artist || !track) return null; + const listen: ListenFields = { listenedAt: new Date(Number(uts) * 1000).toISOString(), artist, track }; + const album = nonEmpty(t.album?.["#text"]); + if (album) listen.album = album; + const mbids: NonNullable = {}; + const rec = nonEmpty(t.mbid), rel = nonEmpty(t.album?.mbid), art = nonEmpty(t.artist?.mbid); + if (rec) mbids.recording = rec; + if (rel) mbids.release = rel; + if (art) mbids.artist = art; + if (Object.keys(mbids).length > 0) listen.mbids = mbids; + return { kind: "listen", key: `${uts}:${artist}:${track}`, raw: t, listen }; +} + +async function* recent(ctx: PluginContext, from: number | undefined): AsyncGenerator { + const user = String(ctx.config.username ?? ""); + let max = from ?? 0; + for (let page = 1, total = 1; page <= total; page++) { + const res = await call<{ recenttracks?: { track?: LfmTrack | LfmTrack[]; "@attr"?: { totalPages?: string } } }>(ctx, { + method: "user.getRecentTracks", user, limit: 200, page, from, + }); + total = Number(res.recenttracks?.["@attr"]?.totalPages ?? 1) || 1; + for (const t of arr(res.recenttracks?.track)) { + if (t["@attr"]?.nowplaying === "true") continue; + const rec = toListen(t); + if (!rec) continue; + max = Math.max(max, Number(t.date?.uts)); + yield rec; + } + } + if (max > 0) await ctx.cursor.set(String(max)); +} + +export async function* pull(ctx: PluginContext): AsyncGenerator { + yield* recent(ctx, undefined); + yield* loved(ctx); +} + +export async function* pullChanges(ctx: PluginContext): AsyncGenerator { + const c = await ctx.cursor.get(); + yield* recent(ctx, c ? Number(c) : undefined); +} + +async function* loved(ctx: PluginContext): AsyncGenerator { + const user = String(ctx.config.username ?? ""); + for (let page = 1, total = 1; page <= total; page++) { + const res = await call<{ lovedtracks?: { track?: LfmTrack | LfmTrack[]; "@attr"?: { totalPages?: string } } }>(ctx, { + method: "user.getLovedTracks", user, limit: 200, page, + }); + total = Number(res.lovedtracks?.["@attr"]?.totalPages ?? 1) || 1; + for (const t of arr(res.lovedtracks?.track)) { + const artist = nonEmpty(t.artist?.name ?? t.artist?.["#text"]); + const title = nonEmpty(t.name); + if (!artist || !title) continue; + const mbid = nonEmpty(t.mbid); + yield { + kind: "entry", + key: `loved:${artist}:${title}`, + raw: t, + work: { + title, + mediaType: "track", + creators: [{ name: artist, role: "artist" }], + externalIds: mbid ? { musicbrainz_recording: mbid } : {}, + }, + entry: { favorite: true }, + }; + } + } +} diff --git a/plugins/lastfm/src/options.ts b/plugins/lastfm/src/options.ts new file mode 100644 index 0000000..61ac6c6 --- /dev/null +++ b/plugins/lastfm/src/options.ts @@ -0,0 +1,8 @@ +/** Injectable sleep. Tests replace it with an immediate one via `setSleep`. */ +let sleepImpl: (ms: number) => Promise = (ms) => Bun.sleep(ms); + +export function setSleep(fn: ((ms: number) => Promise) | null): void { + sleepImpl = fn ?? ((ms) => Bun.sleep(ms)); +} + +export const sleep = (ms: number): Promise => sleepImpl(ms); diff --git a/plugins/lastfm/test/lastfm.test.ts b/plugins/lastfm/test/lastfm.test.ts new file mode 100644 index 0000000..477de15 --- /dev/null +++ b/plugins/lastfm/test/lastfm.test.ts @@ -0,0 +1,92 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { collect, createFixtureHttp, createTestContext } from "@sttracker/plugin-sdk/testing"; +import plugin from "../index.ts"; +import { setSleep } from "../src/options.ts"; + +const sleeps: number[] = []; +beforeEach(() => { sleeps.length = 0; setSleep(async (ms) => { sleeps.push(ms); }); }); +afterEach(() => setSleep(null)); + +const URL0 = "GET https://ws.audioscrobbler.com/2.0/"; +const rt = (name: string, uts: string, mbid = "", extra = {}) => ({ + name, mbid, artist: { "#text": "Art", mbid: "" }, album: { "#text": "Alb", mbid: "" }, date: { uts }, ...extra, +}); + +function routes(handler: (q: URLSearchParams) => unknown) { + return { [URL0]: (req: Request) => handler(new URL(req.url).searchParams) }; +} + +test("skips nowplaying, omits empty mbids, includes loved entries", async () => { + const f = createFixtureHttp(routes((q) => { + expect(q.get("api_key")).toBe("key"); + expect(q.get("format")).toBe("json"); + if (q.get("method") === "user.getLovedTracks") + return { lovedtracks: { track: [{ name: "Fav", mbid: "m9", artist: { name: "Art" } }], "@attr": { totalPages: "1" } } }; + return { recenttracks: { track: [ + { name: "Now", artist: { "#text": "Art" }, "@attr": { nowplaying: "true" } }, + rt("One", "1700000000", "rec1", { artist: { "#text": "Art", mbid: "a1" } }), + rt("Two", "1699999999"), + ], "@attr": { totalPages: "1" } } }; + })); + const ctx = createTestContext({ fetch: f.fetch, config: { username: "u", apiKey: "key" } }); + const recs = await collect(plugin.pull!(ctx)); + expect(recs.map((r) => r.key)).toEqual(["1700000000:Art:One", "1699999999:Art:Two", "loved:Art:Fav"]); + const one = recs[0]!, two = recs[1]!, fav = recs[2]!; + expect(one.kind === "listen" && one.listen.mbids).toEqual({ recording: "rec1", artist: "a1" }); + expect(two.kind === "listen" && two.listen.mbids).toBeUndefined(); + expect(fav.kind === "entry" && fav.entry.favorite).toBe(true); + expect(fav.kind === "entry" && fav.work.externalIds).toEqual({ musicbrainz_recording: "m9" }); + expect(ctx.saved.cursor).toEqual(["1700000000"]); +}); + +test("paginates recent tracks", async () => { + const pagesSeen: (string | null)[] = []; + const f = createFixtureHttp(routes((q) => { + if (q.get("method") === "user.getLovedTracks") return { lovedtracks: { track: [], "@attr": { totalPages: "0" } } }; + pagesSeen.push(q.get("page")); + return { recenttracks: { track: [rt(`T${q.get("page")}`, `10${q.get("page")}`)], "@attr": { totalPages: "2" } } }; + })); + const recs = await collect(plugin.pull!(createTestContext({ fetch: f.fetch, config: { username: "u", apiKey: "k" } }))); + expect(pagesSeen).toEqual(["1", "2"]); + expect(recs).toHaveLength(2); +}); + +test("error 29 is retried with exponential sleeps (body on HTTP 200 and 403)", async () => { + let n = 0; + const f = createFixtureHttp(routes((q) => { + if (q.get("method") === "user.getLovedTracks") return { lovedtracks: { track: [] } }; + n++; + if (n === 1) return { error: 29, message: "Rate Limit Exceeded" }; + if (n === 2) return new Response(JSON.stringify({ error: 29, message: "Rate Limit Exceeded" }), { status: 403 }); + return { recenttracks: { track: rt("X", "5") } }; + })); + const recs = await collect(plugin.pull!(createTestContext({ fetch: f.fetch, config: { username: "u", apiKey: "k" } }))); + expect(n).toBe(3); + expect(sleeps).toEqual([2000, 4000]); + expect(recs).toHaveLength(1); +}); + +test("error 29 gives up after 3 attempts; other codes throw immediately", async () => { + const f29 = createFixtureHttp(routes(() => ({ error: 29, message: "Rate Limit Exceeded" }))); + const p = collect(plugin.pull!(createTestContext({ fetch: f29.fetch, config: { username: "u", apiKey: "k" } }))); + await expect(p).rejects.toMatchObject({ code: 29 }); + expect(f29.calls).toHaveLength(3); + + const f6 = createFixtureHttp(routes(() => new Response(JSON.stringify({ error: 6, message: "User not found" }), { status: 404 }))); + const p6 = collect(plugin.pull!(createTestContext({ fetch: f6.fetch, config: { username: "u", apiKey: "k" } }))); + await expect(p6).rejects.toThrow(/Last\.fm error 6: User not found/); + expect(f6.calls).toHaveLength(1); +}); + +test("pullChanges sends from= and advances it", async () => { + let from: string | null = null; + const f = createFixtureHttp(routes((q) => { + from = q.get("from"); + return { recenttracks: { track: [rt("N", "2000")], "@attr": { totalPages: "1" } } }; + })); + const ctx = createTestContext({ fetch: f.fetch, config: { username: "u", apiKey: "k" } }); + await ctx.cursor.set("1500"); + await collect(plugin.pullChanges!(ctx)); + expect(from as string | null).toBe("1500"); + expect(await ctx.cursor.get()).toBe("2000"); +}); diff --git a/plugins/listenbrainz/index.ts b/plugins/listenbrainz/index.ts new file mode 100644 index 0000000..cc8f901 --- /dev/null +++ b/plugins/listenbrainz/index.ts @@ -0,0 +1,19 @@ +import { definePlugin } from "@sttracker/plugin-sdk/types"; +import { pull, pullChanges } from "./src/listenbrainz.ts"; + +export default definePlugin({ + manifest: { + id: "listenbrainz", + name: "ListenBrainz", + version: "0.1.0", + sdkVersion: 1, + mediaTypes: ["track", "album", "artist"], + capabilities: ["pull", "pullChanges"], + config: [ + { key: "username", label: "Username", type: "string", required: true }, + { key: "token", label: "User token", type: "secret", required: true }, + ], + }, + pull, + pullChanges, +}); diff --git a/plugins/listenbrainz/src/listenbrainz.ts b/plugins/listenbrainz/src/listenbrainz.ts new file mode 100644 index 0000000..7f7935c --- /dev/null +++ b/plugins/listenbrainz/src/listenbrainz.ts @@ -0,0 +1,164 @@ +import { strFromU8, unzipSync } from "fflate"; +import { HttpError } from "@sttracker/plugin-sdk/http"; +import type { ListenFields, PluginContext, SourceRecord } from "@sttracker/plugin-sdk/types"; +import { sleep } from "./options.ts"; + +export const BASE = "https://api.listenbrainz.org"; +export const POLL_INTERVAL_MS = 30_000; +export const MAX_POLLS = 60; + +interface TrackMetadata { + artist_name?: string; + track_name?: string; + release_name?: string; + additional_info?: { duration_ms?: number }; + mbid_mapping?: { recording_mbid?: string; release_mbid?: string; artist_mbids?: string[] }; +} +export interface LbListen { + listened_at: number | string; + recording_msid?: string; + track_metadata?: TrackMetadata; +} +interface LbFeedback { + recording_mbid?: string; + recording_msid?: string; + score?: number; + track_metadata?: TrackMetadata; +} +interface ExportInfo { export_id: number | string; status: string; created?: string } + +async function authHeaders(ctx: PluginContext): Promise> { + const creds = await ctx.credentials.get(); + const token = creds?.token ?? (typeof ctx.config.token === "string" ? ctx.config.token : undefined); + return token ? { authorization: `Token ${token}` } : {}; +} + +/** Unix seconds from either a number or an ISO string; null if unparseable. */ +export function toSeconds(v: number | string | undefined): number | null { + if (typeof v === "number") return Number.isFinite(v) ? Math.floor(v) : null; + if (typeof v === "string" && v.trim() !== "") { + if (/^\d+$/.test(v.trim())) return Number(v); + const ms = Date.parse(v); + return Number.isNaN(ms) ? null : Math.floor(ms / 1000); + } + return null; +} + +export function listenToRecord(l: LbListen): SourceRecord | null { + const secs = toSeconds(l.listened_at); + const tm = l.track_metadata; + if (secs === null || !tm?.artist_name || !tm.track_name) return null; + const listen: ListenFields = { + listenedAt: new Date(secs * 1000).toISOString(), + artist: tm.artist_name, + track: tm.track_name, + }; + if (tm.release_name) listen.album = tm.release_name; + const dur = tm.additional_info?.duration_ms; + if (typeof dur === "number" && dur > 0) listen.durationMs = dur; + const m = tm.mbid_mapping; + const mbids: NonNullable = {}; + if (m?.recording_mbid) mbids.recording = m.recording_mbid; + if (m?.release_mbid) mbids.release = m.release_mbid; + if (m?.artist_mbids?.[0]) mbids.artist = m.artist_mbids[0]; + if (Object.keys(mbids).length > 0) listen.mbids = mbids; + return { kind: "listen", key: `${secs}:${l.recording_msid ?? ""}`, raw: l, listen }; +} + +function feedbackToRecord(f: LbFeedback): SourceRecord | null { + if (f.score !== 1 || !f.recording_mbid) return null; + return { + kind: "entry", + key: `feedback:${f.recording_mbid}`, + raw: f, + work: { + title: f.track_metadata?.track_name ?? f.recording_mbid, + mediaType: "track", + externalIds: { musicbrainz_recording: f.recording_mbid }, + ...(f.track_metadata?.artist_name ? { creators: [{ name: f.track_metadata.artist_name, role: "artist" }] } : {}), + }, + entry: { favorite: true }, + }; +} + +function* jsonlLines(bytes: Uint8Array): Generator { + for (const line of strFromU8(bytes).split("\n")) { + const s = line.trim(); + if (s === "") continue; + yield JSON.parse(s) as T; + } +} + +async function requestExport(ctx: PluginContext, headers: Record): Promise { + try { + await ctx.http.json(`${BASE}/1/export/`, { method: "POST", headers }); + } catch (e) { + if (e instanceof HttpError && e.status === 400 && /pending/i.test(e.body)) return; + throw e; + } +} + +async function waitForExport(ctx: PluginContext, headers: Record): Promise { + for (let poll = 0; poll < MAX_POLLS; poll++) { + if (poll > 0) await sleep(POLL_INTERVAL_MS); + ctx.signal.throwIfAborted(); + const list = await ctx.http.json(`${BASE}/1/export/list`, { headers }); + const newest = [...list].sort((a, b) => (b.created ?? "").localeCompare(a.created ?? ""))[0]; + if (newest?.status === "complete") return newest.export_id; + if (newest?.status === "failed") throw new Error(`ListenBrainz export ${newest.export_id} failed`); + } + throw new Error(`ListenBrainz export did not complete after ${MAX_POLLS} polls`); +} + +export async function* pull(ctx: PluginContext): AsyncGenerator { + const headers = await authHeaders(ctx); + await requestExport(ctx, headers); + const id = await waitForExport(ctx, headers); + const res = await ctx.http.raw(`${BASE}/1/export/${id}/download`, { headers }); + if (!res.ok) throw new HttpError(res.status, await res.text(), res.url); + const files = unzipSync(new Uint8Array(await res.arrayBuffer()), { + filter: (f) => f.name.endsWith(".jsonl"), + }); + let max = 0; + for (const name of Object.keys(files).sort()) { + const bytes = files[name]!; + if (name === "feedback.jsonl" || name.endsWith("/feedback.jsonl")) { + for (const f of jsonlLines(bytes)) { + const rec = feedbackToRecord(f); + if (rec) yield rec; + } + } else if (/(^|\/)listens\/.+\.jsonl$/.test(name)) { + for (const l of jsonlLines(bytes)) { + const rec = listenToRecord(l); + if (!rec) continue; + max = Math.max(max, toSeconds(l.listened_at) ?? 0); + yield rec; + } + } + } + if (max > 0) await ctx.cursor.set(String(max)); +} + +export async function* pullChanges(ctx: PluginContext): AsyncGenerator { + const headers = await authHeaders(ctx); + const username = String(ctx.config.username ?? ""); + let cursor = Number((await ctx.cursor.get()) ?? 0) || 0; + for (;;) { + ctx.signal.throwIfAborted(); + const res = await ctx.http.json<{ payload?: { listens?: LbListen[] } }>( + `${BASE}/1/user/${encodeURIComponent(username)}/listens`, + { headers, query: { min_ts: cursor, count: 1000 } }, + ); + const listens = res.payload?.listens ?? []; + if (listens.length === 0) return; + let max = cursor; + for (const l of listens) { + max = Math.max(max, toSeconds(l.listened_at) ?? 0); + const rec = listenToRecord(l); + if (rec) yield rec; + } + if (max <= cursor) return; + cursor = max; + await ctx.cursor.set(String(cursor)); + } +} diff --git a/plugins/listenbrainz/src/options.ts b/plugins/listenbrainz/src/options.ts new file mode 100644 index 0000000..61ac6c6 --- /dev/null +++ b/plugins/listenbrainz/src/options.ts @@ -0,0 +1,8 @@ +/** Injectable sleep. Tests replace it with an immediate one via `setSleep`. */ +let sleepImpl: (ms: number) => Promise = (ms) => Bun.sleep(ms); + +export function setSleep(fn: ((ms: number) => Promise) | null): void { + sleepImpl = fn ?? ((ms) => Bun.sleep(ms)); +} + +export const sleep = (ms: number): Promise => sleepImpl(ms); diff --git a/plugins/listenbrainz/test/listenbrainz.test.ts b/plugins/listenbrainz/test/listenbrainz.test.ts new file mode 100644 index 0000000..003b4cf --- /dev/null +++ b/plugins/listenbrainz/test/listenbrainz.test.ts @@ -0,0 +1,98 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import { zipSync, strToU8 } from "fflate"; +import { HttpError } from "@sttracker/plugin-sdk/http"; +import { collect, createFixtureHttp, createTestContext } from "@sttracker/plugin-sdk/testing"; +import plugin from "../index.ts"; +import { setSleep } from "../src/options.ts"; + +const sleeps: number[] = []; +beforeEach(() => { sleeps.length = 0; setSleep(async (ms) => { sleeps.push(ms); }); }); +afterEach(() => setSleep(null)); + +const jl = (rows: unknown[]) => strToU8(rows.map((r) => JSON.stringify(r)).join("\n") + "\n"); +const listen = (at: number | string, msid: string, extra = {}) => ({ + listened_at: at, recording_msid: msid, + track_metadata: { artist_name: "A", track_name: `T${msid}`, release_name: "R", additional_info: { duration_ms: 1000 }, ...extra }, +}); +const mapped = { mbid_mapping: { recording_mbid: "rec", release_mbid: "rel", artist_mbids: ["art"] } }; + +function exportRoutes(listRuns: unknown[][], post: () => Response) { + let i = 0; + const zip = zipSync({ + "listens/2024/1.jsonl": jl([listen(1700000000, "m1", mapped), listen("2023-11-14T22:13:20Z", "m2")]), + "feedback.jsonl": jl([ + { recording_mbid: "love1", score: 1, track_metadata: { track_name: "Loved", artist_name: "A" } }, + { recording_mbid: "hate1", score: -1 }, + ]), + }); + return { + "POST https://api.listenbrainz.org/1/export/": post, + "GET https://api.listenbrainz.org/1/export/list": () => Response.json(listRuns[Math.min(i++, listRuns.length - 1)]), + "GET https://api.listenbrainz.org/1/export/7/download": () => new Response(zip), + }; +} +const pending = [{ export_id: 7, status: "in_progress", created: "2024-01-01" }]; +const done = [{ export_id: 6, status: "complete", created: "2023-01-01" }, { export_id: 7, status: "complete", created: "2024-01-01" }]; + +test("pull: polls pending then complete, parses listens and feedback", async () => { + const f = createFixtureHttp(exportRoutes([pending, done], () => Response.json({ export_id: 7 }))); + const ctx = createTestContext({ fetch: f.fetch, config: { username: "u", token: "tok" } }); + const recs = await collect(plugin.pull!(ctx)); + expect(sleeps).toEqual([30_000]); + expect(f.calls[0]!.headers.get("authorization")).toBe("Token tok"); + expect(recs.map((r) => r.key)).toEqual(["feedback:love1", "1700000000:m1", "1700000000:m2"]); + const l = recs.find((r) => r.key === "1700000000:m1")!; + expect(l.kind === "listen" && l.listen).toEqual({ + listenedAt: "2023-11-14T22:13:20.000Z", artist: "A", track: "Tm1", album: "R", durationMs: 1000, + mbids: { recording: "rec", release: "rel", artist: "art" }, + }); + const m2 = recs.find((r) => r.key === "1700000000:m2")!; + expect(m2.kind === "listen" && m2.listen.mbids).toBeUndefined(); + const fav = recs[0]!; + expect(fav.kind === "entry" && fav.entry.favorite).toBe(true); + expect(fav.kind === "entry" && fav.work.externalIds).toEqual({ musicbrainz_recording: "love1" }); + expect(ctx.saved.cursor).toEqual(["1700000000"]); +}); + +test("pull: 400 already pending falls through to listing", async () => { + const f = createFixtureHttp(exportRoutes([done], () => new Response("An export is already pending", { status: 400 }))); + const recs = await collect(plugin.pull!(createTestContext({ fetch: f.fetch, config: { username: "u" } }))); + expect(recs).toHaveLength(3); +}); + +test("pull: other POST errors propagate", async () => { + const f = createFixtureHttp(exportRoutes([done], () => new Response("nope", { status: 401 }))); + const p = collect(plugin.pull!(createTestContext({ fetch: f.fetch, config: { username: "u" } }))); + await expect(p).rejects.toBeInstanceOf(HttpError); +}); + +test("pull: gives up after 60 polls", async () => { + const f = createFixtureHttp(exportRoutes([pending], () => Response.json({ export_id: 7 }))); + const p = collect(plugin.pull!(createTestContext({ fetch: f.fetch, config: { username: "u" } }))); + await expect(p).rejects.toThrow(/60 polls/); + expect(sleeps).toHaveLength(59); +}); + +test("pullChanges: pages until empty, cursor advances to max listened_at", async () => { + const pages = [ + { payload: { listens: [listen(200, "b"), listen(150, "a")] } }, + { payload: { listens: [listen(300, "c")] } }, + { payload: { listens: [] } }, + ]; + const mins: (string | null)[] = []; + let i = 0; + const f = createFixtureHttp({ + "GET https://api.listenbrainz.org/1/user/u/listens": (req: Request) => { + const u = new URL(req.url); + mins.push(u.searchParams.get("min_ts")); + expect(u.searchParams.get("count")).toBe("1000"); + return pages[i++]; + }, + }); + const ctx = createTestContext({ fetch: f.fetch, config: { username: "u" } }); + await ctx.cursor.set("100"); + const recs = await collect(plugin.pullChanges!(ctx)); + expect(recs.map((r) => r.key)).toEqual(["200:b", "150:a", "300:c"]); + expect(mins).toEqual(["100", "200", "300"]); + expect(await ctx.cursor.get()).toBe("300"); +});