Add webhook triggered on Order Cycle Open

Gemfile

@@ -136,6 +136,9 @@ gem 'view_component_reflex', '3.1.14.pre9'
 
 gem 'mini_portile2', '~> 2.8'
 
+gem "faraday"
+gem "private_address_check"
+
 group :production, :staging do
   gem 'ddtrace'
   gem 'rack-timeout'

Gemfile.lock

@@ -474,6 +474,7 @@ GEM
       ttfunk
     pg (1.2.3)
     power_assert (2.0.2)
+    private_address_check (0.5.0)
     pry (0.13.1)
       coderay (~> 1.1)
       method_source (~> 1.0)
@@ -801,6 +802,7 @@ DEPENDENCIES
   digest
   dotenv-rails
   factory_bot_rails (= 6.2.0)
+  faraday
   ffaker
   flipper
   flipper-active_record
@@ -843,6 +845,7 @@ DEPENDENCIES
   paypal-sdk-merchant (= 1.117.2)
   pdf-reader
   pg (~> 1.2.3)
+  private_address_check
   pry (~> 0.13.0)
   puma
   rack-mini-profiler (< 3.0.0)

app/controllers/webhook_endpoints_controller.rb

@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+
+class WebhookEndpointsController < ::BaseController
+  before_action :load_resource, only: :destroy
+
+  def create
+    webhook_endpoint = spree_current_user.webhook_endpoints.new(webhook_endpoint_params)
+
+    if webhook_endpoint.save
+      flash[:success] = t('.success')
+    else
+      flash[:error] = t('.error')
+    end
+
+    redirect_to redirect_path
+  end
+
+  def destroy
+    if @webhook_endpoint.destroy
+      flash[:success] = t('.success')
+    else
+      flash[:error] = t('.error')
+    end
+
+    redirect_to redirect_path
+  end
+
+  def load_resource
+    @webhook_endpoint = spree_current_user.webhook_endpoints.find(params[:id])
+  end
+
+  def webhook_endpoint_params
+    params.require(:webhook_endpoint).permit(:url)
+  end
+
+  def redirect_path
+    if request.referer.blank? || request.referer.include?(spree.account_path)
+      developer_settings_path
+    else
+      request.referer
+    end
+  end
+
+  def developer_settings_path
+    "#{spree.account_path}#/developer_settings"
+  end
+end

app/jobs/order_cycle_opened_job.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+# Trigger jobs for any order cycles that recently opened
+class OrderCycleOpenedJob < ApplicationJob
+  def perform
+    ActiveRecord::Base.transaction do
+      recently_opened_order_cycles.find_each do |order_cycle|
+        OrderCycleWebhookService.create_webhook_job(order_cycle, 'order_cycle.opened')
+      end
+      mark_as_opened(recently_opened_order_cycles)
+    end
+  end
+
+  private
+
+  def recently_opened_order_cycles
+    @recently_opened_order_cycles ||= OrderCycle
+      .where(opened_at: nil)
+      .where(orders_open_at: 1.hour.ago..Time.zone.now)
+      .lock.order(:id)
+  end
+
+  def mark_as_opened(order_cycles)
+    now = Time.zone.now
+    order_cycles.update_all(opened_at: now, updated_at: now)
+  end
+end

app/jobs/webhook_delivery_job.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+require "faraday"
+require "private_address_check"
+require "private_address_check/tcpsocket_ext"
+
+# Deliver a webhook payload
+# As a delayed job, it can run asynchronously and handle retries.
+class WebhookDeliveryJob < ApplicationJob
+  # General failed request error that we're going to use to signal
+  # the job runner to retry our webhook worker.
+  class FailedWebhookRequestError < StandardError; end
+
+  queue_as :default
+
+  def perform(url, event, payload)
+    body = {
+      id: job_id,
+      at: Time.zone.now.to_s,
+      event: event,
+      data: payload,
+    }
+
+    # Request user-submitted url, preventing any private connections being made
+    # (SSRF).
+    # This method may allow the socket to open, but is necessary in order to
+    # protect from TOC/TOU.
+    # Note that private_address_check provides some methods for pre-validating,
+    # but they're not as comprehensive and so unnecessary here. Simply
+    # momentarily opening sockets probably can't cause DoS or other damage.
+    PrivateAddressCheck.only_public_connections do
+      notify_endpoint(url, body)
+    end
+  end
+
+  def notify_endpoint(url, body)
+    connection = Faraday.new(
+      request: { timeout: 30 },
+      headers: {
+        'User-Agent' => 'openfoodnetwork_webhook/1.0',
+        'Content-Type' => 'application/json',
+      }
+    )
+    response = connection.post(url, body.to_json)
+
+    # Raise a failed request error and let job runner handle retrying.
+    # In theory, only 5xx errors should be retried, but who knows.
+    raise FailedWebhookRequestError, response.status.to_s unless response.success?
+  end
+end

