feat: order export and upload stream (#14243)

* feat: order export

* Merge branch 'develop' of https://github.com/medusajs/medusa into feat/order-export

* normalize status

* rm util

* serialize totals

* test

* lock

* comments

* configurable order list

packages/core/core-flows/src/order/steps/export-orders.ts

@@ -0,0 +1,178 @@
+import {
+  FilterableOrderProps,
+  IFileModuleService,
+  OrderDTO,
+} from "@medusajs/framework/types"
+import {
+  ContainerRegistrationKeys,
+  Modules,
+  deduplicate,
+} from "@medusajs/framework/utils"
+import { StepResponse, createStep } from "@medusajs/framework/workflows-sdk"
+import { json2csv } from "json-2-csv"
+
+import {
+  getLastFulfillmentStatus,
+  getLastPaymentStatus,
+} from "../utils/aggregate-status"
+
+export type ExportOrdersStepInput = {
+  batch_size?: number | string
+  select: string[]
+  filter?: FilterableOrderProps
+}
+
+export type ExportOrdersStepOutput = {
+  id: string
+  filename: string
+}
+
+export const exportOrdersStepId = "export-orders"
+
+const normalizeOrderForExport = (order: OrderDTO): object => {
+  const order_ = order as any
+  const customer = order_.customer || {}
+  const shippingAddress = order_.shipping_address || {}
+
+  return JSON.parse(
+    JSON.stringify({
+      Order_ID: order.id,
+      Display_ID: order.display_id,
+      "Order status": order.status,
+      Date: order.created_at,
+      "Customer First name": customer.first_name || "",
+      "Customer Last name": customer.last_name || "",
+      "Customer Email": customer.email || "",
+      "Customer ID": customer.id || "",
+      "Shipping Address 1": shippingAddress.address_1 || "",
+      "Shipping Address 2": shippingAddress.address_2 || "",
+      "Shipping Country Code": shippingAddress.country_code || "",
+      "Shipping City": shippingAddress.city || "",
+      "Shipping Postal Code": shippingAddress.postal_code || "",
+      "Shipping Region ID": order.region_id,
+      "Fulfillment Status": order_.fulfillment_status,
+      "Payment Status": order_.payment_status,
+      Subtotal: order.subtotal,
+      "Shipping Total": order.shipping_total,
+      "Discount Total": order.discount_total,
+      "Gift Card Total": order.gift_card_total,
+      "Refunded Total": order_.refunded_total,
+      "Tax Total": order.tax_total,
+      Total: order.total,
+      "Currency Code": order.currency_code,
+    })
+  )
+}
+
+export const exportOrdersStep = createStep(
+  exportOrdersStepId,
+  async (input: ExportOrdersStepInput, { container }) => {
+    const query = container.resolve(ContainerRegistrationKeys.QUERY)
+    const fileModule = container.resolve(Modules.FILE)
+
+    const filename = `${Date.now()}-order-exports.csv`
+    const { writeStream, promise, fileKey } = await fileModule.getUploadStream({
+      filename,
+      mimeType: "text/csv",
+    })
+
+    const pageSize = !isNaN(parseInt(input?.batch_size as string))
+      ? parseInt(input?.batch_size as string, 10)
+      : 50
+
+    let page = 0
+    let hasHeader = false
+
+    const fields = deduplicate([
+      ...input.select,
+      "id",
+      "status",
+      "items.*",
+      "customer.*",
+      "shipping_address.*",
+      "payment_collections.status",
+      "payment_collections.amount",
+      "payment_collections.captured_amount",
+      "payment_collections.refunded_amount",
+      "fulfillments.packed_at",
+      "fulfillments.shipped_at",
+      "fulfillments.delivered_at",
+      "fulfillments.canceled_at",
+    ])
+
+    while (true) {
+      const { data: orders } = await query.graph({
+        entity: "order",
+        filters: {
+          ...input.filter,
+          status: {
+            $ne: "draft",
+          },
+        },
+        pagination: {
+          skip: page * pageSize,
+          take: pageSize,
+        },
+        fields,
+      })
+
+      if (orders.length === 0) {
+        break
+      }
+
+      for (let i = 0; i < orders.length; i++) {
+        const order = orders[i]
+        const order_ = order as any
+
+        order_.payment_status = getLastPaymentStatus(order_)
+        order_.fulfillment_status = getLastFulfillmentStatus(order_)
+
+        delete order_.version
+        delete order.payment_collections
+        delete order.fulfillments
+
+        orders[i] = normalizeOrderForExport(order)
+      }
+
+      const batchCsv = json2csv(orders, {
+        prependHeader: !hasHeader,
+        arrayIndexesAsKeys: true,
+        expandNestedObjects: true,
+        expandArrayObjects: true,
+        unwindArrays: false,
+        preventCsvInjection: true,
+        emptyFieldValue: "",
+      })
+
+      const ok = writeStream.write((hasHeader ? "\n" : "") + batchCsv)
+      if (!ok) {
+        await new Promise((resolve) => writeStream.once("drain", resolve))
+      }
+
+      hasHeader = true
+
+      if (orders.length < pageSize) {
+        break
+      }
+
+      page += 1
+    }
+
+    writeStream.end()
+
+    await promise
+
+    return new StepResponse(
+      { id: fileKey, filename } as ExportOrdersStepOutput,
+      fileKey
+    )
+  },
+  async (fileId, { container }) => {
+    if (!fileId) {
+      return
+    }
+
+    const fileModule: IFileModuleService = container.resolve(Modules.FILE)
+    await fileModule.deleteFiles(fileId)
+  }
+)

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

@@ -17,10 +17,11 @@ export * from "./delete-order-change-actions"
 export * from "./delete-order-changes"
 export * from "./delete-order-shipping-methods"
 export * from "./exchange/cancel-exchange"
-export * from "./list-order-change-actions-by-type"
 export * from "./exchange/create-exchange"
 export * from "./exchange/create-exchange-items-from-actions"
 export * from "./exchange/delete-exchanges"
+export * from "./export-orders"
+export * from "./list-order-change-actions-by-type"
 export * from "./preview-order-change"
 export * from "./register-delivery"
 export * from "./register-fulfillment"

packages/core/core-flows/src/order/workflows/export-orders.ts

@@ -0,0 +1,142 @@
+import { FilterableOrderProps } from "@medusajs/framework/types"
+import {
+  WorkflowData,
+  createWorkflow,
+  transform,
+} from "@medusajs/framework/workflows-sdk"
+import { useRemoteQueryStep } from "../../common"
+import { notifyOnFailureStep, sendNotificationsStep } from "../../notification"
+import { exportOrdersStep } from "../steps"
+
+/**
+ * The data to export orders.
+ */
+export type ExportOrdersDTO = {
+  /**
+   * The fields to select. These fields will be passed to
+   * [Query](https://docs.medusajs.com/learn/fundamentals/module-links/query), so you can
+   * pass order properties or any relation names, including custom links.
+   */
+  select: string[]
+  /**
+   * The filters to select which orders to export.
+   */
+  filter?: FilterableOrderProps
+}
+
+export const exportOrdersWorkflowId = "export-orders"
+/**
+ * This workflow exports orders matching the specified filters. It's used to
+ * export orders to a CSV file.
+ *
+ * :::note
+ *
+ * This workflow doesn't return the exported orders. Instead, it sends a notification to the admin
+ * users that they can download the exported orders.
+ *
+ * :::
+ *
+ * @example
+ * To export all orders:
+ *
+ * ```ts
+ * const { result } = await exportOrdersWorkflow(container)
+ * .run({
+ *   input: {
+ *     select: ["*"],
+ *   }
+ * })
+ * ```
+ *
+ * To export orders matching a criteria:
+ *
+ * ```ts
+ * const { result } = await exportOrdersWorkflow(container)
+ * .run({
+ *   input: {
+ *     select: ["*"],
+ *     filter: {
+ *       created_at: {
+ *         $gte: "2024-01-01",
+ *         $lte: "2024-12-31"
+ *       }
+ *     }
+ *   }
+ * })
+ * ```
+ *
+ * To export orders within a date range:
+ *
+ * ```ts
+ * const { result } = await exportOrdersWorkflow(container)
+ * .run({
+ *   input: {
+ *     select: ["*"],
+ *     filter: {
+ *       created_at: {
+ *         $gte: "2024-01-01T00:00:00Z",
+ *         $lte: "2024-01-31T23:59:59Z"
+ *       }
+ *     }
+ *   }
+ * })
+ * ```
+ *
+ * @summary
+ *
+ * Export orders with filtering capabilities.
+ */
+export const exportOrdersWorkflow = createWorkflow(
+  exportOrdersWorkflowId,
+  (input: WorkflowData<ExportOrdersDTO>): WorkflowData<void> => {
+    const file = exportOrdersStep(input).config({
+      async: true,
+      backgroundExecution: true,
+    })
+
+    const failureNotification = transform({ input }, (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: "Order export",
+            description: `Failed to export orders, please try again later.`,
+          },
+        },
+      ]
+    })
+    notifyOnFailureStep(failureNotification)
+
+    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: "Order export",
+            description: "Order export completed successfully!",
+            file: {
+              filename: data.file.filename,
+              url: data.fileDetails.url,
+              mimeType: "text/csv",
+            },
+          },
+        },
+      ]
+    })
+
+    sendNotificationsStep(notifications)
+  }
+)

