fix(sync): plug outbox hole via seek pagination

e2e/src/query-params-validation.spec.ts

@@ -76,7 +76,14 @@ describe('Query param validation', () => {
 
             messages.forEach((response) => {
                 if (expectedNextObjectCount !== null && expectedNextObjectCount >= 0) {
-                    expect(response.remaining_objects).toEqual(expectedNextObjectCount);
+                    // carrier-details paginates per id_reference (one ref = one
+                    // page, all its rows emitted together), so remaining_objects
+                    // does not decrease by `limit` rows per page.
+                    if (shopContent === 'carrier-details') {
+                        expect(response.remaining_objects).toBeLessThanOrEqual(expectedNextObjectCount + limit);
+                    } else {
+                        expect(response.remaining_objects).toEqual(expectedNextObjectCount);
+                    }
                 }
 
                 expectedNextObjectCount = response.remaining_objects - limit;

sql/install.sql

@@ -1,7 +1,7 @@
 CREATE TABLE IF NOT EXISTS `PREFIX_eventbus_type_sync`
 (
     `type`               VARCHAR(50)      NOT NULL,
-    `offset`             INT(10) UNSIGNED NOT NULL DEFAULT 0,
+    `last_seek_key`      VARCHAR(190)              DEFAULT NULL,
     `id_shop`            INT(10) UNSIGNED NOT NULL,
     `lang_iso`           VARCHAR(3),
     `full_sync_finished` TINYINT(1)       NOT NULL DEFAULT 0,

src/Repository/AbstractRepository.php

@@ -162,4 +162,18 @@ private function debugQuery()
 
         CommonService::exitWithResponse($response);
     }
+
+    /**
+     * Split a composite seek key "a-b" into its two int components.
+     *
+     * @param string $seekKey
+     *
+     * @return array{0: int, 1: int}
+     */
+    protected function decodeCompositeSeekKey($seekKey)
+    {
+        $parts = explode('-', $seekKey, 2);
+
+        return [(int) $parts[0], isset($parts[1]) ? (int) $parts[1] : 0];
+    }
 }

src/Repository/BundleRepository.php

@@ -60,12 +60,15 @@ public function generateFullQuery($langIso, $withSelecParameters)
                 ->select('pac.id_product_pack as id_bundle')
                 ->select('pac.id_product_attribute_item as id_product_attribute')
                 ->select('pac.quantity')
+                ->orderBy('pac.id_product_pack ASC')
             ;
         }
     }
 
     /**
-     * @param int $offset
+     * Seek-based page: returns rows strictly after $lastSeekKey, ordered by pac.id_product_pack.
+     *
+     * @param string|null $lastSeekKey
      * @param int $limit
      * @param string $langIso
      *
@@ -74,11 +77,15 @@ public function generateFullQuery($langIso, $withSelecParameters)
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function retrieveContentsForFull($offset, $limit, $langIso)
+    public function retrieveContentsForFull($lastSeekKey, $limit, $langIso)
     {
         $this->generateFullQuery($langIso, true);
 
-        $this->query->limit((int) $limit, (int) $offset);
+        if ($lastSeekKey !== null) {
+            $this->query->where('pac.id_product_pack > ' . (int) $lastSeekKey);
+        }
+
+        $this->query->limit((int) $limit);
 
         return $this->runQuery();
     }
@@ -106,23 +113,28 @@ public function retrieveContentsForIncremental($limit, $contentIds, $langIso)
     }
 
     /**
-     * @param int $offset
-     * @param int $limit
+     * Count rows with pac.id_product_pack > cursor.
+     *
+     * @param string|null $lastSeekKey
      * @param string $langIso
      *
      * @return int
      *
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function countFullSyncContentLeft($offset, $limit, $langIso)
+    public function countFullSyncContentLeft($lastSeekKey, $langIso)
     {
         $this->generateFullQuery($langIso, false);
 
-        $this->query->select('(COUNT(*) - ' . (int) $offset . ') as count');
+        if ($lastSeekKey !== null) {
+            $this->query->where('pac.id_product_pack > ' . (int) $lastSeekKey);
+        }
+
+        $this->query->select('COUNT(*) as count');
 
         $result = $this->runQuery(true);
 
-        return !empty($result[0]['count']) ? $result[0]['count'] : 0;
+        return !empty($result[0]['count']) ? (int) $result[0]['count'] : 0;
     }
 }

src/Repository/CarrierDetailRepository.php

@@ -127,20 +127,37 @@ public function generateFullQuery($langIso, $withSelecParameters)
     }
 
     /**
-     * @param int $offset
-     * @param int $limit
+     * Seek per id_reference: one page emits every (id_zone, id_range) row for
+     * the next $limit references after the cursor. $limit applies to the
+     * reference count, not the row count.
+     *
+     * @param string|null $lastSeekKey
+     * @param int $limit max number of references emitted in this page
      * @param string $langIso
      *
      * @return array<mixed>
      *
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function retrieveContentsForFull($offset, $limit, $langIso)
+    public function retrieveContentsForFull($lastSeekKey, $limit, $langIso)
     {
         $this->generateFullQuery($langIso, true);
 
-        $this->query->limit((int) $limit, (int) $offset);
+        $refs = $this->db->executeS('
+            SELECT DISTINCT ca.id_reference
+              FROM ' . _DB_PREFIX_ . self::TABLE_NAME . ' ca
+             WHERE ca.id_reference > ' . (int) $lastSeekKey . '
+             ORDER BY ca.id_reference ASC
+             LIMIT ' . (int) $limit . '
+        ');
+
+        if (!is_array($refs) || empty($refs)) {
+            return [];
+        }
+
+        $ids = array_map('intval', array_column($refs, 'id_reference'));
+        $this->query->where('ca.id_reference IN (' . implode(',', $ids) . ')');
 
         return $this->runQuery();
     }
@@ -168,24 +185,26 @@ public function retrieveContentsForIncremental($limit, $contentIds, $langIso)
     }
 
     /**
-     * @param int $offset
-     * @param int $limit
+     * Remaining grouped rows for references strictly after the cursor.
+     *
+     * @param string|null $lastSeekKey
      * @param string $langIso
      *
      * @return int
      *
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function countFullSyncContentLeft($offset, $limit, $langIso)
+    public function countFullSyncContentLeft($lastSeekKey, $langIso)
     {
         $this->generateFullQuery($langIso, true);
+        $this->query->where('ca.id_reference > ' . (int) $lastSeekKey);
 
         $result = $this->db->executeS('
-            SELECT COUNT(*) - ' . (int) $offset . ' AS count
+            SELECT COUNT(*) AS count
                 FROM (' . $this->query->build() . ') as subquery;
         ');
 
-        return is_array($result) ? $result[0]['count'] : 0;
+        return is_array($result) && isset($result[0]['count']) ? (int) $result[0]['count'] : 0;
     }
 }

src/Repository/CarrierRepository.php

@@ -79,12 +79,15 @@ public function generateFullQuery($langIso, $withSelecParameters)
                 ->select('c.grade')
                 ->select('cl.delay AS delay')
                 ->select('c.shipping_handling')
+                ->orderBy('c.id_carrier ASC')
             ;
         }
     }
 
     /**
-     * @param int $offset
+     * Seek-based page: returns rows strictly after $lastSeekKey, ordered by c.id_carrier.
+     *
+     * @param string|null $lastSeekKey
      * @param int $limit
      * @param string $langIso
      *
@@ -93,11 +96,15 @@ public function generateFullQuery($langIso, $withSelecParameters)
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function retrieveContentsForFull($offset, $limit, $langIso)
+    public function retrieveContentsForFull($lastSeekKey, $limit, $langIso)
     {
         $this->generateFullQuery($langIso, true);
 
-        $this->query->limit((int) $limit, (int) $offset);
+        if ($lastSeekKey !== null) {
+            $this->query->where('c.id_carrier > ' . (int) $lastSeekKey);
+        }
+
+        $this->query->limit((int) $limit);
 
         return $this->runQuery();
     }
@@ -124,23 +131,28 @@ public function retrieveContentsForIncremental($limit, $contentIds, $langIso)
     }
 
     /**
-     * @param int $offset
-     * @param int $limit
+     * Count rows with c.id_carrier > cursor.
+     *
+     * @param string|null $lastSeekKey
      * @param string $langIso
      *
      * @return int
      *
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function countFullSyncContentLeft($offset, $limit, $langIso)
+    public function countFullSyncContentLeft($lastSeekKey, $langIso)
     {
         $this->generateFullQuery($langIso, false);
 
-        $this->query->select('(COUNT(*) - ' . (int) $offset . ') as count');
+        if ($lastSeekKey !== null) {
+            $this->query->where('c.id_carrier > ' . (int) $lastSeekKey);
+        }
+
+        $this->query->select('COUNT(*) as count');
 
         $result = $this->runQuery(true);
 
-        return !empty($result[0]['count']) ? $result[0]['count'] : 0;
+        return !empty($result[0]['count']) ? (int) $result[0]['count'] : 0;
     }
 }

src/Repository/CarrierTaxeRepository.php

@@ -56,7 +56,8 @@ public function generateFullQuery($langIso, $withSelecParameters)
             ->leftJoin('state', 's', 'tr.id_state = s.id_state AND s.active = 1')
             ->where('(co.id_zone = d.id_zone OR s.id_zone = d.id_zone)')
             ->select('ca.id_reference')
-            ->groupBy('ca.id_reference, co.id_zone, id_range, country_id')
+            ->select('ca.id_carrier')
+            ->groupBy('ca.id_carrier, ca.id_reference, co.id_zone, id_range, country_id')
         ;
 
         if ($withSelecParameters) {
@@ -79,12 +80,18 @@ public function generateFullQuery($langIso, $withSelecParameters)
                     ) AS state_ids
                 ')
                 ->select('t.rate AS tax_rate')
+                ->orderBy('ca.id_carrier ASC')
             ;
         }
     }
 
     /**
-     * @param int $offset
+     * Seek-based page: returns rows strictly after $lastSeekKey, ordered by ca.id_carrier.
+     * The outbox id_object for carrier_taxes is the parent id_carrier, so pagination
+     * is also by id_carrier — a page may include multiple rows per carrier and
+     * $limit is therefore approximate.
+     *
+     * @param string|null $lastSeekKey
      * @param int $limit
      * @param string $langIso
      *
@@ -93,11 +100,15 @@ public function generateFullQuery($langIso, $withSelecParameters)
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function retrieveContentsForFull($offset, $limit, $langIso)
+    public function retrieveContentsForFull($lastSeekKey, $limit, $langIso)
     {
         $this->generateFullQuery($langIso, true);
 
-        $this->query->limit((int) $limit, (int) $offset);
+        if ($lastSeekKey !== null) {
+            $this->query->where('ca.id_carrier > ' . (int) $lastSeekKey);
+        }
+
+        $this->query->limit((int) $limit);
 
         return $this->runQuery();
     }
@@ -125,24 +136,30 @@ public function retrieveContentsForIncremental($limit, $contentIds, $langIso)
     }
 
     /**
-     * @param int $offset
-     * @param int $limit
+     * Count rows with ca.id_carrier > cursor. Uses a subquery wrapper because
+     * the underlying query has a GROUP BY and aggregate selects.
+     *
+     * @param string|null $lastSeekKey
      * @param string $langIso
      *
      * @return int
      *
      * @throws \PrestaShopException
      * @throws \PrestaShopDatabaseException
      */
-    public function countFullSyncContentLeft($offset, $limit, $langIso)
+    public function countFullSyncContentLeft($lastSeekKey, $langIso)
     {
         $this->generateFullQuery($langIso, true);
 
+        if ($lastSeekKey !== null) {
+            $this->query->where('ca.id_carrier > ' . (int) $lastSeekKey);
+        }
+
         $result = $this->db->executeS('
-            SELECT COUNT(*) - ' . (int) $offset . ' AS count
+            SELECT COUNT(*) AS count
                 FROM (' . $this->query->build() . ') as subquery;
         ');
 
-        return is_array($result) ? $result[0]['count'] : 0;
+        return is_array($result) && isset($result[0]['count']) ? (int) $result[0]['count'] : 0;
     }
 }