finalize sync process

integration-tests/modules/__tests__/index/sync.spec.ts

@@ -65,13 +65,13 @@ medusaIntegrationTestRunner({
     })
 
     describe("Index engine syncing", () => {
-      it.only("should sync the data to the index based on the indexation configuration", async () => {
+      it("should sync the data to the index based on the indexation configuration", async () => {
         console.info("[Index engine] Creating products")
 
         await populateData(api, {
-          productCount: 10,
-          variantCount: 50,
-          priceCount: 10,
+          productCount: 2,
+          variantCount: 2,
+          priceCount: 2,
         })
 
         console.info("[Index engine] Creating products done")
@@ -102,7 +102,7 @@ medusaIntegrationTestRunner({
 
         console.info("[Index engine] Sync done")
 
-        // 28 ms - 10 + 10 * 50 + 10 * 50 * 10 = 510
+        // 28 ms - 6511 records
         const { data: results } = await indexEngine.query<"product">({
           fields: [
             "product.*",
@@ -111,11 +111,11 @@ medusaIntegrationTestRunner({
           ],
         })
 
-        expect(results.length).toBe(10)
+        expect(results.length).toBe(2)
         for (const result of results) {
-          expect(result.variants.length).toBe(50)
+          expect(result.variants.length).toBe(2)
           for (const variant of result.variants) {
-            expect(variant.prices.length).toBe(10)
+            expect(variant.prices.length).toBe(2)
           }
         }
       })
@@ -216,6 +216,17 @@ medusaIntegrationTestRunner({
 
       expect(updatedResults.length).toBe(1)
       expect(updatedResults[0].variants.length).toBe(1)
+
+      let staledRaws = await dbConnection.raw(
+        'SELECT * FROM "index_data" WHERE "staled_at" IS NOT NULL'
+      )
+
+      expect(staledRaws.rows.length).toBe(0)
+
+      staledRaws = await dbConnection.raw(
+        'SELECT * FROM "index_relation" WHERE "staled_at" IS NOT NULL'
+      )
+      expect(staledRaws.rows.length).toBe(0)
     })
   },
 })

packages/modules/index/src/services/data-synchronizer.ts

@@ -1,6 +1,7 @@
 import {
   CommonEvents,
   ContainerRegistrationKeys,
+  groupBy,
   Modules,
   promiseAll,
 } from "@medusajs/framework/utils"
@@ -44,6 +45,7 @@ export class DataSynchronizer {
     return this.#container.indexDataService
   }
 
+  // @ts-ignore
   get #indexRelationService(): ModulesSdkTypes.IMedusaInternalService<any> {
     return this.#container.indexRelationService
   }
@@ -103,28 +105,48 @@ export class DataSynchronizer {
 
     const staleCondition = staleOnly ? { staled_at: { $ne: null } } : {}
 
-    // Clean up staled data
-    await promiseAll([
-      this.#indexRelationService.delete({
-        selector: {
-          ...staleCondition,
-          $or: [
-            {
-              parent_name: entities,
+    const dataToDelete = await this.#indexDataService.list({
+      ...staleCondition,
+      name: entities,
+    })
+
+    const toDeleteByEntity = groupBy(dataToDelete, "name")
+
+    for (const entity of toDeleteByEntity.keys()) {
+      const records = toDeleteByEntity.get(entity)
+      const ids = records?.map(
+        (record: { data: { id: string } }) => record.data.id
+      )
+      if (!ids?.length) {
+        continue
+      }
+
+      if (this.#schemaObjectRepresentation[entity]) {
+        // Here we assume that some data have been deleted from from the source and we are cleaning since they are still staled in the index and we remove them from the index
+
+        // TODO: expand storage provider interface
+        await (this.#storageProvider as any).onDelete({
+          entity,
+          data: ids,
+          schemaEntityObjectRepresentation:
+            this.#schemaObjectRepresentation[entity],
+        })
+      } else {
+        // Here we assume that the entity is not indexed anymore as it is not part of the schema object representation and we are cleaning the index
+        await promiseAll([
+          this.#indexDataService.delete({
+            selector: {
+              name: entity,
             },
-            {
-              child_name: entities,
+          }),
+          this.#indexRelationService.delete({
+            selector: {
+              $or: [{ parent_id: entity }, { child_id: entity }],
             },
-          ],
-        },
-      }),
-      this.#indexDataService.delete({
-        selector: {
-          ...staleCondition,
-          name: entities,
-        },
-      }),
-    ])
+          }),
+        ])
+      }
+    }
   }
 
   async #updatedStatus(entity: string, status: IndexMetadataStatus) {
@@ -157,21 +179,6 @@ export class DataSynchronizer {
           name: entity,
         },
       }),
-      this.#indexRelationService.update({
-        data: {
-          staled_at: new Date(),
-        },
-        selector: {
-          $or: [
-            {
-              parent_name: entity,
-            },
-            {
-              child_name: entity,
-            },
-          ],
-        },
-      }),
     ])
 
     const finalAcknoledgement = await this.syncEntity({

packages/modules/index/src/services/postgres-provider.ts

@@ -326,9 +326,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
    * @protected
    */
   @InjectTransactionManager()
-  protected async onCreate<
-    TData extends { id: string; [key: string]: unknown }
-  >(
+  async onCreate<TData extends { id: string; [key: string]: unknown }>(
     {
       entity,
       data,
@@ -434,9 +432,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
    * @protected
    */
   @InjectTransactionManager()
-  protected async onUpdate<
-    TData extends { id: string; [key: string]: unknown }
-  >(
+  async onUpdate<TData extends { id: string; [key: string]: unknown }>(
     {
       entity,
       data,
@@ -480,9 +476,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
    * @protected
    */
   @InjectTransactionManager()
-  protected async onDelete<
-    TData extends { id: string; [key: string]: unknown }
-  >(
+  async onDelete<TData extends { id: string; [key: string]: unknown }>(
     {
       entity,
       data,
@@ -534,9 +528,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
    * @protected
    */
   @InjectTransactionManager()
-  protected async onAttach<
-    TData extends { id: string; [key: string]: unknown }
-  >(
+  async onAttach<TData extends { id: string; [key: string]: unknown }>(
     {
       entity,
       data,
@@ -677,9 +669,7 @@ export class PostgresProvider implements IndexTypes.StorageProvider {
    * @protected
    */
   @InjectTransactionManager()
-  protected async onDetach<
-    TData extends { id: string; [key: string]: unknown }
-  >(
+  async onDetach<TData extends { id: string; [key: string]: unknown }>(
     {
       entity,
       data,