From 6efb97eb51cff83c016aaa857d734b1c2fc13a47 Mon Sep 17 00:00:00 2001
From: Lucas Winther <lucwin@ece.au.dk>
Date: Fri, 30 Oct 2026 20:43:08 +0100
Subject: [PATCH] fix(core): batch listen ingestion per 500

One existing-key lookup, one resolve per distinct recording, and one
multi-row insert-or-ignore per batch instead of per-listen queries.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/core/src/sync/listens.ts       | 72 +++++++++++++++----------
 packages/core/test/sync/listens.test.ts | 23 ++++++++
 2 files changed, 66 insertions(+), 29 deletions(-)

diff --git a/packages/core/src/sync/listens.ts b/packages/core/src/sync/listens.ts
index 2eabc61..a1ccd71 100644
--- a/packages/core/src/sync/listens.ts
+++ b/packages/core/src/sync/listens.ts
@@ -31,10 +31,14 @@ async function trackWork(db: Db, recording: string, l: ListenFields, actor: Acto
   }
 }
 
+/** Listens per query/transaction. 500 rows × 12 columns stays well under both dialects' parameter limits. */
+const BATCH = 500;
+
 /**
  * Append listens (spec §6.4): insert-or-ignore on (user, source, source_key), where `source_key` is the
  * record key. A work is linked only through `mbids.recording`. Non-listen records are ignored.
- * No change events are written.
+ * No change events are written. Per batch: one lookup of existing keys, each distinct recording resolved
+ * once (outside the transaction; `createWork` has its own), then one multi-row insert.
  */
 export async function ingestListens(
   db: Db,
@@ -43,38 +47,48 @@ export async function ingestListens(
   records: SourceRecord[],
   actor: Actor = `plugin:${source}:ingest`,
 ): Promise<{ inserted: number }> {
+  const listens = records.filter((r): r is Extract<SourceRecord, { kind: "listen" }> => r.kind === "listen");
   let inserted = 0;
-  for (const r of records) {
-    if (r.kind !== "listen") continue;
-    const exists = await db.k
+  for (let i = 0; i < listens.length; i += BATCH) {
+    const batch = new Map<string, ListenFields>();
+    for (const r of listens.slice(i, i + BATCH)) if (!batch.has(r.key)) batch.set(r.key, r.listen);
+    const existing = await db.k
       .selectFrom("listens")
-      .select("id")
+      .select("source_key")
       .where("user_id", "=", userId)
       .where("source", "=", source)
-      .where("source_key", "=", r.key)
-      .executeTakeFirst();
-    if (exists) continue;
-    const l = r.listen;
-    const recording = l.mbids?.recording;
-    const workId = recording ? await trackWork(db, recording, l, actor) : null;
-    const res = await db.k
-      .insertInto("listens")
-      .values({
-        id: newId(),
-        user_id: userId,
-        listened_at: l.listenedAt,
-        artist: l.artist,
-        track: l.track,
-        album: l.album ?? null,
-        duration_ms: l.durationMs ?? null,
-        mbids: toJson(l.mbids ?? {}),
-        work_id: workId,
-        source,
-        source_key: r.key,
-        created_at: nowIso(),
-      })
-      .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
-      .executeTakeFirst();
+      .where("source_key", "in", [...batch.keys()])
+      .execute();
+    for (const e of existing) batch.delete(e.source_key);
+    if (batch.size === 0) continue;
+
+    const works = new Map<string, string>();
+    for (const l of batch.values()) {
+      const rec = l.mbids?.recording;
+      if (rec && !works.has(rec)) works.set(rec, await trackWork(db, rec, l, actor));
+    }
+    const now = nowIso();
+    const rows = [...batch].map(([key, l]) => ({
+      id: newId(),
+      user_id: userId,
+      listened_at: l.listenedAt,
+      artist: l.artist,
+      track: l.track,
+      album: l.album ?? null,
+      duration_ms: l.durationMs ?? null,
+      mbids: toJson(l.mbids ?? {}),
+      work_id: l.mbids?.recording ? works.get(l.mbids.recording)! : null,
+      source,
+      source_key: key,
+      created_at: now,
+    }));
+    const res = await db.k.transaction().execute((tx) =>
+      tx
+        .insertInto("listens")
+        .values(rows)
+        .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
+        .executeTakeFirst(),
+    );
     inserted += Number(res.numInsertedOrUpdatedRows ?? 0n);
   }
   return { inserted };
diff --git a/packages/core/test/sync/listens.test.ts b/packages/core/test/sync/listens.test.ts
index 55bf8f5..9422875 100644
--- a/packages/core/test/sync/listens.test.ts
+++ b/packages/core/test/sync/listens.test.ts
@@ -45,6 +45,29 @@ forEachDialect("listens", (getDb) => {
     expect(await ingestListens(db, bob, "listenbrainz", records.slice(0, 1))).toEqual({ inserted: 1 });
   });
 
+  test("1200 listens ingest correctly and idempotently", async () => {
+    const db = getDb();
+    const userId = await makeUser(db);
+    const records = Array.from({ length: 1200 }, (_, i) =>
+      listenRec(`k${i}`, {
+        listenedAt: `2024-01-01T00:00:${String(i % 60).padStart(2, "0")}Z`,
+        artist: "A",
+        track: `T${i % 7}`,
+        ...(i % 3 === 0 ? { mbids: { recording: `rec-${i % 7}` } } : {}),
+      }),
+    );
+    // A duplicate key inside the same call is inserted once.
+    records.push(records[5]!);
+    expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 1200 });
+    expect(await count(db, "listens")).toBe(1200);
+    expect(await count(db, "works")).toBe(7);
+    const linked = await db.k.selectFrom("listens").select("work_id").where("work_id", "is not", null).execute();
+    expect(linked).toHaveLength(400);
+    expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 0 });
+    expect(await count(db, "listens")).toBe(1200);
+    expect(await count(db, "works")).toBe(7);
+  });
+
   test("runSync routes listens to ingestListens and counts them", async () => {
     const db = getDb();
     const userId = await makeUser(db);