4.4.3. Schedule Syncing Brands from Third-Party System

1. Implement Syncing Workflow#

Start by defining the workflow that syncs the brand from the third-party system.

The workflow has the following steps:

1. Retrieve brands from the third-party system.
2. Create new brands in Medusa.
3. Update existing brands in Medusa.

Retrieve Brands Step#

To create the step that retrieves the brands from the third-party service, create the file src/workflows/sync-brands-from-system/steps/retrieve-brands-from-system.ts with the following content:

src/workflows/sync-brands-from-system/steps/retrieve-brands-from-system.ts
Show Imports
6import { BRAND_MODULE } from "../../../modules/brand"7
8export const retrieveBrandsFromSystemStep = createStep(9  "retrieve-brands-from-system",10  async (_, { container }) => {11    const brandModuleService: BrandModuleService = container.resolve(12      BRAND_MODULE13    )14
15    const brands = await brandModuleService.client.retrieveBrands()16
17    return new StepResponse(brands)18  }19)

In this step, you resolve the Brand Module's main service from the container, and use its client service to retrieve the brands from the third-party system.

The step returns the retrieved brands.

Create Brands Step#

Next, create the step that creates new brands in Medusa in the file src/workflows/sync-brands-from-system/steps/create-brands.ts:

src/workflows/sync-brands-from-system/steps/create-brands.ts
Show Imports
8import { Brand } from "../../../modules/brand/models/brand"9
10type CreateBrandsInput = {11  brands: InferTypeOf<typeof Brand>[]12}13
14export const createBrandsStep = createStep(15  "create-brand-step",16  async (input: CreateBrandsInput, { container }) => {17    const brandModuleService: BrandModuleService = container.resolve(18      BRAND_MODULE19    )20
21    const brands = await brandModuleService.createBrands(input.brands)22
23    return new StepResponse(brands, brands.map((brand) => brand.id))24  },25  async (ids: string[], { container }) => {26    const brandModuleService: BrandModuleService = container.resolve(27      BRAND_MODULE28    )29
30    await brandModuleService.deleteBrands(ids)31  }32)

This step receives the brands to create as input.

In the step, you resolve the Brand Module's main service and uses its createBrands method to create the brands.

You return the created brands and pass their IDs to the compensation function, which deletes the brands if an error occurs.

Update Brands Step#

To create the step that updates existing brands in Medusa, create the file src/workflows/sync-brands-from-system/steps/update-brands.ts with the following content:

src/workflows/sync-brands-from-system/steps/update-brands.ts
Show Imports
8import { Brand } from "../../../modules/brand/models/brand"9
10type UpdateBrandsInput = {11  brands: InferTypeOf<typeof Brand>[]12}13
14export const updateBrandsStep = createStep(15  "update-brand-step",16  async ({ brands }: UpdateBrandsInput, { container }) => {17    const brandModuleService: BrandModuleService = container.resolve(18      BRAND_MODULE19    )20
21    const prevUpdatedBrands = await brandModuleService.listBrands({22      id: brands.map((brand) => brand.id),23    })24
25    const updatedBrands = await brandModuleService.updateBrands(brands)26
27    return new StepResponse(updatedBrands, prevUpdatedBrands)28  },29  async (prevUpdatedBrands, { container }) => {30    const brandModuleService: BrandModuleService = container.resolve(31      BRAND_MODULE32    )33
34    await brandModuleService.updateBrands(prevUpdatedBrands)35  }36)

This step receives the brands to update as input.

In the step, you retrieve the brands first to pass them later to the compensation function, then update and return the brands.

In the compensation function, you update the brands are again but to their data before the update made by the step.

Create Workflow#

Finally, create the workflow in the file src/workflows/sync-brands-from-system/index.ts with the following content:

src/workflows/sync-brands-from-system/index.ts
1import {2  createWorkflow,3  WorkflowResponse,4  transform,5} from "@medusajs/workflows-sdk"6import { InferTypeOf } from "@medusajs/types"7import { retrieveBrandsFromSystemStep } from "./steps/retrieve-brands-from-system"8import { createBrandsStep } from "./steps/create-brands"9import { updateBrandsStep } from "./steps/update-brands"10import { Brand } from "../../modules/brand/models/brand"11
12export const syncBrandsFromSystemWorkflow = createWorkflow(13  "sync-brands-from-system",14  () => {15    const brands = retrieveBrandsFromSystemStep()16
17    // TODO create and update brands18  }19)

For now, you only add the retrieveBrandsFromSystemStep to the workflow that retrieves the brands from the third-party system.

Identify Brands to Create or Update in Workflow#

Next, you need to identify which brands must be created or updated.

Since workflows are constructed internally and are only evaluated during execution, you can't access any data's value to perform data manipulation or checks.

Instead, use the transform utility function imported from @medusajs/workflows-sdk, which gives you access to the real-time values of the data to perfrom actions on them.

So, replace the TODO with the following:

src/workflows/sync-brands-from-system/index.ts
1const { toCreate, toUpdate } = transform(2  {3    brands,4  },5  (data) => {6    const toCreate: InferTypeOf<typeof Brand>[] = []7        const toUpdate: InferTypeOf<typeof Brand>[] = []8
9    data.brands.forEach((brand) => {10      if (brand.external_id) {11        toUpdate.push({12          ...brand,13          id: brand.external_id,14        })15      } else {16        toCreate.push(brand)17      }18    })19
20    return { toCreate, toUpdate }21  }22)23
24// TODO create and update the brands

transform accepts two parameters:

1. The data to be passed to the function in the second parameter.
2. A function to execute only when the workflow is executed. Its return value can be consumed by the rest of the workflow.

In the function, you sort the brands as to be created or to be updated based on whether they have an external_id property.

Create and Update the Brands#

Finally, replace the new TODO with the following:

src/workflows/sync-brands-from-system/index.ts
1const created = createBrandsStep({ brands: toCreate })2const updated = updateBrandsStep({ brands: toUpdate })3
4return new WorkflowResponse({5  created,6  updated,7})

You pass the brands to be created to the createBrandsStep, and the brands to be updated to the updateBrandsStep.

Then, you return the created and updated brands.

2. Schedule Syncing Task#

To schedule a task that syncs brands from the third-party system, create a scheduled job at src/jobs/sync-brands-from-system.ts:

src/jobs/sync-brands-from-system.ts
1import { MedusaContainer } from "@medusajs/types"2import { syncBrandsFromSystemWorkflow } from "../workflows/sync-brands-from-system"3
4export default async function (container: MedusaContainer) {5  const logger = container.resolve("logger")6
7  const { result } = await syncBrandsFromSystemWorkflow(container).run()8
9  logger.info(10    `Synced brands from third-party system: ${11      result.created.length12    } brands created and ${result.updated.length} brands updated.`)13}14
15export const config = {16  name: "sync-brands-from-system",17  schedule: "* * * * *",18}

This defines a scheduled job that runs every minute (for testing purposes).

The scheduled job executes the syncBrandsFromSystemWorkflow and prints how many brands were created and updated.

Test it Out#

To test it out, start the Medusa application. In a minute, the scheduled job will run and you'll see a logged message indicating how many brands were created or updated.

Summary#

In the previous chapters, you:

• Created a service that acts as a client integrating a third-party system.
• Implemented two-way sync of brands between the third-party system and Medusa using a subscriber and a scheduled job.