Merge pull request #12899 from mkllnk/dfc-stock-check

DFC Orders update for pilot 1 and 2

app/controllers/admin/dfc_product_imports_controller.rb

@@ -26,7 +26,15 @@ module Admin
       # * First step: import all products for given enterprise.
       # * Second step: render table and let user decide which ones to import.
       imported = graph.map do |subject|
-        import_product(subject, enterprise)
+        next unless subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
+
+        existing_variant = enterprise.supplied_variants.linked_to(subject.semanticId)
+
+        if existing_variant
+          SuppliedProductBuilder.update_product(subject, existing_variant)
+        else
+          SuppliedProductBuilder.store_product(subject, enterprise)
+        end
       end
 
       @count = imported.compact.count
@@ -37,18 +45,5 @@ module Admin
     def fetch_catalog(url)
       DfcRequest.new(spree_current_user).call(url)
     end
-
-    # Most of this code is the same as in the DfcProvider::SuppliedProductsController.
-    def import_product(subject, enterprise)
-      return unless subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
-
-      variant = SuppliedProductBuilder.import_variant(subject, enterprise)
-      product = variant.product
-
-      product.save! if product.new_record?
-      variant.save! if variant.new_record?
-
-      variant
-    end
   end
 end

app/controllers/concerns/checkout_callbacks.rb

@@ -9,6 +9,12 @@ module CheckoutCallbacks
     # Otherwise we fail on duplicate indexes or end up with negative stock.
     prepend_around_action CurrentOrderLocker, only: [:edit, :update]
 
+    # We want to download the latest stock data before anything else happens.
+    # We don't want it to be in the same database transaction as the order
+    # locking because this action locks a different set of variants and it
+    # could cause race conditions.
+    prepend_around_action :sync_stock, only: :update
+
     prepend_before_action :check_hub_ready_for_checkout
     prepend_before_action :check_order_cycle_expiry
     prepend_before_action :require_order_cycle
@@ -25,6 +31,14 @@ module CheckoutCallbacks
 
   private
 
+  def sync_stock
+    if current_order&.state == "confirmation"
+      StockSyncJob.sync_linked_catalogs_now(current_order)
+    end
+
+    yield
+  end
+
   def load_order
     @order = current_order
     @order.manual_shipping_selection = true
@@ -63,12 +77,6 @@ module CheckoutCallbacks
     end
   end
 
-  def valid_order_line_items?
-    @order.insufficient_stock_lines.empty? &&
-      OrderCycles::DistributedVariantsService.new(@order.order_cycle, @order.distributor).
-        distributes_order_variants?(@order)
-  end
-
   def ensure_order_not_completed
     redirect_to main_app.cart_path if @order.completed?
   end

app/jobs/stock_sync_job.rb

@@ -9,15 +9,10 @@ class StockSyncJob < ApplicationJob
   # enqueue a new job. That should save some time loading the order with
   # all the stock data to make this decision.
   def self.sync_linked_catalogs(order)
-    stock_controlled_variants = order.variants.reject(&:on_demand)
-    links = SemanticLink.where(variant_id: stock_controlled_variants.map(&:id))
-    semantic_ids = links.pluck(:semantic_id)
-
-    return if semantic_ids.empty?
-
     user = order.distributor.owner
-    reference_id = semantic_ids.first # Assuming one catalog for now.
-    perform_later(user, reference_id)
+    catalog_ids(order).each do |catalog_id|
+      perform_later(user, catalog_id)
+    end
   rescue StandardError => e
     # Errors here shouldn't affect the shopping. So let's report them
     # separately:
@@ -26,27 +21,59 @@ class StockSyncJob < ApplicationJob
     end
   end
 
-  def perform(user, semantic_id)
-    urls = FdcUrlBuilder.new(semantic_id)
-    json_catalog = DfcRequest.new(user).call(urls.catalog_url)
-    graph = DfcIo.import(json_catalog)
-
-    products = graph.select do |subject|
-      subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
+  def self.sync_linked_catalogs_now(order)
+    user = order.distributor.owner
+    catalog_ids(order).each do |catalog_id|
+      perform_now(user, catalog_id)
     end
