fix(plugins): walk listenbrainz listens backwards with max_ts

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

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" } } };

plugins/listenbrainz/src/listenbrainz.ts

@@ -142,23 +142,42 @@ export async function* pull(ctx: PluginContext): AsyncGenerator<SourceRecord> {
 export async function* pullChanges(ctx: PluginContext): AsyncGenerator<SourceRecord> {
   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<string>();
+  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));
 }

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"]);
 });