fix: replace OFFSET with cursor-based pagination on idlink_va (#10)

* fix: eplace OFFSET with cursor-based pagination on idlink_va

* tests

Author: revolunet

src/lib.mjs

@@ -106,15 +106,15 @@ export const fetchValidActionIds = async (
   log = console.log,
 ) => {
   const validActions = new Set();
-  let actionOffset = 0;
+  let lastIdlinkVa = 0;
+  let totalFetched = 0;
 
   log(" - fetching valid action IDs...");
 
   while (true) {
     const batch = await sourceDB
       .selectFrom("matomo_log_link_visit_action")
-      .select(["idaction_name", "idaction_url"])
-      .distinct()
+      .select(["idlink_va", "idaction_name", "idaction_url"])
       .where("idsite", "=", idsite)
       .where("server_time", ">=", new Date("2025-01-01"))
       .where(({ eb }) =>
@@ -123,20 +123,23 @@ export const fetchValidActionIds = async (
           eb("idaction_url", ">", lastActionId),
         ]),
       )
+      .where("idlink_va", ">", lastIdlinkVa)
+      .orderBy("idlink_va")
       .limit(ACTION_BATCH)
-      .offset(actionOffset)
       .execute();
 
     if (batch.length === 0) break;
 
+    lastIdlinkVa = batch[batch.length - 1].idlink_va;
+    totalFetched += batch.length;
+
     for (const r of batch) {
       if (r.idaction_name !== null && r.idaction_name > lastActionId)
         validActions.add(r.idaction_name);
       if (r.idaction_url !== null && r.idaction_url > lastActionId)
         validActions.add(r.idaction_url);
     }
-    actionOffset += ACTION_BATCH;
-    log(` - fetched ${actionOffset} rows, ${validActions.size} unique actions`);
+    log(` - fetched ${totalFetched} rows, ${validActions.size} unique actions`);
   }
 
   return validActions;
@@ -288,7 +291,12 @@ export const importSite = async ({
 
   // Copy actions
   log(`copy matomo_log_action from ${lastActionId}...`);
-  const validActions = await fetchValidActionIds(sourceDB, idsite, lastActionId, log);
+  const validActions = await fetchValidActionIds(
+    sourceDB,
+    idsite,
+    lastActionId,
+    log,
+  );
   await copyActions({
     sourceDB,
     targetDB,

src/lib.test.mjs

@@ -254,10 +254,11 @@ describe("fetchValidActionIds", () => {
     ]);
     assert.strictEqual(sourceDB._calls.select.length, 1);
     assert.deepStrictEqual(sourceDB._calls.select[0], [
+      "idlink_va",
       "idaction_name",
       "idaction_url",
     ]);
-    assert.deepStrictEqual(sourceDB._calls.distinct, [true]);
+    assert.deepStrictEqual(sourceDB._calls.distinct, []); // no distinct
   });
 
   it("should apply idsite and date filters", async () => {
@@ -271,21 +272,48 @@ describe("fetchValidActionIds", () => {
     assert.strictEqual(sourceDB._calls.where[1][1], ">=");
   });
 
-  it("should use pagination with limit and offset", async () => {
+  it("should use cursor-based pagination on idlink_va", async () => {
     const sourceDB = createMockDB({ selectResults: [[]] });
     await fetchValidActionIds(sourceDB, 1, 0, () => {});
 
     assert.deepStrictEqual(sourceDB._calls.limit, [1000]);
-    assert.deepStrictEqual(sourceDB._calls.offset, [0]);
+    assert.deepStrictEqual(sourceDB._calls.offset, []); // no offset
+    assert.deepStrictEqual(sourceDB._calls.orderBy, ["idlink_va"]);
+    // cursor where: idlink_va > 0
+    const cursorWhere = sourceDB._calls.where.find(
+      (w) => Array.isArray(w) && w[0] === "idlink_va",
+    );
+    assert.deepStrictEqual(cursorWhere, ["idlink_va", ">", 0]);
+  });
+
+  it("should advance cursor using last idlink_va of each batch", async () => {
+    const sourceDB = createMockDB({
+      selectResults: [
+        [
+          { idlink_va: 10, idaction_name: 1, idaction_url: 2 },
+          { idlink_va: 20, idaction_name: 3, idaction_url: 4 },
+        ],
+        [], // End pagination
+      ],
+    });
+
+    await fetchValidActionIds(sourceDB, 1, 0, () => {});
+
+    // Second call should use cursor idlink_va > 20
+    const cursorWheres = sourceDB._calls.where.filter(
+      (w) => Array.isArray(w) && w[0] === "idlink_va",
+    );
+    assert.deepStrictEqual(cursorWheres[0], ["idlink_va", ">", 0]);
+    assert.deepStrictEqual(cursorWheres[1], ["idlink_va", ">", 20]);
   });
 
   it("should collect unique action IDs from multiple batches", async () => {
     const sourceDB = createMockDB({
       selectResults: [
         [
-          { idaction_name: 1, idaction_url: 2 },
-          { idaction_name: 1, idaction_url: 3 },
-          { idaction_name: null, idaction_url: 4 },
+          { idlink_va: 1, idaction_name: 1, idaction_url: 2 },
+          { idlink_va: 2, idaction_name: 1, idaction_url: 3 },
+          { idlink_va: 3, idaction_name: null, idaction_url: 4 },
         ],
         [], // End pagination
       ],
@@ -304,8 +332,8 @@ describe("fetchValidActionIds", () => {
     const sourceDB = createMockDB({
       selectResults: [
         [
-          { idaction_name: 5, idaction_url: 10 },
-          { idaction_name: 3, idaction_url: 15 },
+          { idlink_va: 1, idaction_name: 5, idaction_url: 10 },
+          { idlink_va: 2, idaction_name: 3, idaction_url: 15 },
         ],
         [],
       ],