feat: Event emitting part 1/N (Fulfillment) (#7391)

**What**
Add support for event emitting in the fulfillment module

**NOTE**
It does not include the review of the events for the abstract module factory if the method is not implemented in the module itself and rely on the default implementation

packages/modules/fulfillment/src/models/geo-zone.ts

@@ -115,7 +115,7 @@ export default class GeoZone {
 
   @BeforeCreate()
   onCreate() {
-    this.id = generateEntityId(this.id, " fgz")
+    this.id = generateEntityId(this.id, "fgz")
     this.service_zone_id ??= this.service_zone?.id
   }
 

packages/modules/fulfillment/src/services/fulfillment-module-service.ts

@@ -15,15 +15,16 @@ import {
 } from "@medusajs/types"
 import {
   arrayDifference,
+  deepEqualObj,
   EmitEvents,
-  FulfillmentUtils,
   getSetDifference,
   InjectManager,
   InjectTransactionManager,
+  isDefined,
+  isPresent,
   isString,
   MedusaContext,
   MedusaError,
-  Modules,
   ModulesSdkUtils,
   promiseAll,
 } from "@medusajs/utils"
@@ -38,10 +39,18 @@ import {
   ShippingOptionType,
   ShippingProfile,
 } from "@models"
-import { isContextValid, validateAndNormalizeRules } from "@utils"
+import {
+  buildCreatedFulfillmentEvents,
+  buildCreatedFulfillmentSetEvents,
+  buildCreatedServiceZoneEvents,
+  eventBuilders,
+  isContextValid,
+  validateAndNormalizeRules,
+} from "@utils"
 import { entityNameToLinkableKeysMap, joinerConfig } from "../joiner-config"
 import { UpdateShippingOptionsInput } from "../types/service"
 import FulfillmentProviderService from "./fulfillment-provider"
+import { buildCreatedShippingOptionEvents } from "../utils/events"
 
 const generateMethodForModels = [
   ServiceZone,
@@ -271,9 +280,13 @@ export default class FulfillmentModuleService<
   > {
     const createdFulfillmentSets = await this.create_(data, sharedContext)
 
+    const returnedFulfillmentSets = Array.isArray(data)
+      ? createdFulfillmentSets
+      : createdFulfillmentSets[0]
+
     return await this.baseRepository_.serialize<
       FulfillmentTypes.FulfillmentSetDTO | FulfillmentTypes.FulfillmentSetDTO[]
-    >(createdFulfillmentSets)
+    >(returnedFulfillmentSets)
   }
 
   @InjectTransactionManager("baseRepository_")
@@ -282,7 +295,7 @@ export default class FulfillmentModuleService<
       | FulfillmentTypes.CreateFulfillmentSetDTO
       | FulfillmentTypes.CreateFulfillmentSetDTO[],
     @MedusaContext() sharedContext: Context = {}
-  ): Promise<TEntity | TEntity[]> {
+  ): Promise<TEntity[]> {
     const data_ = Array.isArray(data) ? data : [data]
 
     if (!data_.length) {
@@ -304,14 +317,12 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
-    this.aggregateFulfillmentSetCreatedEvents(
-      createdFulfillmentSets,
-      sharedContext
-    )
+    buildCreatedFulfillmentSetEvents({
+      fulfillmentSets: createdFulfillmentSets,
+      sharedContext,
+    })
 
-    return Array.isArray(data)
-      ? createdFulfillmentSets
-      : createdFulfillmentSets[0]
+    return createdFulfillmentSets
   }
 
   createServiceZones(
@@ -324,6 +335,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ServiceZoneDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createServiceZones(
     data:
       | FulfillmentTypes.CreateServiceZoneDTO[]
@@ -339,7 +351,7 @@ export default class FulfillmentModuleService<
 
     return await this.baseRepository_.serialize<
       FulfillmentTypes.ServiceZoneDTO | FulfillmentTypes.ServiceZoneDTO[]
-    >(createdServiceZones)
+    >(Array.isArray(data) ? createdServiceZones : createdServiceZones[0])
   }
 
   @InjectTransactionManager("baseRepository_")
@@ -348,7 +360,7 @@ export default class FulfillmentModuleService<
       | FulfillmentTypes.CreateServiceZoneDTO[]
       | FulfillmentTypes.CreateServiceZoneDTO,
     @MedusaContext() sharedContext: Context = {}
-  ): Promise<TServiceZoneEntity | TServiceZoneEntity[]> {
+  ): Promise<TServiceZoneEntity[]> {
     const data_ = Array.isArray(data) ? data : [data]
 
     if (!data_.length) {
@@ -368,7 +380,12 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
-    return Array.isArray(data) ? createdServiceZones : createdServiceZones[0]
+    buildCreatedServiceZoneEvents({
+      serviceZones: createdServiceZones,
+      sharedContext,
+    })
+
+    return createdServiceZones
   }
 
   createShippingOptions(
@@ -381,6 +398,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingOptionDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createShippingOptions(
     data:
       | FulfillmentTypes.CreateShippingOptionDTO[]
@@ -396,7 +414,7 @@ export default class FulfillmentModuleService<
 
     return await this.baseRepository_.serialize<
       FulfillmentTypes.ShippingOptionDTO | FulfillmentTypes.ShippingOptionDTO[]
-    >(createdShippingOptions)
+    >(Array.isArray(data) ? createdShippingOptions : createdShippingOptions[0])
   }
 
   @InjectTransactionManager("baseRepository_")
@@ -405,7 +423,7 @@ export default class FulfillmentModuleService<
       | FulfillmentTypes.CreateShippingOptionDTO[]
       | FulfillmentTypes.CreateShippingOptionDTO,
     @MedusaContext() sharedContext: Context = {}
-  ): Promise<TShippingOptionEntity | TShippingOptionEntity[]> {
+  ): Promise<TShippingOptionEntity[]> {
     const data_ = Array.isArray(data) ? data : [data]
 
     if (!data_.length) {
@@ -417,14 +435,17 @@ export default class FulfillmentModuleService<
       validateAndNormalizeRules(rules as Record<string, unknown>[])
     }
 
-    const createdShippingOptions = await this.shippingOptionService_.create(
+    const createdSO = await this.shippingOptionService_.create(
       data_,
       sharedContext
     )
 
-    return Array.isArray(data)
-      ? createdShippingOptions
-      : createdShippingOptions[0]
+    buildCreatedShippingOptionEvents({
+      shippingOptions: createdSO,
+      sharedContext,
+    })
+
+    return createdSO
   }
 
   createShippingProfiles(
@@ -437,6 +458,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingProfileDTO>
 
   @InjectTransactionManager("baseRepository_")
+  @EmitEvents()
   async createShippingProfiles(
     data:
       | FulfillmentTypes.CreateShippingProfileDTO[]
@@ -450,10 +472,17 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
+    eventBuilders.createdShippingProfile({
+      data: createdShippingProfiles,
+      sharedContext,
+    })
+
     return await this.baseRepository_.serialize<
       | FulfillmentTypes.ShippingProfileDTO
       | FulfillmentTypes.ShippingProfileDTO[]
-    >(createdShippingProfiles)
+    >(
+      Array.isArray(data) ? createdShippingProfiles : createdShippingProfiles[0]
+    )
   }
 
   @InjectTransactionManager("baseRepository_")
@@ -462,21 +491,14 @@ export default class FulfillmentModuleService<
       | FulfillmentTypes.CreateShippingProfileDTO[]
       | FulfillmentTypes.CreateShippingProfileDTO,
     @MedusaContext() sharedContext: Context = {}
-  ): Promise<TShippingProfileEntity[] | TShippingProfileEntity> {
+  ): Promise<TShippingProfileEntity[]> {
     const data_ = Array.isArray(data) ? data : [data]
 
     if (!data_.length) {
       return []
     }
 
-    const createdShippingProfiles = await this.shippingProfileService_.create(
-      data_,
-      sharedContext
-    )
-
-    return Array.isArray(data)
-      ? createdShippingProfiles
-      : createdShippingProfiles[0]
+    return await this.shippingProfileService_.create(data_, sharedContext)
   }
 
   createGeoZones(
@@ -489,6 +511,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.GeoZoneDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createGeoZones(
     data:
       | FulfillmentTypes.CreateGeoZoneDTO
@@ -504,6 +527,11 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
+    eventBuilders.createdGeoZone({
+      data: createdGeoZones,
+      sharedContext,
+    })
+
     return await this.baseRepository_.serialize<FulfillmentTypes.GeoZoneDTO[]>(
       Array.isArray(data) ? createdGeoZones : createdGeoZones[0]
     )
@@ -519,6 +547,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingOptionRuleDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createShippingOptionRules(
     data:
       | FulfillmentTypes.CreateShippingOptionRuleDTO[]
@@ -536,7 +565,11 @@ export default class FulfillmentModuleService<
     return await this.baseRepository_.serialize<
       | FulfillmentTypes.ShippingOptionRuleDTO
       | FulfillmentTypes.ShippingOptionRuleDTO[]
-    >(createdShippingOptionRules)
+    >(
+      Array.isArray(data)
+        ? createdShippingOptionRules
+        : createdShippingOptionRules[0]
+    )
   }
 
   @InjectTransactionManager("baseRepository_")
@@ -545,7 +578,7 @@ export default class FulfillmentModuleService<
       | FulfillmentTypes.CreateShippingOptionRuleDTO[]
       | FulfillmentTypes.CreateShippingOptionRuleDTO,
     @MedusaContext() sharedContext: Context = {}
-  ): Promise<TShippingOptionRuleEntity | TShippingOptionRuleEntity[]> {
+  ): Promise<TShippingOptionRuleEntity[]> {
     const data_ = Array.isArray(data) ? data : [data]
 
     if (!data_.length) {
@@ -554,15 +587,21 @@ export default class FulfillmentModuleService<
 
     validateAndNormalizeRules(data_ as unknown as Record<string, unknown>[])
 
-    const createdShippingOptionRules =
-      await this.shippingOptionRuleService_.create(data_, sharedContext)
+    const createdSORules = await this.shippingOptionRuleService_.create(
+      data_,
+      sharedContext
+    )
 
-    return Array.isArray(data)
-      ? createdShippingOptionRules
-      : createdShippingOptionRules[0]
+    eventBuilders.createdShippingOptionRule({
+      data: createdSORules.map((sor) => ({ id: sor.id })),
+      sharedContext,
+    })
+
+    return createdSORules
   }
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createFulfillment(
     data: FulfillmentTypes.CreateFulfillmentDTO,
     @MedusaContext() sharedContext: Context = {}
@@ -603,12 +642,18 @@ export default class FulfillmentModuleService<
       throw error
     }
 
+    buildCreatedFulfillmentEvents({
+      fulfillments: [fulfillment],
+      sharedContext,
+    })
+
     return await this.baseRepository_.serialize<FulfillmentTypes.FulfillmentDTO>(
       fulfillment
     )
   }
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async createReturnFulfillment(
     data: FulfillmentTypes.CreateFulfillmentDTO,
     @MedusaContext() sharedContext: Context = {}
@@ -639,6 +684,11 @@ export default class FulfillmentModuleService<
       throw error
     }
 
+    buildCreatedFulfillmentEvents({
+      fulfillments: [fulfillment],
+      sharedContext,
+    })
+
     return await this.baseRepository_.serialize<FulfillmentTypes.FulfillmentDTO>(
       fulfillment
     )
@@ -654,6 +704,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.FulfillmentSetDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async update(
     data: UpdateFulfillmentSetDTO[] | UpdateFulfillmentSetDTO,
     @MedusaContext() sharedContext: Context = {}
@@ -714,9 +765,11 @@ export default class FulfillmentModuleService<
       fulfillmentSets.map((f) => [f.id, f])
     )
 
-    // find service zones to delete
     const serviceZoneIdsToDelete: string[] = []
     const geoZoneIdsToDelete: string[] = []
+    const existingServiceZoneIds: string[] = []
+    const existingGeoZoneIds: string[] = []
+
     data_.forEach((fulfillmentSet) => {
       if (fulfillmentSet.service_zones) {
         /**
@@ -734,6 +787,7 @@ export default class FulfillmentModuleService<
               .filter((id): id is string => !!id)
           )
         )
+
         if (toDeleteServiceZoneIds.size) {
           serviceZoneIdsToDelete.push(...Array.from(toDeleteServiceZoneIds))
           geoZoneIdsToDelete.push(
@@ -758,11 +812,13 @@ export default class FulfillmentModuleService<
             .map((s) => "id" in s && s.id)
             .filter((id): id is string => !!id)
         )
+
         const expectedServiceZoneSet = new Set(
           fulfillmentSet.service_zones
             .map((s) => "id" in s && s.id)
             .filter((id): id is string => !!id)
         )
+
         const missingServiceZoneIds = getSetDifference(
           expectedServiceZoneSet,
           serviceZonesSet
@@ -789,6 +845,16 @@ export default class FulfillmentModuleService<
                 }
                 return serviceZone
               }
+
+              const existingServiceZone = serviceZonesMap.get(serviceZone.id)!
+              existingServiceZoneIds.push(existingServiceZone.id)
+
+              if (existingServiceZone.geo_zones.length) {
+                existingGeoZoneIds.push(
+                  ...existingServiceZone.geo_zones.map((g) => g.id)
+                )
+              }
+
               return serviceZonesMap.get(serviceZone.id)!
             }
           )
@@ -797,6 +863,15 @@ export default class FulfillmentModuleService<
     })
 
     if (serviceZoneIdsToDelete.length) {
+      eventBuilders.deletedServiceZone({
+        data: serviceZoneIdsToDelete.map((id) => ({ id })),
+        sharedContext,
+      })
+      eventBuilders.deletedGeoZone({
+        data: geoZoneIdsToDelete.map((id) => ({ id })),
+        sharedContext,
+      })
+
       await promiseAll([
         this.geoZoneService_.delete(
           {
@@ -818,6 +893,32 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
+    eventBuilders.updatedFulfillmentSet({
+      data: updatedFulfillmentSets,
+      sharedContext,
+    })
+
+    const createdServiceZoneIds: string[] = []
+    const createdGeoZoneIds = updatedFulfillmentSets
+      .flatMap((f) =>
+        [...f.service_zones].flatMap((serviceZone) => {
+          if (!existingServiceZoneIds.includes(serviceZone.id)) {
+            createdServiceZoneIds.push(serviceZone.id)
+          }
+          return serviceZone.geo_zones.map((g) => g.id)
+        })
+      )
+      .filter((id) => !existingGeoZoneIds.includes(id))
+
+    eventBuilders.createdServiceZone({
+      data: createdServiceZoneIds.map((id) => ({ id })),
+      sharedContext,
+    })
+    eventBuilders.createdGeoZone({
+      data: createdGeoZoneIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
     return Array.isArray(data)
       ? updatedFulfillmentSets
       : updatedFulfillmentSets[0]
@@ -835,6 +936,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ServiceZoneDTO[]>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateServiceZones(
     idOrSelector: string | FulfillmentTypes.FilterableServiceZoneProps,
     data: FulfillmentTypes.UpdateServiceZoneDTO,
@@ -926,20 +1028,28 @@ export default class FulfillmentModuleService<
     )
 
     const geoZoneIdsToDelete: string[] = []
+    const existingGeoZoneIds: string[] = []
+    const updatedGeoZoneIds: string[] = []
 
     data_.forEach((serviceZone) => {
       if (serviceZone.geo_zones) {
         const existingServiceZone = serviceZoneMap.get(serviceZone.id!)!
         const existingGeoZones = existingServiceZone.geo_zones
         const updatedGeoZones = serviceZone.geo_zones
+        const existingGeoZoneIdsForServiceZone = existingGeoZones.map(
+          (g) => g.id
+        )
         const toDeleteGeoZoneIds = getSetDifference(
-          new Set(existingGeoZones.map((g) => g.id)),
+          new Set(existingGeoZoneIdsForServiceZone),
           new Set(
             updatedGeoZones
               .map((g) => "id" in g && g.id)
               .filter((id): id is string => !!id)
           )
         )
+
+        existingGeoZoneIds.push(...existingGeoZoneIdsForServiceZone)
+
         if (toDeleteGeoZoneIds.size) {
           geoZoneIdsToDelete.push(...Array.from(toDeleteGeoZoneIds))
         }
@@ -978,12 +1088,25 @@ export default class FulfillmentModuleService<
           }
           const existing = geoZonesMap.get(geoZone.id)!
 
+          // If only the id is provided we dont consider it as an update
+          if (
+            Object.keys(geoZone).length > 1 &&
+            !deepEqualObj(existing, geoZone)
+          ) {
+            updatedGeoZoneIds.push(geoZone.id)
+          }
+
           return { ...existing, ...geoZone }
         })
       }
     })
 
     if (geoZoneIdsToDelete.length) {
+      eventBuilders.deletedGeoZone({
+        data: geoZoneIdsToDelete.map((id) => ({ id })),
+        sharedContext,
+      })
+
       await this.geoZoneService_.delete(
         {
           id: geoZoneIdsToDelete,
@@ -997,6 +1120,26 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
+    eventBuilders.updatedServiceZone({
+      data: updatedServiceZones,
+      sharedContext,
+    })
+
+    const createdGeoZoneIds = updatedServiceZones
+      .flatMap((serviceZone) => {
+        return serviceZone.geo_zones.map((g) => g.id)
+      })
+      .filter((id) => !existingGeoZoneIds.includes(id))
+
+    eventBuilders.createdGeoZone({
+      data: createdGeoZoneIds.map((id) => ({ id })),
+      sharedContext,
+    })
+    eventBuilders.updatedGeoZone({
+      data: updatedGeoZoneIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
     return Array.isArray(data) ? updatedServiceZones : updatedServiceZones[0]
   }
 
@@ -1010,11 +1153,12 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ServiceZoneDTO[]>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async upsertServiceZones(
     data:
       | FulfillmentTypes.UpsertServiceZoneDTO
       | FulfillmentTypes.UpsertServiceZoneDTO[],
-    sharedContext?: Context
+    @MedusaContext() sharedContext: Context = {}
   ): Promise<
     FulfillmentTypes.ServiceZoneDTO | FulfillmentTypes.ServiceZoneDTO[]
   > {
@@ -1055,9 +1199,11 @@ export default class FulfillmentModuleService<
         forCreate,
         sharedContext
       )
+
       const toPush = Array.isArray(createdServiceZones)
         ? createdServiceZones
         : [createdServiceZones]
+
       created.push(...toPush)
     }
 
@@ -1066,9 +1212,11 @@ export default class FulfillmentModuleService<
         forUpdate,
         sharedContext
       )
+
       const toPush = Array.isArray(updatedServiceZones)
         ? updatedServiceZones
         : [updatedServiceZones]
+
       updated.push(...toPush)
     }
 
@@ -1087,6 +1235,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingOptionDTO[]>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateShippingOptions(
     idOrSelector: string | FulfillmentTypes.FilterableShippingOptionProps,
     data: FulfillmentTypes.UpdateShippingOptionDTO,
@@ -1142,7 +1291,7 @@ export default class FulfillmentModuleService<
         id: shippingOptionIds,
       },
       {
-        relations: ["rules"],
+        relations: ["rules", "type"],
         take: shippingOptionIds.length,
       },
       sharedContext
@@ -1157,16 +1306,28 @@ export default class FulfillmentModuleService<
     )
 
     const ruleIdsToDelete: string[] = []
+    const updatedRuleIds: string[] = []
+    const existingRuleIds: string[] = []
+
+    const optionTypeDeletedIds: string[] = []
+
     dataArray.forEach((shippingOption) => {
+      const existingShippingOption = existingShippingOptions.get(
+        shippingOption.id
+      )! // Garuantueed to exist since the validation above have been performed
+
+      if (shippingOption.type && !("id" in shippingOption.type)) {
+        optionTypeDeletedIds.push(existingShippingOption.type.id)
+      }
+
       if (!shippingOption.rules) {
         return
       }
 
-      const existingShippingOption = existingShippingOptions.get(
-        shippingOption.id
-      )! // Garuantueed to exist since the validation above have been performed
       const existingRules = existingShippingOption.rules
 
+      existingRuleIds.push(...existingRules.map((r) => r.id))
+
       FulfillmentModuleService.validateMissingShippingOptionRules(
         existingShippingOption,
         shippingOption
@@ -1183,6 +1344,10 @@ export default class FulfillmentModuleService<
             const existingRule = (existingRulesMap.get(rule.id) ??
               {}) as FulfillmentTypes.UpdateShippingOptionRuleDTO
 
+            if (existingRulesMap.get(rule.id)) {
+              updatedRuleIds.push(rule.id)
+            }
+
             const ruleData: FulfillmentTypes.UpdateShippingOptionRuleDTO = {
               ...existingRule,
               ...rule,
@@ -1198,8 +1363,6 @@ export default class FulfillmentModuleService<
 
       validateAndNormalizeRules(updatedRules)
 
-      const updatedRuleIds = updatedRules.map((r) => "id" in r && r.id)
-
       const toDeleteRuleIds = arrayDifference(
         updatedRuleIds,
         Array.from(existingRulesMap.keys())
@@ -1219,6 +1382,11 @@ export default class FulfillmentModuleService<
     })
 
     if (ruleIdsToDelete.length) {
+      eventBuilders.deletedShippingOptionRule({
+        data: ruleIdsToDelete.map((id) => ({ id })),
+        sharedContext,
+      })
+
       await this.shippingOptionRuleService_.delete(
         ruleIdsToDelete,
         sharedContext
@@ -1230,11 +1398,71 @@ export default class FulfillmentModuleService<
       sharedContext
     )
 
+    this.handleShippingOptionUpdateEvents({
+      shippingOptionsData: dataArray,
+      updatedShippingOptions,
+      optionTypeDeletedIds,
+      updatedRuleIds,
+      existingRuleIds,
+      sharedContext,
+    })
+
     return Array.isArray(data)
       ? updatedShippingOptions
       : updatedShippingOptions[0]
   }
 
+  private handleShippingOptionUpdateEvents({
+    shippingOptionsData,
+    updatedShippingOptions,
+    optionTypeDeletedIds,
+    updatedRuleIds,
+    existingRuleIds,
+    sharedContext,
+  }) {
+    eventBuilders.updatedShippingOption({
+      data: updatedShippingOptions,
+      sharedContext,
+    })
+    eventBuilders.deletedShippingOptionType({
+      data: optionTypeDeletedIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
+    const createdOptionTypeIds = updatedShippingOptions
+      .filter((so) => {
+        const updateData = shippingOptionsData.find((sod) => sod.id === so.id)
+        return updateData?.type && !("id" in updateData.type)
+      })
+      .map((so) => so.type.id)
+
+    eventBuilders.createdShippingOptionType({
+      data: createdOptionTypeIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
+    const createdRuleIds = updatedShippingOptions
+      .flatMap((so) =>
+        [...so.rules].map((rule) => {
+          if (existingRuleIds.includes(rule.id)) {
+            return
+          }
+
+          return rule.id
+        })
+      )
+      .filter((id): id is string => !!id)
+
+    eventBuilders.createdShippingOptionRule({
+      data: createdRuleIds.map((id) => ({ id })),
+      sharedContext,
+    })
+    eventBuilders.updatedShippingOptionRule({
+      data: updatedRuleIds.map((id) => ({ id })),
+      sharedContext,
+    })
+  }
+
   async upsertShippingOptions(
     data: FulfillmentTypes.UpsertShippingOptionDTO[],
     sharedContext?: Context
@@ -1245,6 +1473,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingOptionDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async upsertShippingOptions(
     data:
       | FulfillmentTypes.UpsertShippingOptionDTO[]
@@ -1329,6 +1558,7 @@ export default class FulfillmentModuleService<
   ): Promise<
     FulfillmentTypes.ShippingProfileDTO | FulfillmentTypes.ShippingProfileDTO[]
   > {
+    // TODO: should we implement that or can we get rid of the profiles concept entirely and link to the so instead?
     return []
   }
 
@@ -1342,6 +1572,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.GeoZoneDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateGeoZones(
     data:
       | FulfillmentTypes.UpdateGeoZoneDTO
@@ -1357,10 +1588,15 @@ export default class FulfillmentModuleService<
     FulfillmentModuleService.validateGeoZones(data_)
 
     const updatedGeoZones = await this.geoZoneService_.update(
-      data,
+      data_,
       sharedContext
     )
 
+    eventBuilders.updatedGeoZone({
+      data: updatedGeoZones,
+      sharedContext,
+    })
+
     const serialized = await this.baseRepository_.serialize<
       FulfillmentTypes.GeoZoneDTO[]
     >(updatedGeoZones)
@@ -1378,6 +1614,7 @@ export default class FulfillmentModuleService<
   ): Promise<FulfillmentTypes.ShippingOptionRuleDTO>
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateShippingOptionRules(
     data:
       | FulfillmentTypes.UpdateShippingOptionRuleDTO[]
@@ -1416,31 +1653,140 @@ export default class FulfillmentModuleService<
     const updatedShippingOptionRules =
       await this.shippingOptionRuleService_.update(data_, sharedContext)
 
+    eventBuilders.updatedShippingOptionRule({
+      data: updatedShippingOptionRules.map((rule) => ({ id: rule.id })),
+      sharedContext,
+    })
+
     return Array.isArray(data)
       ? updatedShippingOptionRules
       : updatedShippingOptionRules[0]
   }
 
-  @InjectTransactionManager("baseRepository_")
+  @InjectManager("baseRepository_")
+  @EmitEvents()
   async updateFulfillment(
     id: string,
     data: FulfillmentTypes.UpdateFulfillmentDTO,
     @MedusaContext() sharedContext: Context = {}
   ): Promise<FulfillmentTypes.FulfillmentDTO> {
-    const fulfillment = await this.fulfillmentService_.update(
-      { id, ...data },
+    const fulfillment = await this.updateFulfillment_(id, data, sharedContext)
+
+    return await this.baseRepository_.serialize<FulfillmentTypes.FulfillmentDTO>(
+      fulfillment
+    )
+  }
+
+  @InjectTransactionManager("baseRepository_")
+  protected async updateFulfillment_(
+    id: string,
+    data: FulfillmentTypes.UpdateFulfillmentDTO,
+    @MedusaContext() sharedContext: Context
+  ): Promise<TFulfillmentEntity> {
+    const existingFulfillment: TFulfillmentEntity =
+      await this.fulfillmentService_.retrieve(
+        id,
+        {
+          relations: ["items", "labels"],
+        },
+        sharedContext
+      )
+
+    const updatedLabelIds: string[] = []
+    let deletedLabelIds: string[] = []
+
+    const existingLabelIds = existingFulfillment.labels.map((label) => label.id)
+
+    /**
+     * @note
+     * Since the relation is a one to many, the deletion, update and creation of labels
+     * is handled b the orm. That means that we dont have to perform any manual deletions or update.
+     * For some reason we use to have upsert and replace handled manually but we could simplify all that just like
+     * we do below which will create the label, update some and delete the one that does not exists in the new data.
+     *
+     * There is a bit of logic as we need to reassign the data of those we want to keep
+     * and we also need to emit the events later on.
+     */
+    if (isDefined(data.labels) && isPresent(data.labels)) {
+      const dataLabelIds: string[] = data.labels
+        .filter((label): label is { id: string } => "id" in label)
+        .map((label) => label.id)
+
+      deletedLabelIds = arrayDifference(existingLabelIds, dataLabelIds)
+
+      for (let label of data.labels) {
+        if (!("id" in label)) {
+          continue
+        }
+
+        const existingLabel = existingFulfillment.labels.find(
+          ({ id }) => id === label.id
+        )!
+
+        if (
+          !existingLabel ||
+          Object.keys(label).length === 1 ||
+          deepEqualObj(existingLabel, label)
+        ) {
+          continue
+        }
+
+        updatedLabelIds.push(label.id)
+        const labelData = { ...label }
+        Object.assign(label, existingLabel, labelData)
+      }
+    }
+
+    const [fulfillment] = await this.fulfillmentService_.update(
+      [{ id, ...data }],
       sharedContext
     )
 
-    const serialized =
-      await this.baseRepository_.serialize<FulfillmentTypes.FulfillmentDTO>(
-        fulfillment
-      )
+    this.handleFulfillmentUpdateEvents(
+      fulfillment,
+      existingLabelIds,
+      updatedLabelIds,
+      deletedLabelIds,
+      sharedContext
+    )
 
-    return Array.isArray(serialized) ? serialized[0] : serialized
+    return fulfillment
+  }
+
+  private handleFulfillmentUpdateEvents(
+    fulfillment: Fulfillment,
+    existingLabelIds: string[],
+    updatedLabelIds: string[],
+    deletedLabelIds: string[],
+    sharedContext: Context
+  ) {
+    eventBuilders.updatedFulfillment({
+      data: [{ id: fulfillment.id }],
+      sharedContext,
+    })
+
+    eventBuilders.deletedFulfillmentLabel({
+      data: deletedLabelIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
+    eventBuilders.updatedFulfillmentLabel({
+      data: updatedLabelIds.map((id) => ({ id })),
+      sharedContext,
+    })
+
+    const createdLabels = fulfillment.labels.filter((label) => {
+      return !existingLabelIds.includes(label.id)
+    })
+
+    eventBuilders.createdFulfillmentLabel({
+      data: createdLabels.map((label) => ({ id: label.id })),
+      sharedContext,
+    })
   }
 
   @InjectManager("baseRepository_")
+  @EmitEvents()
   async cancelFulfillment(
     id: string,
     @MedusaContext() sharedContext: Context = {}
@@ -1473,6 +1819,11 @@ export default class FulfillmentModuleService<
         },
         sharedContext
       )
+
+      eventBuilders.updatedFulfillment({
+        data: [{ id }],
+        sharedContext,
+      })
     }
 
     const result = await this.baseRepository_.serialize(fulfillment)
@@ -1791,60 +2142,4 @@ export default class FulfillmentModuleService<
 
     return geoZoneConstraints
   }
-
-  protected aggregateFulfillmentSetCreatedEvents(
-    createdFulfillmentSets: TEntity[],
-    sharedContext: Context
-  ): void {
-    const buildMessage = ({
-      eventName,
-      id,
-      object,
-    }: {
-      eventName: string
-      id: string
-      object: string
-    }) => {
-      return {
-        eventName,
-        metadata: {
-          object,
-          service: Modules.FULFILLMENT,
-          action: "created",
-          eventGroupId: sharedContext.eventGroupId,
-        },
-        data: { id },
-      }
-    }
-
-    for (const fulfillmentSet of createdFulfillmentSets) {
-      sharedContext.messageAggregator!.saveRawMessageData(
-        buildMessage({
-          eventName: FulfillmentUtils.FulfillmentEvents.created,
-          id: fulfillmentSet.id,
-          object: "fulfillment_set",
-        })
-      )
-
-      for (const serviceZone of fulfillmentSet.service_zones ?? []) {
-        sharedContext.messageAggregator!.saveRawMessageData(
-          buildMessage({
-            eventName: FulfillmentUtils.FulfillmentEvents.service_zone_created,
-            id: serviceZone.id,
-            object: "service_zone",
-          })
-        )
-
-        for (const geoZone of serviceZone.geo_zones ?? []) {
-          sharedContext.messageAggregator!.saveRawMessageData(
-            buildMessage({
-              eventName: FulfillmentUtils.FulfillmentEvents.geo_zone_created,
-              id: geoZone.id,
-              object: "geo_zone",
-            })
-          )
-        }
-      }
-    }
-  }
 }

packages/modules/fulfillment/src/utils/events.ts

@@ -0,0 +1,293 @@
+import {
+  Fulfillment,
+  FulfillmentSet,
+  GeoZone,
+  ServiceZone,
+  ShippingOption,
+  ShippingOptionRule,
+  ShippingOptionType,
+} from "@models"
+import { Context } from "@medusajs/types"
+import {
+  CommonEvents,
+  eventBuilderFactory,
+  FulfillmentEvents,
+  Modules,
+} from "@medusajs/utils"
+
+export const eventBuilders = {
+  createdFulfillment: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "fulfillment",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedFulfillment: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "fulfillment",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdFulfillmentAddress: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "fulfillment_address",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdFulfillmentItem: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "fulfillment_item",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdFulfillmentLabel: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "fulfillment_label",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedFulfillmentLabel: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "fulfillment_label",
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedFulfillmentLabel: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "fulfillment_label",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdShippingProfile: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "shipping_profile",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdShippingOptionType: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "shipping_option_type",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedShippingOptionType: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "shipping_option_type",
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedShippingOptionType: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "shipping_option_type",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdShippingOptionRule: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "shipping_option_rule",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedShippingOptionRule: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "shipping_option_rule",
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedShippingOptionRule: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "shipping_option_rule",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdShippingOption: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "shipping_option",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedShippingOption: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "shipping_option",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdFulfillmentSet: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "fulfillment_set",
+    isMainEntity: true,
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedFulfillmentSet: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "fulfillment_set",
+    isMainEntity: true,
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedFulfillmentSet: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "fulfillment_set",
+    isMainEntity: true,
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdServiceZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "service_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedServiceZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "service_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedServiceZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "service_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+  createdGeoZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.CREATED,
+    object: "geo_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+  updatedGeoZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.UPDATED,
+    object: "geo_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+  deletedGeoZone: eventBuilderFactory({
+    service: Modules.FULFILLMENT,
+    action: CommonEvents.DELETED,
+    object: "geo_zone",
+    eventsEnum: FulfillmentEvents,
+  }),
+}
+
+export function buildCreatedFulfillmentEvents({
+  fulfillments,
+  sharedContext,
+}: {
+  fulfillments: Fulfillment[]
+  sharedContext: Context
+}) {
+  if (!fulfillments.length) {
+    return
+  }
+
+  const fulfillments_: { id: string }[] = []
+  const addresses: { id: string }[] = []
+  const items: { id: string }[] = []
+  const labels: { id: string }[] = []
+
+  fulfillments.forEach((fulfillment) => {
+    fulfillments_.push({ id: fulfillment.id })
+
+    if (fulfillment.delivery_address) {
+      addresses.push({ id: fulfillment.delivery_address.id })
+    }
+
+    if (fulfillment.items) {
+      items.push(...fulfillment.items)
+    }
+
+    if (fulfillment.labels) {
+      labels.push(...fulfillment.labels)
+    }
+  })
+
+  eventBuilders.createdFulfillment({ data: fulfillments_, sharedContext })
+  eventBuilders.createdFulfillmentAddress({ data: addresses, sharedContext })
+  eventBuilders.createdFulfillmentItem({ data: items, sharedContext })
+  eventBuilders.createdFulfillmentLabel({ data: labels, sharedContext })
+}
+
+export function buildCreatedShippingOptionEvents({
+  shippingOptions,
+  sharedContext,
+}: {
+  shippingOptions: ShippingOption[]
+  sharedContext: Context
+}) {
+  if (!shippingOptions.length) {
+    return
+  }
+
+  const options: { id: string }[] = []
+  const types: ShippingOptionType[] = []
+  const rules: ShippingOptionRule[] = []
+
+  shippingOptions.forEach((shippingOption) => {
+    options.push({ id: shippingOption.id })
+
+    if (shippingOption.type) {
+      types.push(shippingOption.type)
+    }
+
+    if (shippingOption.rules) {
+      rules.push(...shippingOption.rules)
+    }
+  })
+
+  eventBuilders.createdShippingOption({ data: options, sharedContext })
+  eventBuilders.createdShippingOptionType({ data: types, sharedContext })
+  eventBuilders.createdShippingOptionRule({ data: rules, sharedContext })
+}
+
+export function buildCreatedFulfillmentSetEvents({
+  fulfillmentSets,
+  sharedContext,
+}: {
+  fulfillmentSets: FulfillmentSet[]
+  sharedContext: Context
+}): void {
+  if (!fulfillmentSets.length) {
+    return
+  }
+
+  const serviceZones: ServiceZone[] = []
+
+  fulfillmentSets.forEach((fulfillmentSet) => {
+    if (!fulfillmentSet.service_zones?.length) {
+      return
+    }
+
+    serviceZones.push(...fulfillmentSet.service_zones)
+  })
+
+  eventBuilders.createdFulfillmentSet({ data: fulfillmentSets, sharedContext })
+
+  buildCreatedServiceZoneEvents({ serviceZones, sharedContext })
+}
+
+export function buildCreatedServiceZoneEvents({
+  serviceZones,
+  sharedContext,
+}: {
+  serviceZones: ServiceZone[]
+  sharedContext: Context
+}): void {
+  if (!serviceZones.length) {
+    return
+  }
+
+  const geoZones: GeoZone[] = []
+
+  serviceZones.forEach((serviceZone) => {
+    if (!serviceZone.geo_zones.length) {
+      return
+    }
+
+    geoZones.push(...serviceZone.geo_zones)
+  })
+
+  eventBuilders.createdServiceZone({ data: serviceZones, sharedContext })
+  eventBuilders.createdGeoZone({ data: geoZones, sharedContext })
+}

packages/modules/fulfillment/src/utils/index.ts

@@ -1 +1,2 @@
-export * from './utils'
+export * from "./utils"
+export * from "./events"