packages/core/core-flows/src/order/workflows/index.ts

@@ -18,6 +18,7 @@ export * from "./claim/update-claim-add-item"
 export * from "./claim/update-claim-item"
 export * from "./claim/update-claim-shipping-method"
 export * from "./complete-orders"
+export * from "./compute-adjustments-for-preview"
 export * from "./create-fulfillment"
 export * from "./create-or-update-order-payment-collection"
 export * from "./create-order"
@@ -41,6 +42,7 @@ export * from "./exchange/remove-exchange-item-action"
 export * from "./exchange/remove-exchange-shipping-method"
 export * from "./exchange/update-exchange-add-item"
 export * from "./exchange/update-exchange-shipping-method"
+export * from "./export-orders"
 export * from "./fetch-shipping-option"
 export * from "./get-order-detail"
 export * from "./get-orders-list"
@@ -48,9 +50,9 @@ export * from "./list-shipping-options-for-order"
 export * from "./mark-order-fulfillment-as-delivered"
 export * from "./mark-payment-collection-as-paid"
 export * from "./maybe-refresh-shipping-methods"
+export * from "./on-carry-promotions-flag-set"
 export * from "./order-edit/begin-order-edit"
 export * from "./order-edit/cancel-begin-order-edit"
-export * from "./compute-adjustments-for-preview"
 export * from "./order-edit/confirm-order-edit-request"
 export * from "./order-edit/create-order-edit-shipping-method"
 export * from "./order-edit/order-edit-add-new-item"
@@ -81,7 +83,6 @@ export * from "./return/update-receive-item-return-request"
 export * from "./return/update-request-item-return"
 export * from "./return/update-return"
 export * from "./return/update-return-shipping-method"
-export * from "./on-carry-promotions-flag-set"
 export * from "./transfer/accept-order-transfer"
 export * from "./transfer/cancel-order-transfer"
 export * from "./transfer/decline-order-transfer"