feat: Add support for exporting products in backend (#8214)

CLOSES CC-221
CLOSES CC-223
CLOSES CC-224

packages/core/core-flows/src/notification/index.ts

@@ -0,0 +1 @@
+export * from "./steps"

packages/core/core-flows/src/notification/steps/index.ts

@@ -0,0 +1 @@
+export * from "./send-notifications"

packages/core/core-flows/src/notification/steps/send-notifications.ts

@@ -0,0 +1,32 @@
+import { INotificationModuleService } from "@medusajs/types"
+import { ModuleRegistrationName } from "@medusajs/utils"
+import { StepResponse, createStep } from "@medusajs/workflows-sdk"
+
+type SendNotificationsStepInput = {
+  to: string
+  channel: string
+  template: string
+  data?: Record<string, unknown> | null
+  trigger_type?: string | null
+  resource_id?: string | null
+  resource_type?: string | null
+  receiver_id?: string | null
+  original_notification_id?: string | null
+  idempotency_key?: string | null
+}[]
+
+export const sendNotificationsStepId = "send-notifications"
+export const sendNotificationsStep = createStep(
+  sendNotificationsStepId,
+  async (data: SendNotificationsStepInput, { container }) => {
+    const service = container.resolve<INotificationModuleService>(
+      ModuleRegistrationName.NOTIFICATION
+    )
+    const created = await service.createNotifications(data)
+    return new StepResponse(
+      created,
+      created.map((notification) => notification.id)
+    )
+  }
+  // Most of the notifications are irreversible, so we can't compensate notifications reliably
+)

packages/core/core-flows/src/product/helpers/normalize-for-export.ts

@@ -0,0 +1,133 @@
+import { BigNumberInput, HttpTypes, PricingTypes } from "@medusajs/types"
+import { upperCaseFirst } from "@medusajs/utils"
+
+// We want to have one row per variant, so we need to normalize the data
+export const normalizeForExport = (
+  product: HttpTypes.AdminProduct[]
+): object[] => {
+  const res = product.reduce((acc: object[], product) => {
+    const variants = product.variants ?? []
+    if (!variants.length) {
+      acc.push(normalizeProductForExport(product))
+      return acc
+    }
+
+    variants.forEach((v) => {
+      const toPush = {
+        ...normalizeProductForExport(product),
+        ...normalizeVariantForExport(v),
+      } as any
+      delete toPush["Product Variants"]
+
+      acc.push(toPush)
+    })
+
+    return acc
+  }, [])
+
+  return res
+}
+
+const normalizeProductForExport = (product: HttpTypes.AdminProduct): object => {
+  const flattenedImages = product.images?.reduce(
+    (acc: Record<string, string>, image, idx) => {
+      acc[beautifyKey(`product_image_${idx + 1}`)] = image.url
+      return acc
+    },
+    {}
+  )
+
+  const flattenedTags = product.tags?.reduce(
+    (acc: Record<string, string>, tag, idx) => {
+      acc[beautifyKey(`product_tag_${idx + 1}`)] = tag.value
+      return acc
+    },
+    {}
+  )
+
+  const flattenedSalesChannels = product.sales_channels?.reduce(
+    (acc: Record<string, string>, salesChannel, idx) => {
+      acc[beautifyKey(`product_sales_channel_${idx + 1}`)] = salesChannel.id
+      return acc
+    },
+    {}
+  )
+
+  const res = {
+    ...prefixFields(product, "product"),
+    ...flattenedImages,
+    ...flattenedTags,
+    ...flattenedSalesChannels,
+  } as any
+
+  delete res["Product Images"]
+  delete res["Product Tags"]
+  delete res["Product Sales Channels"]
+
+  // We can decide if we want the metadata in the export and how that would look like
+  delete res["Product Metadata"]
+
+  // We only want the IDs for the type and collection
+  delete res["Product Type"]
+  delete res["Product Collection"]
+
+  // We just rely on the variant options to reconstruct the product options, so we want to
+  // omit the product options to keep the file simpler
+  delete res["Product Options"]
+
+  return res
+}
+
+const normalizeVariantForExport = (
+  variant: HttpTypes.AdminProductVariant & {
+    price_set?: PricingTypes.PriceSetDTO
+  }
+): object => {
+  const flattenedPrices = variant.price_set?.prices
+    ?.sort((a, b) => b.currency_code!.localeCompare(a.currency_code!))
+    .reduce((acc: Record<string, BigNumberInput>, price) => {
+      const regionRule = price.price_rules?.find(
+        (r) => r.attribute === "region"
+      )
+      if (regionRule) {
+        acc[beautifyKey(`variant_price_${regionRule.value}`)] = price.amount!
+      } else if (!price.price_rules?.length) {
+        acc[
+          beautifyKey(`variant_price_${price.currency_code!.toUpperCase()}`)
+        ] = price.amount!
+      }
+      return acc
+    }, {})
+
+  const flattenedOptions = variant.options?.reduce(
+    (acc: Record<string, string>, option, idx) => {
+      acc[beautifyKey(`variant_option_${idx + 1}_name`)] = option.option?.title!
+      acc[beautifyKey(`variant_option_${idx + 1}_value`)] = option.value
+      return acc
+    },
+    {}
+  )
+
+  const res = {
+    ...prefixFields(variant, "variant"),
+    ...flattenedPrices,
+    ...flattenedOptions,
+  } as any
+  delete res["Variant Price Set"]
+  delete res["Variant Options"]
+
+  return res
+}
+
+const prefixFields = (obj: object, prefix: string): object => {
+  const res = {}
+  Object.keys(obj).forEach((key) => {
+    res[beautifyKey(`${prefix}_${key}`)] = obj[key]
+  })
+
+  return res
+}
+
+const beautifyKey = (key: string): string => {
+  return key.split("_").map(upperCaseFirst).join(" ")
+}

packages/core/core-flows/src/product/steps/generate-product-csv.ts

@@ -0,0 +1,36 @@
+import { IFileModuleService, HttpTypes } from "@medusajs/types"
+import { ModuleRegistrationName, convertJsonToCsv } from "@medusajs/utils"
+import { StepResponse, createStep } from "@medusajs/workflows-sdk"
+import { normalizeForExport } from "../helpers/normalize-for-export"
+
+export const generateProductCsvStepId = "generate-product-csv"
+export const generateProductCsvStep = createStep(
+  generateProductCsvStepId,
+  async (products: HttpTypes.AdminProduct[], { container }) => {
+    const normalizedData = normalizeForExport(products)
+    const csvContent = convertJsonToCsv(normalizedData)
+
+    const fileModule: IFileModuleService = container.resolve(
+      ModuleRegistrationName.FILE
+    )
+
+    const filename = `${Date.now()}-product-exports.csv`
+    const file = await fileModule.createFiles({
+      filename,
+      mimeType: "text/csv",
+      content: csvContent,
+    })
+
+    return new StepResponse({ id: file.id, filename }, file.id)
+  },
+  async (fileId, { container }) => {
+    if (!fileId) {
+      return
+    }
+
+    const fileModule: IFileModuleService = container.resolve(
+      ModuleRegistrationName.FILE
+    )
+    await fileModule.deleteFiles(fileId)
+  }
+)

packages/core/core-flows/src/product/steps/get-all-products.ts

@@ -0,0 +1,49 @@
+import { FilterableProductProps, RemoteQueryFunction } from "@medusajs/types"
+import {
+  ContainerRegistrationKeys,
+  remoteQueryObjectFromString,
+} from "@medusajs/utils"
+import { createStep, StepResponse } from "@medusajs/workflows-sdk"
+
+type StepInput = {
+  select: string[]
+  filter?: FilterableProductProps
+}
+
+export const getAllProductsStepId = "get-all-products"
+export const getAllProductsStep = createStep(
+  getAllProductsStepId,
+  async (data: StepInput, { container }) => {
+    const remoteQuery: RemoteQueryFunction = container.resolve(
+      ContainerRegistrationKeys.REMOTE_QUERY
+    )
+
+    const allProducts: any[] = []
+    const pageSize = 200
+    let page = 0
+
+    // We intentionally fetch the products serially here to avoid putting too much load on the DB
+    while (true) {
+      const remoteQueryObject = remoteQueryObjectFromString({
+        entryPoint: "product",
+        variables: {
+          filters: data.filter,
+          skip: page * pageSize,
+          take: pageSize,
+        },
+        fields: data.select,
+      })
+
+      const { rows: products } = await remoteQuery(remoteQueryObject)
+      allProducts.push(...products)
+
+      if (products.length < pageSize) {
+        break
+      }
+
+      page += 1
+    }
+
+    return new StepResponse(allProducts, allProducts)
+  }
+)

packages/core/core-flows/src/product/steps/index.ts

@@ -2,6 +2,7 @@ export * from "./create-products"
 export * from "./update-products"
 export * from "./delete-products"
 export * from "./get-products"
+export * from "./get-all-products"
 export * from "./create-variant-pricing-link"
 export * from "./create-product-options"
 export * from "./update-product-options"
@@ -19,3 +20,4 @@ export * from "./delete-product-types"
 export * from "./create-product-tags"
 export * from "./update-product-tags"
 export * from "./delete-product-tags"
+export * from "./generate-product-csv"

packages/core/core-flows/src/product/workflows/export-products.ts

@@ -1,6 +1,12 @@
-import { WorkflowData, createWorkflow } from "@medusajs/workflows-sdk"
-import { useRemoteQueryStep } from "../../common"
+import {
+  WorkflowData,
+  createWorkflow,
+  transform,
+} from "@medusajs/workflows-sdk"
 import { WorkflowTypes } from "@medusajs/types"
+import { generateProductCsvStep, getAllProductsStep } from "../steps"
+import { useRemoteQueryStep } from "../../common"
+import { sendNotificationsStep } from "../../notification"
 
 export const exportProductsWorkflowId = "export-products"
 export const exportProductsWorkflow = createWorkflow(
@@ -8,11 +14,39 @@ export const exportProductsWorkflow = createWorkflow(
   (
     input: WorkflowData<WorkflowTypes.ProductWorkflow.ExportProductsDTO>
   ): WorkflowData<void> => {
-    const products = useRemoteQueryStep({
-      entry_point: "product",
-      fields: input.select,
-      variables: input.filter,
-      list: true,
+    const products = getAllProductsStep(input).config({
+      async: true,
+      backgroundExecution: true,
     })
+
+    const file = generateProductCsvStep(products)
+    const fileDetails = useRemoteQueryStep({
+      fields: ["id", "url"],
+      entry_point: "file",
+      variables: { id: file.id },
+      list: false,
+    })
+
+    const notifications = transform({ fileDetails, file }, (data) => {
+      return [
+        {
+          // We don't need the recipient here for now, but if we want to push feed notifications to a specific user we could add it.
+          to: "",
+          channel: "feed",
+          template: "admin-ui",
+          data: {
+            title: "Product export",
+            description: "Product export completed successfully!",
+            file: {
+              filename: data.file.filename,
+              url: data.fileDetails.url,
+              mimeType: "text/csv",
+            },
+          },
+        },
+      ]
+    })
+
+    sendNotificationsStep(notifications)
   }
 )

packages/core/medusa-test-utils/src/events.ts

@@ -1,4 +1,5 @@
 import { IEventBusModuleService } from "@medusajs/types"
+import { EventEmitter } from "events"
 
 // Allows you to wait for all subscribers to execute for a given event. Only works with the local event bus.
 export const waitSubscribersExecution = (
@@ -6,23 +7,36 @@ export const waitSubscribersExecution = (
   eventBus: IEventBusModuleService
 ) => {
   const subscriberPromises: Promise<any>[] = []
+  const eventEmitter: EventEmitter = (eventBus as any).eventEmitter_
 
-  ;(eventBus as any).eventEmitter_.listeners(eventName).forEach((listener) => {
-    ;(eventBus as any).eventEmitter_.removeListener("order.created", listener)
-
+  // If there are no existing listeners, resolve once the event happens. Otherwise, wrap the existing subscribers in a promise and resolve once they are done.
+  if (!eventEmitter.listeners(eventName).length) {
     let ok, nok
     const promise = new Promise((resolve, reject) => {
       ok = resolve
       nok = reject
     })
+
     subscriberPromises.push(promise)
+    eventEmitter.on(eventName, ok)
+  } else {
+    eventEmitter.listeners(eventName).forEach((listener: any) => {
+      eventEmitter.removeListener(eventName, listener)
 
-    const newListener = async (...args2) => {
-      return await listener.apply(eventBus, args2).then(ok).catch(nok)
-    }
+      let ok, nok
+      const promise = new Promise((resolve, reject) => {
+        ok = resolve
+        nok = reject
+      })
+      subscriberPromises.push(promise)
 
-    ;(eventBus as any).eventEmitter_.on("order.created", newListener)
-  })
+      const newListener = async (...args2) => {
+        return await listener.apply(eventBus, args2).then(ok).catch(nok)
+      }
+
+      eventEmitter.on(eventName, newListener)
+    })
+  }
 
   return Promise.all(subscriberPromises)
 }

packages/core/modules-sdk/src/definitions.ts

@@ -329,7 +329,7 @@ export const ModulesDefinition: {
     label: upperCaseFirst(ModuleRegistrationName.NOTIFICATION),
     isRequired: false,
     isQueryable: true,
-    dependencies: ["logger"],
+    dependencies: [ModuleRegistrationName.EVENT_BUS, "logger"],
     defaultModuleDeclaration: {
       scope: MODULE_SCOPE.INTERNAL,
       resources: MODULE_RESOURCE_TYPE.SHARED,

packages/core/types/src/pricing/common/money-amount.ts

@@ -1,5 +1,6 @@
 import { BaseFilterable } from "../../dal"
 import { BigNumberInput, BigNumberValue } from "../../totals"
+import { PriceRuleDTO } from "./price-rule"
 
 /**
  * @interface
@@ -27,6 +28,16 @@ export interface MoneyAmountDTO {
    * The maximum quantity required to be purchased for this price to be applied.
    */
   max_quantity?: BigNumberValue
+  /**
+   * The number of rules that apply to this price
+   */
+  rules_count?: number
+
+  /**
+   * The price rules that apply to this price
+   */
+  price_rules?: PriceRuleDTO[]
+
   /**
    * When the money_amount was created.
    */

packages/core/utils/package.json

@@ -38,6 +38,7 @@
     "awilix": "^8.0.1",
     "bignumber.js": "^9.1.2",
     "dotenv": "^16.4.5",
+    "json-2-csv": "^5.5.4",
     "jsonwebtoken": "^9.0.2",
     "knex": "2.4.2",
     "pluralize": "^8.0.0",

packages/core/utils/src/csv/index.ts

@@ -0,0 +1 @@
+export * from "./jsontocsv"

packages/core/utils/src/csv/jsontocsv.ts

@@ -0,0 +1,17 @@
+import { json2csv } from "json-2-csv"
+
+export interface ConvertJsonToCsvOptions<T> {}
+
+export const convertJsonToCsv = <T extends object>(
+  data: T[],
+  options?: ConvertJsonToCsvOptions<T>
+) => {
+  return json2csv(data, {
+    prependHeader: true,
+    arrayIndexesAsKeys: true,
+    expandNestedObjects: true,
+    expandArrayObjects: true,
+    unwindArrays: false,
+    emptyFieldValue: "",
+  })
+}

packages/core/utils/src/index.ts

@@ -26,5 +26,6 @@ export * from "./shipping"
 export * from "./totals"
 export * from "./totals/big-number"
 export * from "./user"
+export * from "./csv"
 
 export const MedusaModuleType = Symbol.for("MedusaModule")

packages/core/utils/src/notification/events.ts

@@ -0,0 +1,9 @@
+import { buildEventNamesFromEntityName } from "../event-bus"
+import { Modules } from "../modules-sdk"
+
+const eventBaseNames: ["notification"] = ["notification"]
+
+export const NotificationEvents = buildEventNamesFromEntityName(
+  eventBaseNames,
+  Modules.NOTIFICATION
+)

packages/core/utils/src/notification/index.ts

@@ -1 +1,2 @@
 export * from "./abstract-notification-provider"
+export * from "./events"