feat(medusa): Add batch strategy for order exports (#1603)

integration-tests/api/__tests__/batch-jobs/api.js

@@ -63,7 +63,7 @@ describe("/admin/batch-jobs", () => {
   beforeAll(async () => {
     const cwd = path.resolve(path.join(__dirname, "..", ".."))
     dbConnection = await initDb({ cwd })
-    medusaProcess = await setupServer({ cwd, verbose: false })
+    medusaProcess = await setupServer({ cwd })
   })
 
   afterAll(async () => {
@@ -131,19 +131,19 @@ describe("/admin/batch-jobs", () => {
             id: "job_3",
             created_at: expect.any(String),
             updated_at: expect.any(String),
-            created_by: "admin_user"
+            created_by: "admin_user",
           },
           {
             id: "job_2",
             created_at: expect.any(String),
             updated_at: expect.any(String),
-            created_by: "admin_user"
+            created_by: "admin_user",
           },
           {
             id: "job_1",
             created_at: expect.any(String),
             updated_at: expect.any(String),
-            created_by: "admin_user"
+            created_by: "admin_user",
           },
         ],
       })
@@ -165,23 +165,24 @@ describe("/admin/batch-jobs", () => {
       const response = await api.get("/admin/batch-jobs/job_1", adminReqConfig)
 
       expect(response.status).toEqual(200)
-      expect(response.data.batch_job).toEqual(expect.objectContaining({
-        created_at: expect.any(String),
-        updated_at: expect.any(String),
-        created_by: "admin_user"
-      }))
+      expect(response.data.batch_job).toEqual(
+        expect.objectContaining({
+          created_at: expect.any(String),
+          updated_at: expect.any(String),
+          created_by: "admin_user",
+        })
+      )
     })
 
     it("should fail on batch job created by other user", async () => {
       const api = useApi()
-      await api.get("/admin/batch-jobs/job_4", adminReqConfig)
-        .catch((err) => {
-          expect(err.response.status).toEqual(400)
-          expect(err.response.data.type).toEqual("not_allowed")
-          expect(err.response.data.message).toEqual(
-            "Cannot access a batch job that does not belong to the logged in user"
-          )
-        })
+      await api.get("/admin/batch-jobs/job_4", adminReqConfig).catch((err) => {
+        expect(err.response.status).toEqual(400)
+        expect(err.response.data.type).toEqual("not_allowed")
+        expect(err.response.data.message).toEqual(
+          "Cannot access a batch job that does not belong to the logged in user"
+        )
+      })
     })
   })
 
@@ -195,7 +196,7 @@ describe("/admin/batch-jobs", () => {
       await db.teardown()
     })
 
-    it("Creates a batch job", async() => {
+    it("Creates a batch job", async () => {
       const api = useApi()
 
       const response = await api.post(
@@ -261,7 +262,7 @@ describe("/admin/batch-jobs", () => {
       }
     })
 
-    afterEach(async() => {
+    afterEach(async () => {
       const db = useDb()
       await db.teardown()
     })

integration-tests/api/__tests__/batch-jobs/order/export.js

@@ -0,0 +1,251 @@
+const path = require("path")
+const fs = require("fs/promises")
+import { sep, resolve } from "path"
+
+const setupServer = require("../../../../helpers/setup-server")
+const { useApi } = require("../../../../helpers/use-api")
+const { initDb, useDb } = require("../../../../helpers/use-db")
+
+const adminSeeder = require("../../../helpers/admin-seeder")
+const userSeeder = require("../../../helpers/user-seeder")
+const orderSeeder = require("../../../helpers/order-seeder")
+
+const adminReqConfig = {
+  headers: {
+    Authorization: "Bearer test_token",
+  },
+}
+
+jest.setTimeout(1000000)
+
+describe("Batchjob with type order-export", () => {
+  let medusaProcess
+  let dbConnection
+  let exportFilePath = ""
+  let topDir = ""
+
+  beforeAll(async () => {
+    const cwd = path.resolve(path.join(__dirname, "..", "..", ".."))
+    dbConnection = await initDb({ cwd })
+    medusaProcess = await setupServer({
+      cwd,
+      redisUrl: "redis://127.0.0.1:6379",
+      uploadDir: __dirname,
+      verbose: false,
+    })
+  })
+
+  afterAll(async () => {
+    if (topDir !== "") {
+      await fs.rm(resolve(__dirname, topDir), { recursive: true })
+    }
+
+    const db = useDb()
+    await db.shutdown()
+
+    medusaProcess.kill()
+  })
+
+  beforeEach(async () => {
+    try {
+      await adminSeeder(dbConnection)
+      await userSeeder(dbConnection)
+      await orderSeeder(dbConnection)
+    } catch (e) {
+      console.log(e)
+      throw e
+    }
+  })
+
+  afterEach(async () => {
+    const db = useDb()
+    await db.teardown()
+
+    const isFileExists = (await fs.stat(exportFilePath))?.isFile()
+
+    if (isFileExists) {
+      const [, relativeRoot] = exportFilePath.replace(__dirname, "").split(sep)
+
+      if ((await fs.stat(resolve(__dirname, relativeRoot)))?.isDirectory()) {
+        topDir = relativeRoot
+      }
+
+      await fs.unlink(exportFilePath)
+    }
+  })
+
+  it("Should export a file containing all orders", async () => {
+    jest.setTimeout(1000000)
+    const api = useApi()
+
+    const batchPayload = {
+      type: "order-export",
+      context: {},
+    }
+
+    const batchJobRes = await api.post(
+      "/admin/batch-jobs",
+      batchPayload,
+      adminReqConfig
+    )
+    const batchJobId = batchJobRes.data.batch_job.id
+
+    expect(batchJobId).toBeTruthy()
+
+    // Pull to check the status until it is completed
+    let batchJob
+    let shouldContinuePulling = true
+
+    while (shouldContinuePulling) {
+      const res = await api.get(
+        `/admin/batch-jobs/${batchJobId}`,
+        adminReqConfig
+      )
+
+      batchJob = res.data.batch_job
+      shouldContinuePulling = !(
+        batchJob.status === "completed" || batchJob.status === "failed"
+      )
+
+      if (shouldContinuePulling) {
+        await new Promise((resolve, _) => {
+          setTimeout(resolve, 1000)
+        })
+      }
+    }
+
+    expect(batchJob.status).toBe("completed")
+
+    expect(batchJob.status).toBe("completed")
+
+    exportFilePath = path.resolve(__dirname, batchJob.result.file_key)
+    const isFileExists = (await fs.stat(exportFilePath)).isFile()
+
+    expect(isFileExists).toBeTruthy()
+
+    const fileSize = (await fs.stat(exportFilePath)).size
+    expect(batchJob.result?.file_size).toBe(fileSize)
+
+    const data = (await fs.readFile(exportFilePath)).toString()
+    const [, ...lines] = data.split("\r\n").filter((l) => l)
+
+    expect(lines.length).toBe(6)
+
+    const csvLine = lines[0].split(";")
+
+    expect(csvLine[0]).toBe("discount-order")
+    expect(csvLine[1]).toBe("6")
+    expect(csvLine[14]).toBe("fulfilled")
+    expect(csvLine[15]).toBe("captured")
+    expect(csvLine[16]).toBe("8000")
+  })
+
+  it("Should export a file containing a limited number of orders", async () => {
+    jest.setTimeout(1000000)
+    const api = useApi()
+
+    const batchPayload = {
+      type: "order-export",
+      context: { batch_size: 3 },
+    }
+
+    const batchJobRes = await api.post(
+      "/admin/batch-jobs",
+      batchPayload,
+      adminReqConfig
+    )
+    const batchJobId = batchJobRes.data.batch_job.id
+
+    expect(batchJobId).toBeTruthy()
+
+    // Pull to check the status until it is completed
+    let batchJob
+    let shouldContinuePulling = true
+
+    while (shouldContinuePulling) {
+      const res = await api.get(
+        `/admin/batch-jobs/${batchJobId}`,
+        adminReqConfig
+      )
+
+      batchJob = res.data.batch_job
+      shouldContinuePulling = !(
+        batchJob.status === "completed" || batchJob.status === "failed"
+      )
+
+      if (shouldContinuePulling) {
+        await new Promise((resolve, _) => {
+          setTimeout(resolve, 1000)
+        })
+      }
+    }
+
+    exportFilePath = path.resolve(__dirname, batchJob.result.file_key)
+    const isFileExists = (await fs.stat(exportFilePath)).isFile()
+
+    expect(isFileExists).toBeTruthy()
+
+    const data = (await fs.readFile(exportFilePath)).toString()
+    const [, ...lines] = data.split("\r\n").filter((l) => l)
+
+    expect(lines.length).toBe(3)
+  })
+
+  it("Should export a file with orders from a single customer", async () => {
+    jest.setTimeout(1000000)
+    const api = useApi()
+
+    const batchPayload = {
+      type: "order-export",
+      context: { filterable_fields: { email: "test@email.com" } },
+    }
+
+    const batchJobRes = await api.post(
+      "/admin/batch-jobs",
+      batchPayload,
+      adminReqConfig
+    )
+    const batchJobId = batchJobRes.data.batch_job.id
+
+    expect(batchJobId).toBeTruthy()
+
+    // Pull to check the status until it is completed
+    let batchJob
+    let shouldContinuePulling = true
+
+    while (shouldContinuePulling) {
+      const res = await api.get(
+        `/admin/batch-jobs/${batchJobId}`,
+        adminReqConfig
+      )
+
+      batchJob = res.data.batch_job
+      shouldContinuePulling = !(
+        batchJob.status === "completed" || batchJob.status === "failed"
+      )
+
+      if (shouldContinuePulling) {
+        await new Promise((resolve, _) => {
+          setTimeout(resolve, 1000)
+        })
+      }
+    }
+
+    expect(batchJob.status).toBe("completed")
+
+    exportFilePath = path.resolve(__dirname, batchJob.result.file_key)
+    const isFileExists = (await fs.stat(exportFilePath)).isFile()
+
+    expect(isFileExists).toBeTruthy()
+
+    const data = (await fs.readFile(exportFilePath)).toString()
+    const [, ...lines] = data.split("\r\n").filter((l) => l)
+
+    expect(lines.length).toBe(1)
+
+    const csvLine = lines[0].split(";")
+
+    expect(csvLine[0]).toBe("test-order")
+    expect(csvLine[6]).toBe("test@email.com")
+  })
+})

integration-tests/api/__tests__/batch-jobs/product/export.js

@@ -1,5 +1,6 @@
 const path = require("path")
-const fs = require('fs/promises')
+const fs = require("fs/promises")
+import { sep, resolve } from "path"
 
 const setupServer = require("../../../../helpers/setup-server")
 const { useApi } = require("../../../../helpers/use-api")
@@ -20,6 +21,8 @@ jest.setTimeout(1000000)
 describe("Batch job of product-export type", () => {
   let medusaProcess
   let dbConnection
+  let exportFilePath = ""
+  let topDir = ""
 
   beforeAll(async () => {
     const cwd = path.resolve(path.join(__dirname, "..", "..", ".."))
@@ -28,13 +31,15 @@ describe("Batch job of product-export type", () => {
       cwd,
       redisUrl: "redis://127.0.0.1:6379",
       uploadDir: __dirname,
-      verbose: false
+      verbose: false,
     })
   })
 
-  let exportFilePath = ""
-
   afterAll(async () => {
+    if (topDir !== "") {
+      await fs.rm(resolve(__dirname, topDir), { recursive: true })
+    }
+
     const db = useDb()
     await db.shutdown()
 
@@ -52,17 +57,24 @@ describe("Batch job of product-export type", () => {
     }
   })
 
-  afterEach(async() => {
+  afterEach(async () => {
     const db = useDb()
     await db.teardown()
 
     const isFileExists = (await fs.stat(exportFilePath))?.isFile()
+
     if (isFileExists) {
+      const [, relativeRoot] = exportFilePath.replace(__dirname, "").split(sep)
+
+      if ((await fs.stat(resolve(__dirname, relativeRoot)))?.isDirectory()) {
+        topDir = relativeRoot
+      }
+
       await fs.unlink(exportFilePath)
     }
   })
 
-  it('should export a csv file containing the expected products', async () => {
+  it("should export a csv file containing the expected products", async () => {
     jest.setTimeout(1000000)
     const api = useApi()
 
@@ -97,37 +109,52 @@ describe("Batch job of product-export type", () => {
         },
       ],
     }
-    const createProductRes =
-      await api.post("/admin/products", productPayload, adminReqConfig)
+    const createProductRes = await api.post(
+      "/admin/products",
+      productPayload,
+      adminReqConfig
+    )
     const productId = createProductRes.data.product.id
     const variantId = createProductRes.data.product.variants[0].id
 
     const batchPayload = {
       type: "product-export",
       context: {
-        filterable_fields: { title: "Test export product" }
+        filterable_fields: {
+          title: "Test export product",
+        },
       },
     }
-    const batchJobRes = await api.post("/admin/batch-jobs", batchPayload, adminReqConfig)
+    const batchJobRes = await api.post(
+      "/admin/batch-jobs",
+      batchPayload,
+      adminReqConfig
+    )
     const batchJobId = batchJobRes.data.batch_job.id
 
     expect(batchJobId).toBeTruthy()
 
     // Pull to check the status until it is completed
-    let batchJob;
+    let batchJob
     let shouldContinuePulling = true
     while (shouldContinuePulling) {
-      const res = await api
-      .get(`/admin/batch-jobs/${batchJobId}`, adminReqConfig)
+      const res = await api.get(
+        `/admin/batch-jobs/${batchJobId}`,
+        adminReqConfig
+      )
 
       await new Promise((resolve, _) => {
         setTimeout(resolve, 1000)
       })
 
       batchJob = res.data.batch_job
-      shouldContinuePulling = !(batchJob.status === "completed")
+      shouldContinuePulling = !(
+        batchJob.status === "completed" || batchJob.status === "failed"
+      )
     }
 
+    expect(batchJob.status).toBe("completed")
+
     exportFilePath = path.resolve(__dirname, batchJob.result.file_key)
     const isFileExists = (await fs.stat(exportFilePath)).isFile()
 
@@ -137,7 +164,7 @@ describe("Batch job of product-export type", () => {
     expect(batchJob.result?.file_size).toBe(fileSize)
 
     const data = (await fs.readFile(exportFilePath)).toString()
-    const [, ...lines] = data.split("\r\n").filter(l => l)
+    const [, ...lines] = data.split("\r\n").filter((l) => l)
 
     expect(lines.length).toBe(1)
 
@@ -150,4 +177,61 @@ describe("Batch job of product-export type", () => {
     expect(lineColumn[24]).toBe(productPayload.variants[0].title)
     expect(lineColumn[25]).toBe(productPayload.variants[0].sku)
   })
-})
+
+  it("should export a csv file containing a limited number of products", async () => {
+    jest.setTimeout(1000000)
+    const api = useApi()
+
+    const batchPayload = {
+      type: "product-export",
+      context: {
+        batch_size: 1,
+        filterable_fields: { collection_id: "test-collection" },
+        order: "created_at",
+      },
+    }
+
+    const batchJobRes = await api.post(
+      "/admin/batch-jobs",
+      batchPayload,
+      adminReqConfig
+    )
+    const batchJobId = batchJobRes.data.batch_job.id
+
+    expect(batchJobId).toBeTruthy()
+
+    // Pull to check the status until it is completed
+    let batchJob
+    let shouldContinuePulling = true
+    while (shouldContinuePulling) {
+      const res = await api.get(
+        `/admin/batch-jobs/${batchJobId}`,
+        adminReqConfig
+      )
+
+      await new Promise((resolve, _) => {
+        setTimeout(resolve, 1000)
+      })
+
+      batchJob = res.data.batch_job
+      shouldContinuePulling = !(
+        batchJob.status === "completed" || batchJob.status === "failed"
+      )
+    }
+
+    expect(batchJob.status).toBe("completed")
+
+    exportFilePath = path.resolve(__dirname, batchJob.result.file_key)
+    const isFileExists = (await fs.stat(exportFilePath)).isFile()
+
+    expect(isFileExists).toBeTruthy()
+
+    const data = (await fs.readFile(exportFilePath)).toString()
+    const [, ...lines] = data.split("\r\n").filter((l) => l)
+
+    expect(lines.length).toBe(4)
+
+    const csvLine = lines[0].split(";")
+    expect(csvLine[0]).toBe("test-product")
+  })
+})

integration-tests/api/medusa-config.js

@@ -8,7 +8,7 @@ module.exports = {
     redis_url: process.env.REDIS_URL,
     database_url: `postgres://${DB_USERNAME}:${DB_PASSWORD}@localhost/medusa-integration-${workerId}`,
     database_type: "postgres",
-    jwt_secret: 'test',
-    cookie_secret: 'test'
+    jwt_secret: "test",
+    cookie_secret: "test",
   },
 }

integration-tests/api/src/services/local-file-service.js

@@ -1,55 +1,63 @@
 import { AbstractFileService } from "@medusajs/medusa"
 import stream from "stream"
-import { resolve } from "path"
 import * as fs from "fs"
+import * as path from "path"
 
 export default class LocalFileService extends AbstractFileService {
+  // eslint-disable-next-line no-empty-pattern
   constructor({}, options) {
-    super({});
-    this.upload_dir_ = process.env.UPLOAD_DIR ?? options.upload_dir ?? "uploads/images";
+    super({})
+    this.upload_dir_ =
+      process.env.UPLOAD_DIR ?? options.upload_dir ?? "uploads/images"
 
     if (!fs.existsSync(this.upload_dir_)) {
-      fs.mkdirSync(this.upload_dir_);
+      fs.mkdirSync(this.upload_dir_)
     }
   }
 
-  upload(file) {
-    return new Promise((resolve, reject) => {
-      const path = resolve(this.upload_dir_, file.originalname)
-      fs.writeFile(path, "", err => {
-        if (err) {
-          reject(err);
-        }
+  async upload(file) {
+    const uploadPath = path.join(
+      this.upload_dir_,
+      path.dirname(file.originalname)
+    )
 
-        resolve({ url: path });
-      });
-    });
+    if (!fs.existsSync(uploadPath)) {
+      fs.mkdirSync(uploadPath, { recursive: true })
+    }
+
+    const filePath = path.resolve(this.upload_dir_, file.originalname)
+    fs.writeFile(filePath, "", (error) => {
+      if (error) {
+        throw error
+      }
+    })
+    return { url: filePath }
   }
 
-  delete({ name }) {
+  async delete({ name }) {
     return new Promise((resolve, _) => {
       const path = resolve(this.upload_dir_, name)
-      fs.unlink(path, err => {
+      fs.unlink(path, (err) => {
         if (err) {
-          throw err;
+          throw err
         }
 
-        resolve("file unlinked");
-      });
-    });
+        resolve("file unlinked")
+      })
+    })
   }
 
   async getUploadStreamDescriptor({ name, ext }) {
-    const fileKey = `${name}.${ext}`
-    const path = resolve(this.upload_dir_, fileKey)
+    const fileKey = `${name}-${Date.now()}.${ext}`
+    const filePath = path.resolve(this.upload_dir_, fileKey)
 
-    const isFileExists = fs.existsSync(path)
+    const isFileExists = fs.existsSync(filePath)
     if (!isFileExists) {
       await this.upload({ originalname: fileKey })
     }
 
     const pass = new stream.PassThrough()
-    pass.pipe(fs.createWriteStream(path))
+    pass.pipe(fs.createWriteStream(filePath))
 
     return {
       writeStream: pass,