-    products_by_id = products.index_by(&:semanticId)
-    product_ids = products_by_id.keys
-    variants = Spree::Variant.where(supplier: user.enterprises)
-      .includes(:semantic_links).references(:semantic_links)
-      .where(semantic_links: { semantic_id: product_ids })
-
-    variants.each do |variant|
-      next if variant.on_demand
-
-      product = products_by_id[variant.semantic_links[0].semantic_id]
-      catalog_item = product&.catalogItems&.first
-      CatalogItemBuilder.apply_stock(catalog_item, variant)
-      variant.stock_items[0].save!
+  rescue StandardError => e
+    # Errors here shouldn't affect the shopping. So let's report them
+    # separately:
+    Bugsnag.notify(e) do |payload|
+      payload.add_metadata(:order, order)
     end
   end
+
+  def self.catalog_ids(order)
+    stock_controlled_variants = order.variants.reject(&:on_demand)
+    links = SemanticLink.where(variant_id: stock_controlled_variants.map(&:id))
+    semantic_ids = links.pluck(:semantic_id)
+    semantic_ids.map do |product_id|
+      FdcUrlBuilder.new(product_id).catalog_url
+    end.uniq
+  end
+
+  def perform(user, catalog_id)
+    products = load_products(user, catalog_id)
+    products_by_id = products.index_by(&:semanticId)
+    product_ids = products_by_id.keys
+    variants = linked_variants(user.enterprises, product_ids)
+
+    # Avoid race condition between checkout and stock sync.
+    Spree::Variant.transaction do
+      variants.order(:id).lock.each do |variant|
+        next if variant.on_demand
+
+        product = products_by_id[variant.semantic_links[0].semantic_id]
+        catalog_item = product&.catalogItems&.first
+        CatalogItemBuilder.apply_stock(catalog_item, variant)
+        variant.stock_items[0].save!
+      end
+    end
+  end
+
+  def load_products(user, catalog_id)
+    json_catalog = DfcRequest.new(user).call(catalog_id)
+    graph = DfcIo.import(json_catalog)
+
+    graph.select do |subject|
+      subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
+    end
+  end
+
+  def linked_variants(enterprises, product_ids)
+    Spree::Variant.where(supplier: enterprises)
+      .includes(:semantic_links).references(:semantic_links)
+      .where(semantic_links: { semantic_id: product_ids })
+  end
 end

engines/dfc_provider/app/controllers/dfc_provider/supplied_products_controller.rb

@@ -14,18 +14,10 @@ module DfcProvider
 
       return head :bad_request unless supplied_product
 
