From 00742937c9b0d9326cf5440556266f4b2ceee7cf Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 15:51:58 +0100 Subject: [PATCH] fix(plugins): walk listenbrainz listens backwards with max_ts Co-Authored-By: Claude Sonnet 5.5 --- plugins/lastfm/test/lastfm.test.ts | 2 +- plugins/listenbrainz/src/listenbrainz.ts | 37 +++++++++++---- .../listenbrainz/test/listenbrainz.test.ts | 45 ++++++++++++------- 3 files changed, 59 insertions(+), 25 deletions(-) diff --git a/plugins/lastfm/test/lastfm.test.ts b/plugins/lastfm/test/lastfm.test.ts index 477de15..b87a904 100644 --- a/plugins/lastfm/test/lastfm.test.ts +++ b/plugins/lastfm/test/lastfm.test.ts @@ -23,7 +23,7 @@ test("skips nowplaying, omits empty mbids, includes loved entries", async () => 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" } }, + { name: "Now", artist: { "#text": "Art" }, date: { uts: "1700000005" }, "@attr": { nowplaying: "true" } }, rt("One", "1700000000", "rec1", { artist: { "#text": "Art", mbid: "a1" } }), rt("Two", "1699999999"), ], "@attr": { totalPages: "1" } } }; diff --git a/plugins/listenbrainz/src/listenbrainz.ts b/plugins/listenbrainz/src/listenbrainz.ts index 7f7935c..a9f99c7 100644 --- a/plugins/listenbrainz/src/listenbrainz.ts +++ b/plugins/listenbrainz/src/listenbrainz.ts @@ -142,23 +142,42 @@ export async function* pull(ctx: PluginContext): AsyncGenerator { 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; + const cursor = Number((await ctx.cursor.get()) ?? 0) || 0; + // We do not rely on the ordering of `min_ts` queries (and LB forbids min_ts together with max_ts). + // Instead walk backwards from now with `max_ts` (listens older than max_ts, newest first) until we + // reach a listen at or before the cursor, or run out of pages. + // max_ts = smallest + 1 re-includes the page-boundary second so listens sharing that second are not + // lost; the duplicates are dropped by the seen-set here and by idempotent keys downstream. + // If that would not make progress (a whole page in one second), fall back to max_ts = smallest. + const seen = new Set(); + let max = cursor; + let maxTs: number | undefined; 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 } }, + { headers, query: { max_ts: maxTs, count: 1000 } }, ); const listens = res.payload?.listens ?? []; - if (listens.length === 0) return; - let max = cursor; + if (listens.length === 0) break; + let smallest = Infinity; + let reachedCursor = false; for (const l of listens) { - max = Math.max(max, toSeconds(l.listened_at) ?? 0); + const secs = toSeconds(l.listened_at); + if (secs === null) continue; + smallest = Math.min(smallest, secs); + if (secs <= cursor) { reachedCursor = true; continue; } const rec = listenToRecord(l); - if (rec) yield rec; + if (!rec || seen.has(rec.key)) continue; + seen.add(rec.key); + max = Math.max(max, secs); + yield rec; } - if (max <= cursor) return; - cursor = max; - await ctx.cursor.set(String(cursor)); + if (reachedCursor || !Number.isFinite(smallest)) break; + let next = smallest + 1; + if (maxTs !== undefined && next >= maxTs) next = smallest; + if (maxTs !== undefined && next >= maxTs) break; + maxTs = next; } + if (max > cursor) await ctx.cursor.set(String(max)); } diff --git a/plugins/listenbrainz/test/listenbrainz.test.ts b/plugins/listenbrainz/test/listenbrainz.test.ts index 003b4cf..3dee08e 100644 --- a/plugins/listenbrainz/test/listenbrainz.test.ts +++ b/plugins/listenbrainz/test/listenbrainz.test.ts @@ -73,26 +73,41 @@ test("pull: gives up after 60 polls", async () => { 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)[] = []; +function changesFixture(pages: unknown[][], seen: (string | null)[]) { let i = 0; - const f = createFixtureHttp({ + return 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++]; + expect(u.searchParams.get("min_ts")).toBeNull(); + seen.push(u.searchParams.get("max_ts")); + return { payload: { listens: pages[i++] ?? [] } }; }, }); +} + +test("pullChanges: walks back over 3 pages with max_ts, cursor ends at max", async () => { + const seen: (string | null)[] = []; + const f = changesFixture( + [ + [listen(600, "f"), listen(500, "e")], + [listen(500, "e2"), listen(400, "d")], + [listen(300, "c"), listen(200, "b")], + [listen(100, "a"), listen(50, "z")], + ], + seen, + ); const ctx = createTestContext({ fetch: f.fetch, config: { username: "u" } }); - await ctx.cursor.set("100"); + await ctx.cursor.set("150"); 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"); + expect(recs.map((r) => r.key)).toEqual(["600:f", "500:e", "500:e2", "400:d", "300:c", "200:b"]); + expect(seen).toEqual([null, "501", "401", "201"]); + expect(ctx.saved.cursor).toEqual(["150", "600"]); +}); + +test("pullChanges: unchanged cursor yields nothing", async () => { + const f = changesFixture([[listen(600, "f"), listen(500, "e")]], []); + const ctx = createTestContext({ fetch: f.fetch, config: { username: "u" } }); + await ctx.cursor.set("600"); + expect(await collect(plugin.pullChanges!(ctx))).toEqual([]); + expect(ctx.saved.cursor).toEqual(["600"]); });