fix: source shopify handle duplicate unique values and more efficient updates (#1154)

* fix: added functionality to handle duplicate sku, ean, upc, and barcodes on product variants

* handle unique variant values

* save

* clear ignored keys in redis

* fixed plugin to handle duplicates and use last build time to make queries more effective

* v1.1.1-canary.0

* removed console log

* v1.1.1-canary.1

* reduce amounts of calls to shopify api

* v1.1.1-canary.2

* added README and support for SmartCollections

* v1.1.1-canary.4

* v1.1.1-canary.5

* v1.1.1-canary.5

* fix dropped products in smart collections

* cleanup

Co-authored-by: Sebastian Rindom <skrindom@gmail.com>

packages/medusa-source-shopify/src/services/shopify-client.js

@@ -1,7 +1,7 @@
-import { BaseService } from "medusa-interfaces"
-import { pager } from "../utils/pager"
-import { createClient } from "../utils/create-client"
 import { DataType } from "@shopify/shopify-api"
+import { BaseService } from "medusa-interfaces"
+import { createClient } from "../utils/create-client"
+import { pager } from "../utils/pager"
 
 class ShopifyClientService extends BaseService {
   // eslint-disable-next-line no-empty-pattern
@@ -23,7 +23,7 @@ class ShopifyClientService extends BaseService {
   }
 
   delete(params) {
-    return this.client_.post(params)
+    return this.client_.delete(params)
   }
 
   post(params) {

packages/medusa-source-shopify/src/services/shopify-collection.js

@@ -3,7 +3,14 @@ import { removeIndex } from "../utils/remove-index"
 
 class ShopifyCollectionService extends BaseService {
   constructor(
-    { manager, shopifyProductService, productCollectionService },
+    {
+      manager,
+      shopifyProductService,
+      productCollectionService,
+      productService,
+      storeService,
+      productRepository,
+    },
     options
   ) {
     super()
@@ -16,6 +23,13 @@ class ShopifyCollectionService extends BaseService {
     this.productService_ = shopifyProductService
     /** @private @const {ProductCollectionService} */
     this.collectionService_ = productCollectionService
+    /** @private @const {StoreService} */
+    this.storeService_ = storeService
+    /** @private @const {ProductService} */
+    this.medusaProductService_ = productService
+
+    /** @private @const {Product} */
+    this.productRepository_ = productRepository
   }
 
   withTransaction(transactionManager) {
@@ -28,6 +42,9 @@ class ShopifyCollectionService extends BaseService {
       options: this.options,
       shopifyProductService: this.productService_,
       productCollectionService: this.collectionService_,
+      storeService: this.storeService_,
+      productService: this.medusaProductService_,
+      productRepository: this.productRepository_,
     })
 
     cloned.transactionManager_ = transactionManager
@@ -42,10 +59,10 @@ class ShopifyCollectionService extends BaseService {
    * @param {Object[]} products
    * @return {Promise}
    */
-  async createWithProducts(collects, collections, products) {
+  async createCustomCollections(collects, collections, products) {
     return this.atomicPhase_(async (manager) => {
       const normalizedCollections = collections.map((c) =>
-        this.normalizeCollection_(c)
+        this.normalizeCustomCollection_(c)
       )
 
       const result = []
@@ -61,25 +78,71 @@ class ShopifyCollectionService extends BaseService {
             .create(nc)
         }
 
-        const productIds = collects.reduce((productIds, c) => {
-          if (c.collection_id === collection.metadata.sh_id) {
-            productIds.push(c.product_id)
-          }
-          return productIds
-        }, [])
+        const productIds = this.getCustomCollectionProducts_(
+          collection.metadata.sh_id,
+          collects,
+          products
+        )
 
-        const reducedProducts = products.reduce((reducedProducts, p) => {
-          if (productIds.includes(p.id)) {
-            reducedProducts.push(p)
-            removeIndex(products, p)
-          }
-          return reducedProducts
-        }, [])
+        await this.addProductsToCollection(collection.id, productIds)
 
-        for (const product of reducedProducts) {
-          await this.productService_
+        result.push(collection)
+      }
+
+      return result
+    })
+  }
+
+  async createSmartCollections(collections, products) {
+    return this.atomicPhase_(async (manager) => {
+      if (!collections) {
+        return Promise.resolve()
+      }
+
+      const productRepo = this.manager_.getCustomRepository(
+        this.productRepository_
+      )
+
+      const ids = products.map((p) => p.id)
+      const completeProducts = await productRepo.findWithRelations(
+        ["variants", "tags", "type"],
+        ids
+      )
+
+      const defaultCurrency = await this.storeService_
+        .retrieve()
+        .then((store) => {
+          return store.default_currency_code
+        })
+        .catch((_) => undefined)
+
+      const normalizedCollections = collections.map((c) =>
+        this.normalizeSmartCollection_(c)
+      )
+
+      const result = []
+
+      for (const nc of normalizedCollections) {
+        let collection = await this.collectionService_
+          .retrieveByHandle(nc.collection.handle)
+          .catch((_) => undefined)
+
+        if (!collection) {
+          collection = await this.collectionService_
             .withTransaction(manager)
-            .create(product, collection.id)
+            .create(nc.collection)
+        }
+
+        const validProducts = this.getValidProducts_(
+          nc.rules,
+          completeProducts,
+          nc.disjunctive ?? false,
+          defaultCurrency
+        )
+
+        if (validProducts.length) {
+          const productIds = validProducts.map((p) => p.id)
+          await this.addProductsToCollection(collection.id, productIds)
         }
 
         result.push(collection)
@@ -89,7 +152,222 @@ class ShopifyCollectionService extends BaseService {
     })
   }
 
-  normalizeCollection_(shopifyCollection) {
+  async addProductsToCollection(collectionId, productIds) {
+    return this.atomicPhase_(async (manager) => {
+      const result = await this.collectionService_
+        .withTransaction(manager)
+        .addProducts(collectionId, productIds)
+
+      return result
+    })
+  }
+
+  getCustomCollectionProducts_(shCollectionId, collects, products) {
+    const medusaProductIds = products.reduce((prev, curr) => {
+      if (curr.external_id) {
+        prev[curr.external_id] = curr.id
+      }
+
+      return prev
+    }, {})
+
+    const productIds = collects.reduce((productIds, c) => {
+      if (c.collection_id === shCollectionId) {
+        productIds.push(`${c.product_id}`)
+      }
+      return productIds
+    }, [])
+
+    const productIdsToAdd = Object.keys(medusaProductIds).map((shopifyId) => {
+      if (productIds.includes(shopifyId)) {
+        const medusaId = medusaProductIds[shopifyId]
+        delete medusaProductIds[shopifyId]
+        return medusaId
+      }
+    })
+
+    // remove added products from the array
+    for (const id of productIdsToAdd) {
+      const productToRemove = products.find((p) => p.id === id)
+      if (productToRemove) {
+        removeIndex(products, productToRemove)
+      }
+    }
+
+    return productIdsToAdd
+  }
+
+  getValidProducts_(rules, products, disjunctive, defaultCurrency) {
+    const validProducts = []
+
+    for (const product of products) {
+      const results = rules.map((r) =>
+        this.testRule_(r, product, defaultCurrency)
+      )
+
+      if (disjunctive && !results.includes(false)) {
+        validProducts.push(product)
+      } else if (!disjunctive && results.includes(true)) {
+        validProducts.push(product)
+      }
+    }
+
+    // remove valid products from the array
+    for (const validProduct of validProducts) {
+      removeIndex(products, validProduct)
+    }
+
+    return validProducts
+  }
+
+  testRule_(rule, product, defaultCurrency = undefined) {
+    const { column, relation, condition } = rule
+
+    if (column === "title") {
+      return this.testTextRelation_(product.title, relation, condition)
+    }
+
+    if (column === "type") {
+      return this.testTextRelation_(product.type.value, relation, condition)
+    }
+
+    if (column === "vendor") {
+      if (product.metadata?.vendor) {
+        return this.testTextRelation_(
+          product.metadata?.vendor,
+          relation,
+          condition
+        )
+      }
+
+      return false
+    }
+
+    if (column === "variant_title") {
+      if (product.variants?.length) {
+        const anyMatch = product.variants.some((variant) => {
+          return this.testTextRelation_(variant.title, relation, condition)
+        })
+
+        return anyMatch
+      }
+
+      return false
+    }
+
+    if (column === "tag") {
+      if (product.tags) {
+        const anyMatch = product.tags.some((tag) =>
+          this.testTextRelation_(tag.value, relation, condition)
+        )
+
+        return anyMatch
+      }
+
+      return false
+    }
+
+    if (column === "variant_inventory") {
+      if (product.variants?.length) {
+        const anyMatch = product.variants.some((variant) =>
+          this.testNumberRelation_(
+            variant.inventory_quantity,
+            relation,
+            condition
+          )
+        )
+
+        return anyMatch
+      }
+
+      return false
+    }
+
+    if (column === "variant_price") {
+      if (product.variants?.length && defaultCurrency) {
+        const prices = []
+
+        for (const variant of product.variants) {
+          if (variant.prices) {
+            for (const price of variant.prices) {
+              if (price.currency_code === defaultCurrency) {
+                prices.push(price.amount)
+              }
+            }
+          }
+        }
+
+        const anyMatch = prices.some((price) => {
+          return this.testNumberRelation_(price, relation, condition)
+        })
+
+        return anyMatch
+      }
+
+      return false
+    }
+
+    if (column === "variant_weight") {
+      if (product.variants?.length) {
+        const anyMatch = product.variants.some((variant) =>
+          this.testNumberRelation_(variant.weight, relation, condition)
+        )
+
+        return anyMatch
+      }
+
+      return false
+    }
+
+    // If we get here, it means the column is variant_compare_at_price which we don't support until we extend MoneyAmount
+    return true
+  }
+
+  testTextRelation_(text, relation, condition) {
+    if (relation === "contains") {
+      return text.includes(condition)
+    }
+
+    if (relation === "equals") {
+      return text === condition
+    }
+
+    if (relation === "not_equals") {
+      return text !== condition
+    }
+
+    if (relation === "starts_with") {
+      return text.startsWith(condition)
+    }
+
+    if (relation === "ends_with") {
+      return text.endsWith(condition)
+    }
+
+    return false
+  }
+
+  testNumberRelation_(number, relation, condition) {
+    if (relation === "greater_than") {
+      return number > condition
+    }
+
+    if (relation === "less_than") {
+      return number < condition
+    }
+
+    if (relation === "equals") {
+      return number === condition
+    }
+
+    if (relation === "not_equals") {
+      return number !== condition
+    }
+
+    return false
+  }
+
+  normalizeCustomCollection_(shopifyCollection) {
     return {
       title: shopifyCollection.title,
       handle: shopifyCollection.handle,
@@ -99,6 +377,21 @@ class ShopifyCollectionService extends BaseService {
       },
     }
   }
+
+  normalizeSmartCollection_(smartCollection) {
+    return {
+      collection: {
+        title: smartCollection.title,
+        handle: smartCollection.handle,
+        metadata: {
+          sh_id: smartCollection.id,
+          sh_body: smartCollection.body_html,
+        },
+      },
+      rules: smartCollection.rules,
+      disjunctive: smartCollection.disjunctive,
+    }
+  }
 }
 
 export default ShopifyCollectionService

packages/medusa-source-shopify/src/services/shopify-product.js

@@ -1,9 +1,10 @@
-import { BaseService } from "medusa-interfaces"
-import { MedusaError } from "medusa-core-utils"
-import _ from "lodash"
-import { parsePrice } from "../utils/parse-price"
-import { INCLUDE_PRESENTMENT_PRICES } from "../utils/const"
 import axios from "axios"
+import isEmpty from "lodash/isEmpty"
+import omit from "lodash/omit"
+import random from "lodash/random"
+import { MedusaError } from "medusa-core-utils"
+import { BaseService } from "medusa-interfaces"
+import { parsePrice } from "../utils/parse-price"
 
 class ShopifyProductService extends BaseService {
   constructor(
@@ -62,7 +63,7 @@ class ShopifyProductService extends BaseService {
    * @param {string} collectionId optional
    * @return {Product} the created product
    */
-  async create(data, collectionId) {
+  async create(data) {
     return this.atomicPhase_(async (manager) => {
       const ignore = await this.redis_.shouldIgnore(data.id, "product.created")
       if (ignore) {
@@ -71,14 +72,16 @@ class ShopifyProductService extends BaseService {
 
       const existingProduct = await this.productService_
         .withTransaction(manager)
-        .retrieveByExternalId(data.id)
+        .retrieveByExternalId(data.id, {
+          relations: ["variants", "options"],
+        })
         .catch((_) => undefined)
 
       if (existingProduct) {
-        return await this.update(data)
+        return await this.update(existingProduct, data)
       }
 
-      const normalizedProduct = this.normalizeProduct_(data, collectionId)
+      const normalizedProduct = this.normalizeProduct_(data)
       normalizedProduct.profile_id = await this.getShippingProfile_(
         normalizedProduct.is_giftcard
       )
@@ -95,7 +98,8 @@ class ShopifyProductService extends BaseService {
           this.addVariantOptions_(v, product.options)
         )
 
-        for (const variant of variants) {
+        for (let variant of variants) {
+          variant = await this.ensureVariantUnique_(variant)
           await this.productVariantService_
             .withTransaction(manager)
             .create(product.id, variant)
@@ -108,45 +112,24 @@ class ShopifyProductService extends BaseService {
     })
   }
 
-  async update(data) {
+  async update(existing, shopifyUpdate) {
     return this.atomicPhase_(async (manager) => {
-      const ignore = await this.redis_.shouldIgnore(data.id, "product.updated")
+      const ignore = await this.redis_.shouldIgnore(
+        shopifyUpdate.id,
+        "product.updated"
+      )
       if (ignore) {
         return
       }
 
-      let existing = await this.productService_
-        .retrieveByExternalId(data.id, {
-          relations: ["variants", "options"],
-        })
-        .catch((_) => undefined)
-
-      if (!existing) {
-        return await this.create(data)
-      }
-
-      /**
-       * Variants received from webhook do not include
-       * presentment prices. Therefore, we fetch them
-       * separately, and add to the data object.
-       */
-      const { variants } = await this.shopify_
-        .get({
-          path: `products/${data.id}`,
-          extraHeaders: INCLUDE_PRESENTMENT_PRICES,
-        })
-        .then((res) => {
-          return res.body.product
-        })
-
-      data.variants = variants || []
-      const normalized = this.normalizeProduct_(data)
+      const normalized = this.normalizeProduct_(shopifyUpdate)
 
       existing = await this.addProductOptions_(existing, normalized.options)
 
       await this.updateVariants_(existing, normalized.variants)
       await this.deleteVariants_(existing, normalized.variants)
       delete normalized.variants
+      delete normalized.options
 
       const update = {}
 
@@ -156,8 +139,8 @@ class ShopifyProductService extends BaseService {
         }
       }
 
-      if (!_.isEmpty(update)) {
-        await this.redis_.addIgnore(data.id, "product.updated")
+      if (!isEmpty(update)) {
+        await this.redis_.addIgnore(shopifyUpdate.id, "product.updated")
         return await this.productService_
           .withTransaction(manager)
           .update(existing.id, update)
@@ -344,8 +327,12 @@ class ShopifyProductService extends BaseService {
         }
 
         variant = this.addVariantOptions_(variant, options)
-        const match = variants.find((v) => v.sku === variant.sku)
+        const match = variants.find(
+          (v) => v.metadata.sh_id === variant.metadata.sh_id
+        )
         if (match) {
+          variant = this.removeUniqueConstraint_(variant)
+
           await this.productVariantService_
             .withTransaction(manager)
             .update(match.id, variant)
@@ -370,7 +357,9 @@ class ShopifyProductService extends BaseService {
           continue
         }
 
-        const match = updateVariants.find((v) => v.sku === variant.sku)
+        const match = updateVariants.find(
+          (v) => v.metadata.sh_id === variant.metadata.sh_id
+        )
         if (!match) {
           await this.productVariantService_
             .withTransaction(manager)
@@ -396,7 +385,11 @@ class ShopifyProductService extends BaseService {
 
       for (const option of updateOptions) {
         const match = options.find((o) => o.title === option.title)
-        if (!match) {
+        if (match) {
+          await this.productService_
+            .withTransaction(manager)
+            .updateOption(product.id, match.id, { title: option.title })
+        } else if (!match) {
           await this.productService_
             .withTransaction(manager)
             .addOption(id, option.title)
@@ -406,6 +399,7 @@ class ShopifyProductService extends BaseService {
       const result = await this.productService_.retrieve(id, {
         relations: ["variants", "options"],
       })
+
       return result
     })
   }
@@ -428,7 +422,7 @@ class ShopifyProductService extends BaseService {
    * @param {string} collectionId optional
    * @return {object} normalized object
    */
-  normalizeProduct_(product, collectionId) {
+  normalizeProduct_(product) {
     return {
       title: product.title,
       handle: product.handle,
@@ -446,9 +440,11 @@ class ShopifyProductService extends BaseService {
       tags: product.tags.split(",").map((tag) => this.normalizeTag_(tag)) || [],
       images: product.images.map((img) => img.src) || [],
       thumbnail: product.image?.src || null,
-      collection_id: collectionId || null,
       external_id: product.id,
       status: "proposed",
+      metadata: {
+        vendor: product.vendor,
+      },
     }
   }
 
@@ -548,6 +544,52 @@ class ShopifyProductService extends BaseService {
       value: tag,
     }
   }
+
+  handleDuplicateConstraint_(uniqueVal) {
+    return `DUP-${random(100, 999)}-${uniqueVal}`
+  }
+
+  async testUnique_(uniqueVal, type) {
+    // Test if the unique value has already been added, if it was then pass the value onto the duplicate handler and return the new value
+    const exists = await this.redis_.getUniqueValue(uniqueVal, type)
+
+    if (exists) {
+      const dupValue = this.handleDuplicateConstraint_(uniqueVal)
+      await this.redis_.addUniqueValue(dupValue, type)
+      return dupValue
+    }
+    // If it doesn't exist, we return the value
+    await this.redis_.addUniqueValue(uniqueVal, type)
+    return uniqueVal
+  }
+
+  async ensureVariantUnique_(variant) {
+    let { sku, ean, upc, barcode } = variant
+
+    if (sku) {
+      sku = await this.testUnique_(sku, "SKU")
+    }
+
+    if (ean) {
+      ean = await this.testUnique_(ean, "EAN")
+    }
+
+    if (upc) {
+      upc = await this.testUnique_(upc, "UPC")
+    }
+
+    if (barcode) {
+      barcode = await this.testUnique_(barcode, "BARCODE")
+    }
+
+    return { ...variant, sku, ean, upc, barcode }
+  }
+
+  removeUniqueConstraint_(update) {
+    const payload = omit(update, ["sku", "ean", "upc", "barcode"])
+
+    return payload
+  }
 }
 
 export default ShopifyProductService

packages/medusa-source-shopify/src/services/shopify-redis.js

@@ -1,3 +1,4 @@
+// shopify-redis
 import { BaseService } from "medusa-interfaces"
 import { IGNORE_THRESHOLD } from "../utils/const"
 
@@ -12,7 +13,7 @@ class shopifyRedisService extends BaseService {
   }
 
   async addIgnore(id, side) {
-    const key = `${id}_ignore_${side}`
+    const key = `sh_${id}_ignore_${side}`
     return await this.redis_.set(
       key,
       1,
@@ -22,7 +23,17 @@ class shopifyRedisService extends BaseService {
   }
 
   async shouldIgnore(id, action) {
-    const key = `${id}_ignore_${action}`
+    const key = `sh_${id}_ignore_${action}`
+    return await this.redis_.get(key)
+  }
+
+  async addUniqueValue(uniqueVal, type) {
+    const key = `sh_${uniqueVal}_${type}`
+    return await this.redis_.set(key, 1, "EX", 60 * 5)
+  }
+
+  async getUniqueValue(uniqueVal, type) {
+    const key = `sh_${uniqueVal}_${type}`
     return await this.redis_.get(key)
   }
 }

packages/medusa-source-shopify/src/services/shopify.js

@@ -6,6 +6,7 @@ class ShopifyService extends BaseService {
     {
       manager,
       shippingProfileService,
+      storeService,
       shopifyProductService,
       shopifyCollectionService,
       shopifyClientService,
@@ -26,6 +27,8 @@ class ShopifyService extends BaseService {
     this.collectionService_ = shopifyCollectionService
     /** @private @const {ShopifyRestClient} */
     this.client_ = shopifyClientService
+    /** @private @const {StoreService} */
+    this.store_ = storeService
   }
 
   withTransaction(transactionManager) {
@@ -40,6 +43,8 @@ class ShopifyService extends BaseService {
       shopifyClientService: this.client_,
       shopifyProductService: this.productService_,
       shopifyCollectionService: this.collectionService_,
+      shopifyBuildService: this.buildService_,
+      storeService: this.store_,
     })
 
     cloned.transactionManager_ = transactionManager
@@ -49,30 +54,80 @@ class ShopifyService extends BaseService {
 
   async importShopify() {
     return this.atomicPhase_(async (manager) => {
+      const updatedSinceQuery = await this.getAndUpdateBuildTime_()
+
       await this.shippingProfileService_.createDefault()
       await this.shippingProfileService_.createGiftCardDefault()
 
       const products = await this.client_.list(
         "products",
-        INCLUDE_PRESENTMENT_PRICES
+        INCLUDE_PRESENTMENT_PRICES,
+        updatedSinceQuery
+      )
+
+      const customCollections = await this.client_.list(
+        "custom_collections",
+        null,
+        updatedSinceQuery
+      )
+
+      const smartCollections = await this.client_.list(
+        "smart_collections",
+        null,
+        updatedSinceQuery
+      )
+
+      const collects = await this.client_.list(
+        "collects",
+        null,
+        updatedSinceQuery
+      )
+
+      const resolvedProducts = await Promise.all(
+        products.map(async (product) => {
+          return await this.productService_
+            .withTransaction(manager)
+            .create(product)
+        })
       )
-      const customCollections = await this.client_.list("custom_collections")
-      const smartCollections = await this.client_.list("smart_collections")
-      const collects = await this.client_.list("collects")
 
       await this.collectionService_
         .withTransaction(manager)
-        .createWithProducts(
-          collects,
-          [...customCollections, ...smartCollections],
-          products
-        )
+        .createCustomCollections(collects, customCollections, resolvedProducts)
 
-      for (const product of products) {
-        await this.productService_.withTransaction(manager).create(product)
-      }
+      await this.collectionService_
+        .withTransaction(manager)
+        .createSmartCollections(smartCollections, resolvedProducts)
     })
   }
+
+  async getAndUpdateBuildTime_() {
+    let buildtime = null
+    const store = await this.store_.retrieve()
+    if (!store) {
+      return {}
+    }
+
+    if (store.metadata?.source_shopify_bt) {
+      buildtime = store.metadata.source_shopify_bt
+    }
+
+    const payload = {
+      metadata: {
+        source_shopify_bt: new Date().toISOString(),
+      },
+    }
+
+    await this.store_.update(payload)
+
+    if (!buildtime) {
+      return {}
+    }
+
+    return {
+      updated_at_min: buildtime,
+    }
+  }
 }
 
 export default ShopifyService