-      variant = SuppliedProductBuilder.import_variant(
+      variant = SuppliedProductBuilder.store_product(
         supplied_product,
         current_enterprise,
       )
-      product = variant.product
-
-      if variant.new_record?
-        variant.supplier = current_enterprise
-        variant.save!
-      end
-
-      product.save! if product.new_record?
 
       supplied_product = SuppliedProductBuilder.supplied_product(variant)
       render json: DfcIo.export(supplied_product)
@@ -41,10 +33,7 @@ module DfcProvider
 
       return head :bad_request unless supplied_product
 
-      SuppliedProductBuilder.apply(supplied_product, variant)
-
-      variant.product.save!
-      variant.save!
+      SuppliedProductBuilder.update_product(supplied_product, variant)
     end
 
     private

engines/dfc_provider/app/services/image_builder.rb

@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+require "private_address_check"
+require "private_address_check/tcpsocket_ext"
+
+class ImageBuilder < DfcBuilder
+  def self.import(image_link)
+    url = URI.parse(image_link)
+    filename = File.basename(image_link)
+
+    Spree::Image.new.tap do |image|
+      PrivateAddressCheck.only_public_connections do
+        image.attachment.attach(io: url.open, filename:)
+      end
+    end
+  rescue StandardError
+    # Any URL parsing or network error shouldn't impact the product import
+    # at all. Maybe we'll add UX for error handling later.
+    nil
+  end
+end

engines/dfc_provider/app/services/quantitative_value_builder.rb

@@ -29,11 +29,18 @@ class QuantitativeValueBuilder < DfcBuilder
 
   def self.apply(quantity, product)
     measure, unit_name, unit_scale = map_unit(quantity.unit)
+    value = quantity.value.to_f * unit_scale
+
+    if measure.in?(%w(weight volume)) && value <= 0
+      measure = "items"
+      unit_name = "items"
+      value = 1
+    end
 
     product.variant_unit = measure
     product.variant_unit_name = unit_name if measure == "items"
     product.variant_unit_scale = unit_scale
-    product.unit_value = quantity.value.to_f * unit_scale
+    product.unit_value = value
   end
 
   # Map DFC units to OFN fields:

engines/dfc_provider/app/services/supplied_product_builder.rb

@@ -1,8 +1,5 @@
 # frozen_string_literal: true
 
-require "private_address_check"
-require "private_address_check/tcpsocket_ext"
-
 class SuppliedProductBuilder < DfcBuilder
   def self.supplied_product(variant)
     id = urls.enterprise_supplied_product_url(
@@ -26,27 +23,40 @@ class SuppliedProductBuilder < DfcBuilder
     )
   end
 
+  def self.store_product(subject, enterprise)
+    return unless subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
+
+    variant = SuppliedProductBuilder.import_variant(subject, enterprise)
+    product = variant.product
+
+    product.save! if product.new_record?
+    variant.save! if variant.new_record?
+
+    variant
+  end
+
+  def self.update_product(supplied_product, variant)
+    apply(supplied_product, variant)
+
+    variant.product.save!
+    variant.save!
+
+    variant
+  end
+
   def self.import_variant(supplied_product, supplier)
     product = referenced_spree_product(supplied_product, supplier)
 
     if product
-      Spree::Variant.new(
-        product:,
-        supplier:,
-        price: 0,
-      ).tap do |variant|
+      Spree::Variant.new( product:, supplier:, price: 0,).tap do |variant|
         apply(supplied_product, variant)
       end
     else
       product = import_product(supplied_product, supplier)
-      product.variants.first
+      product.variants.first.tap { |variant| apply(supplied_product, variant) }
     end.tap do |variant|
       link = supplied_product.semanticId
-      catalog_item = supplied_product&.catalogItems&.first
-      offer = catalog_item&.offers&.first
       variant.semantic_links.new(semantic_id: link) if link.present?
-      CatalogItemBuilder.apply_stock(catalog_item, variant)
-      OfferBuilder.apply(offer, variant)
     end
   end
 
@@ -74,7 +84,7 @@ class SuppliedProductBuilder < DfcBuilder
       price: 0, # will be in DFC Offer
       supplier_id: supplier.id,
       primary_taxon_id: taxon(supplied_product).id,
-      image: image(supplied_product),
+      image: ImageBuilder.import(supplied_product.image),
     ).tap do |product|
       QuantitativeValueBuilder.apply(supplied_product.quantity, product)
       product.ensure_standard_variant
@@ -88,6 +98,11 @@ class SuppliedProductBuilder < DfcBuilder
     variant.primary_taxon = taxon(supplied_product)
     QuantitativeValueBuilder.apply(supplied_product.quantity, variant.product)
     variant.unit_value = variant.product.unit_value
+
+    catalog_item = supplied_product&.catalogItems&.first
+    offer = catalog_item&.offers&.first
+    CatalogItemBuilder.apply_stock(catalog_item, variant)
+    OfferBuilder.apply(offer, variant)
   end
 
   def self.product_type(variant)
@@ -104,20 +119,5 @@ class SuppliedProductBuilder < DfcBuilder
     Spree::Taxon.find_by(dfc_id:) || Spree::Taxon.first
   end
 