app/models/order_cycle.rb

@@ -34,6 +34,7 @@ class OrderCycle < ApplicationRecord
 
   attr_accessor :incoming_exchanges, :outgoing_exchanges
 
+  before_update :reset_opened_at, if: :will_save_change_to_orders_open_at?
   before_update :reset_processed_at, if: :will_save_change_to_orders_close_at?
   after_save :sync_subscriptions, if: :opening?
 
@@ -333,6 +334,14 @@ def orders_close_at_after_orders_open_at?
     errors.add(:orders_close_at, :after_orders_open_at)
   end
 
+  def reset_opened_at
+    # Reset only if order cycle is opening again at a later date
+    return unless orders_open_at.present? && orders_open_at_was.present?
+    return unless orders_open_at > orders_open_at_was
+
+    self.opened_at = nil
+  end
+
   def reset_processed_at
     return unless orders_close_at.present? && orders_close_at_was.present?
     return unless orders_close_at > orders_close_at_was

app/models/spree/user.rb

@@ -38,7 +38,10 @@ class User < ApplicationRecord
     has_many :customers
     has_many :credit_cards
     has_many :report_rendering_options, class_name: "::ReportRenderingOptions", dependent: :destroy
+    has_many :webhook_endpoints, dependent: :destroy
+
     accepts_nested_attributes_for :enterprise_roles, allow_destroy: true
+    accepts_nested_attributes_for :webhook_endpoints
 
     accepts_nested_attributes_for :bill_address
     accepts_nested_attributes_for :ship_address

app/models/webhook_endpoint.rb

@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+
+# Records a webhook url to send notifications to
+class WebhookEndpoint < ApplicationRecord
+  validates :url, presence: true
+end

app/services/order_cycle_webhook_service.rb

@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+# Create a webhook payload for an order cycle event.
+# The payload will be delivered asynchronously.
+class OrderCycleWebhookService
+  def self.create_webhook_job(order_cycle, event)
+    webhook_payload = order_cycle
+      .slice(:id, :name, :orders_open_at, :orders_close_at, :coordinator_id)
+      .merge(coordinator_name: order_cycle.coordinator.name)
+
+    # Endpoints for coordinator owner
+    webhook_endpoints = order_cycle.coordinator.owner.webhook_endpoints
+
+    # Plus unique endpoints for distributor owners (ignore duplicates)
+    webhook_endpoints |= order_cycle.distributors.map(&:owner).flat_map(&:webhook_endpoints)
+
+    webhook_endpoints.each do |endpoint|
+      WebhookDeliveryJob.perform_later(endpoint.url, event, webhook_payload)
+    end
+  end
+end

app/services/permitted_attributes/user.rb

@@ -15,7 +15,10 @@ def call(extra_permitted_attributes = [])
     private
 
     def permitted_attributes
-      [:email, :password, :password_confirmation, :disabled]
+      [
+        :email, :password, :password_confirmation, :disabled,
+        { webhook_endpoints_attributes: [:id, :url] },
+      ]
     end
   end
 end

app/views/spree/users/_developer_settings.html.haml

@@ -1,3 +1,4 @@
 %script{ type: "text/ng-template", id: "account/developer_settings.html" }
   %h3= t('.title')
   = render partial: 'api_keys'
+  = render partial: 'webhook_endpoints'

app/views/spree/users/_webhook_endpoints.html.haml

@@ -0,0 +1,33 @@
+%section{ id: "webhook_endpoints" }
+  %hr
+  %h3= t('.title')
+  %p= t('.description')
+
+  %table{width: "100%"}
+    %thead
+      %tr
+        %th= t('.event_type.header')
+        %th= t('.url.header')
+        %th.actions
+    %tbody
+      -# Existing endpoints
+      - @user.webhook_endpoints.each do |webhook_endpoint|
+        %tr
+          %td= t('.event_types.order_cycle_opened') # For now, we only support one type.
+          %td= webhook_endpoint.url
+          %td.actions
+            - if webhook_endpoint.persisted?
+              = button_to account_webhook_endpoint_path(webhook_endpoint), method: :delete,
+                class: "tiny alert no-margin",
+                data: { confirm: I18n.t(:are_you_sure)} do
+                = I18n.t(:delete)
+
+      -# Create new
+      - if @user.webhook_endpoints.empty? # Only one allowed for now.
+        %tr
+          %td= t('.event_types.order_cycle_opened') # For now, we only support one type.
+          %td
+            = form_for(@user.webhook_endpoints.build, url: account_webhook_endpoints_path, id: 'new_webhook_endpoint') do |f|
+              = f.url_field :url, placeholder: t('.url.create_placeholder'), required: true, size: 64
+          %td.actions
+            = button_tag t(:create), class: 'button primary tiny no-margin', form: 'new_webhook_endpoint'

