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 <[email protected]>

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);