-  def self.image(supplied_product)
-    url = URI.parse(supplied_product.image)
-    filename = File.basename(supplied_product.image)
-
-    Spree::Image.new.tap do |image|
-      PrivateAddressCheck.only_public_connections do
-        image.attachment.attach(io: url.open, filename:)
-      end
-    end
-  rescue StandardError
-    # Any URL parsing or network error shouldn't impact the product import
-    # at all. Maybe we'll add UX for error handling later.
-    nil
-  end
-
   private_class_method :product_type, :taxon
 end

engines/dfc_provider/spec/fixtures/files/product.GET.json

@@ -0,0 +1,38 @@
+{
+  "@context": "https://www.datafoodconsortium.org",
+  "@graph": [
+    {
+      "@id": "_:b433",
+      "@type": "dfc-b:QuantitativeValue",
+      "dfc-b:hasUnit": "dfc-m:Kilogram",
+      "dfc-b:value": "0"
+    },
+    {
+      "@id": "_:b434",
+      "@type": "dfc-b:Price",
+      "dfc-b:VATrate": "1",
+      "dfc-b:hasUnit": "dfc-m:Euro",
+      "dfc-b:value": "12.06"
+    },
+    {
+      "@id": "https://example-producer.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/5c21b9-95/SuppliedProducts/49055026544964",
+      "@type": "dfc-b:SuppliedProduct",
+      "dfc-b:hasQuantity": "_:b433",
+      "dfc-b:name": "Fillet Steak - 201g x 1 Steak",
+      "dfc-b:referencedBy": "https://example-producer.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/5c21b9-95/SuppliedProducts/49055026544964/CatalogItem"
+    },
+    {
+      "@id": "https://example-producer.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/5c21b9-95/SuppliedProducts/49055026544964/CatalogItem",
+      "@type": "dfc-b:CatalogItem",
+      "dfc-b:offeredThrough": "https://example-producer.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/5c21b9-95/SuppliedProducts/49055026544964/Offer",
+      "dfc-b:stockLimitation": "11"
+    },
+    {
+      "@id": "https://example-producer.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/5c21b9-95/SuppliedProducts/49055026544964/Offer",
+      "@type": "dfc-b:Offer",
+      "dfc-b:hasPrice": {
+        "@id": "_:b434"
+      }
+    }
+  ]
+}

engines/dfc_provider/spec/services/affiliate_sales_query_spec.rb

@@ -11,6 +11,13 @@ RSpec.describe AffiliateSalesQuery do
     let(:yesterday) { Time.zone.yesterday }
     let(:tomorrow) { Time.zone.tomorrow }
 
+    around do |example|
+      # Query dates are interpreted as UTC while the spec runs in
+      # Melbourne time. At noon in Melbourne, the date is the same.
+      # That simplifies the spec.
+      Timecop.travel(Time.zone.today.noon, &example)
+    end
+
     it "returns data" do
       # Test data creation takes time.
       # So I'm executing more tests in one `it` block here.

engines/dfc_provider/spec/services/supplied_product_builder_spec.rb

@@ -84,6 +84,33 @@ RSpec.describe SuppliedProductBuilder do
     end
   end
 