config/locales/en.yml

@@ -3562,6 +3562,14 @@ See the %{link} to find out more about %{sitename}'s features and to start using
   previous: "Previous"
   last: "Last"
 
+  webhook_endpoints:
+    create:
+      success: Webhook endpoint successfully created
+      error: Webhook endpoint failed to create
+    destroy:
+      success: Webhook endpoint successfully deleted
+      error: Webhook endpoint failed to delete
+
   spree:
     order_updated: "Order Updated"
     add_country: "Add country"
@@ -4357,6 +4365,16 @@ See the %{link} to find out more about %{sitename}'s features and to start using
       api_keys:
         regenerate_key: "Regenerate Key"
         title: API key
+      webhook_endpoints:
+        title: Webhook Endpoints
+        description: Events in the system may trigger webhooks to external systems.
+        event_types:
+          order_cycle_opened: Order Cycle Opened
+        event_type:
+          header: Event type
+        url:
+          header: Endpoint URL
+          create_placeholder: Enter the URL of the remote webhook endpoint
       developer_settings:
         title: Developer Settings
       form:

config/routes/spree.rb

@@ -32,7 +32,9 @@
     put '/password/change' => 'user_passwords#update', :as => :update_password
   end
 
-  resource :account, :controller => 'users'
+  resource :account, :controller => 'users' do
+    resources :webhook_endpoints, only: [:create, :destroy], controller: '/webhook_endpoints'
+  end
 
   match '/admin/orders/bulk_management' => 'admin/orders#bulk_management', :as => "admin_bulk_order_management", via: :get
   match '/admin/payment_methods/show_provider_preferences' => 'admin/payment_methods#show_provider_preferences', :via => :get

config/sidekiq.yml

@@ -15,5 +15,7 @@
       every: "5m"
     SubscriptionConfirmJob:
       every: "5m"
+    OrderCycleOpenedJob:
+      every: "5m"
     OrderCycleClosingJob:
       every: "5m"

db/migrate/20220919063831_add_opened_at_to_order_cycle.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddOpenedAtToOrderCycle < ActiveRecord::Migration[6.1]
+  def change
+    add_column :order_cycles, :opened_at, :timestamp
+  end
+end

db/migrate/20221028051650_create_webhook_endpoints.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+class CreateWebhookEndpoints < ActiveRecord::Migration[6.1]
+  def change
+    create_table :webhook_endpoints do |t|
+      t.string :url, null: false
+
+      t.timestamps null: false
+    end
+  end
+end

db/migrate/20221028053214_add_spree_user_reference_to_webhook_endpoint.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class AddSpreeUserReferenceToWebhookEndpoint < ActiveRecord::Migration[6.1]
+  def change
+    add_column :webhook_endpoints, :user_id, :bigint, default: 0, null: false
+    add_index :webhook_endpoints, :user_id
+    add_foreign_key :webhook_endpoints, :spree_users, column: :user_id
+  end
+end

db/schema.rb

@@ -310,6 +310,7 @@
     t.datetime "processed_at"
     t.boolean "automatic_notifications", default: false
     t.boolean "mails_sent", default: false
+    t.datetime "opened_at"
   end
 
   create_table "order_cycles_distributor_payment_methods", id: false, force: :cascade do |t|
@@ -1189,6 +1190,14 @@
     t.index ["item_type", "item_id"], name: "index_versions_on_item_type_and_item_id"
   end
 
+  create_table "webhook_endpoints", force: :cascade do |t|
+    t.string "url", null: false
+    t.datetime "created_at", precision: 6, null: false
+    t.datetime "updated_at", precision: 6, null: false
+    t.bigint "user_id", default: 0, null: false
+    t.index ["user_id"], name: "index_webhook_endpoints_on_user_id"
+  end
+
   add_foreign_key "active_storage_attachments", "active_storage_blobs", column: "blob_id"
   add_foreign_key "active_storage_variant_records", "active_storage_blobs", column: "blob_id"
   add_foreign_key "adjustment_metadata", "enterprises", name: "adjustment_metadata_enterprise_id_fk"
@@ -1293,4 +1302,5 @@
   add_foreign_key "subscriptions", "spree_shipping_methods", column: "shipping_method_id", name: "subscriptions_shipping_method_id_fk"
   add_foreign_key "variant_overrides", "enterprises", column: "hub_id", name: "variant_overrides_hub_id_fk"
   add_foreign_key "variant_overrides", "spree_variants", column: "variant_id", name: "variant_overrides_variant_id_fk"
+  add_foreign_key "webhook_endpoints", "spree_users", column: "user_id"
 end