+  describe ".store_product" do
+    let(:subject) { builder.store_product(product, supplier) }
+    let(:product) {
+      DfcIo.import(product_json).find do |subject|
+        subject.is_a? DataFoodConsortium::Connector::SuppliedProduct
+      end
+    }
+    let(:product_json) { ExampleJson.read("product.GET") }
+
+    before do
+      taxon.save!
+    end
+
+    it "stores a new Spree Product and Variant" do
+      expect { subject }.to change {
+        Spree::Product.count
+      }.by(1)
+
+      expect(subject).to be_a(Spree::Variant)
+      expect(subject).to be_valid
+      expect(subject).to be_persisted
+      expect(subject.name).to eq("Fillet Steak - 201g x 1 Steak")
+      expect(subject.variant_unit).to eq("items")
+      expect(subject.unit_value).to eq(1)
+    end
+  end
+
   describe ".import_product" do
     let(:supplied_product) do
       DfcProvider::SuppliedProduct.new(

spec/controllers/checkout_controller_spec.rb

@@ -451,6 +451,20 @@ RSpec.describe CheckoutController, type: :controller do
           expect(response).to redirect_to order_path(order, order_token: order.token)
           expect(order.reload.state).to eq "complete"
         end
+
+        it "syncs stock before locking the order" do
+          actions = []
+          expect(StockSyncJob).to receive(:sync_linked_catalogs_now) do
+            actions << "sync stock"
+          end
+          expect(CurrentOrderLocker).to receive(:around) do
+            actions << "lock order"
+          end
+
+          put(:update, params:)
+
+          expect(actions).to eq ["sync stock", "lock order"]
+        end
       end
 
       context "when accepting T&Cs is required" do

spec/jobs/stock_sync_job_spec.rb

@@ -10,6 +10,9 @@ RSpec.describe StockSyncJob do
   let(:beans_retail_link) {
     "https://env-0105831.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/test-hodmedod/SuppliedProducts/44519466467635"
   }
+  let(:catalog_link) {
+    "https://env-0105831.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/test-hodmedod/SuppliedProducts"
+  }
 
   describe ".sync_linked_catalogs" do
     subject { StockSyncJob.sync_linked_catalogs(order) }
@@ -23,7 +26,7 @@ RSpec.describe StockSyncJob do
       )
 
       expect { subject }.to enqueue_job(StockSyncJob)
-        .with(user, beans_retail_link)
+        .with(user, catalog_link)
     end
 
     it "reports errors" do
@@ -34,8 +37,32 @@ RSpec.describe StockSyncJob do
     end
   end
 
-  describe "#peform" do
-    subject { StockSyncJob.perform_now(user, beans_retail_link) }
+  describe ".sync_linked_catalogs_now" do
+    subject { StockSyncJob.sync_linked_catalogs_now(order) }
+    it "ignores products without semantic link" do
+      expect(StockSyncJob).not_to receive(:perform_now)
+      expect { subject }.not_to enqueue_job(StockSyncJob)
+    end
+
+    it "performs stock check now" do
+      beans.semantic_links << SemanticLink.new(
+        semantic_id: beans_retail_link
+      )
+
+      expect(StockSyncJob).to receive(:perform_now).with(user, catalog_link)
+      expect { subject }.not_to raise_error
+    end
+
+    it "reports errors" do
+      expect(order).to receive(:variants).and_raise("test error")
+      expect(Bugsnag).to receive(:notify).and_call_original
+
+      expect { subject }.not_to raise_error
+    end
+  end
+
+  describe "#perform" do
+    subject { StockSyncJob.perform_now(user, catalog_link) }
 
     before do
       distributor.save!

spec/system/admin/dfc_product_import_spec.rb

@@ -48,6 +48,10 @@ RSpec.describe "DFC Product Import" do
       refresh_token: ENV.fetch("OPENID_REFRESH_TOKEN"),
       updated_at: 1.day.ago,
     )
+    product_id =
+      "https://env-0105831.jcloud-ver-jpe.ik-server.com/api/dfc/Enterprises/test-hodmedod/SuppliedProducts/44519466467635"
+    linked_variant = source_product.variants.first
+    linked_variant.semantic_links << SemanticLink.new(semantic_id: product_id)
 
     visit admin_product_import_path
 
@@ -58,14 +62,15 @@ RSpec.describe "DFC Product Import" do
 
     expect {
       click_button "Import"
-    }.to change {
-      enterprise.supplied_products.count
-    }
+      linked_variant.reload
+    }.to change { enterprise.supplied_products.count }
+      .and change { linked_variant.display_name }
+      .and change { linked_variant.unit_value }
+      .and change { linked_variant.price }
 
     expect(page).to have_content "Importing a DFC product catalog"
 
     product = Spree::Product.last
-
     expect(product.variants[0].semantic_links).to be_present
     expect(product.image).to be_present
   end