add_to_cart_operation.rb

# NOTE: Each task is a map function wrapped in a HOC for handling the return data.
# The annotation of each task is `state -> mixed | state | exception` and the HOC is
# `state -> (state -> mixed | exception) -> state`. `error` is like a task, but instead:
# `exception -> mixed` wrapped in a HOC that matches `exception -> (exception -> mixed) -> exception`.
class AddToCartOperation < ApplicationOperation
  task :check_for_missing_product
  task :carbon_copy_cart_item
  task :lock
  task :persist
  task :publish
  error :notify, catch: ProductMissingFromCartItemError
  error :raise # NOTE: This lets us pass the buck on exceptions to whatever is controlling the operation

  state :check_for_missing_product do
    field :cart_item, type: Strict::Object(CartItem)
  end
  # NOTE: A task either raises an exception or returns a next state, everything 
  # else assumes passing previous state
  def check_for_missing_product(state)
    raise ProductMissingFromCartItemError if state.cart_item.product.nil?
  end

  state :carbon_copy_cart_item do
    field :cart_item, type: Strict::Object(CartItem)
  end
  def carbon_copy_cart_item(state)
    state.cart_item.carbon_copy
  end

  state :lock do
    field :cart_item, type: Strict::Object(CartItem)
  end
  def lock(state)
    GlobalLock.(state.cart_item.account, state.cart_item, expires_in: 15.minutes)
  end

  state :persist do
    field :cart_item, type: Strict::Object(CartItem)
  end
  # NOTE: The `fresh()` will create a new state and indicate to the pipeline
  # to not use the previous state
  def persist(state)
    CartItem.transaction do
      state.cart_item.save!
    end

    return fresh(current_account: state.cart_item.owner, cart_item: state.cart_item)
  end

  state :publish do
    field :cart_item, type: Strict::Object(CartItem)
    field :current_account, type: Strict::Object(Account)
  end
  def publish(state)
    CartItemPickedMessage.(subject: state.cart_item, to: state.current_account).via_pubsub.deliver_later!
  end

  # NOTE: Normally we catch any StandardError or greater exception, but this is 
  # caught via the `catch:` clause so we can notify developers
  def notify(exception)
    Bugsnag.notify(exception)
  end
end

application_operation.rb

class ApplicationOperation
  include Operational
end

operational.rb

require "active_support/concerns"

module Operational
  extend ActiveSupport::Concern

  State = Struct.new(:raw)
  Drift = Struct.new(:raw)

  attr_reader :raw
  attr_reader :current

  def initialize(raw:)
    @raw = raw
  end

  def call(start: 0)
    right.from(start).reduce(current || raw) do |state, function|
      # NOTE: We store this so we can go drift back if an error tells us to
      @current = state

      value = function.receiver.method(function.name).(schemas.fetch(function.name).new(state))

      case value
      when State then value.raw
      when Drift then break call(start: right.at(value.to))
      else state
      end
    end
  rescue *@failures.select(&:catch).map(&:catch).uniq => handled_exception
    left.select do |failure|
      failure.catch === handled_exception
    end.reduce(handled_exception) do |exception, function|
      value = function.receiver.method(function.name).(exception)

      if value.kind_of?(Drift)
        break call(start: right.at(value.to))
      else
        exception
      end
    end
  end

  def fresh(raw)
    State.new(raw)
  end

  def drift(to:)
    Drift.new(to)
  end

  private def left
    self.class.left
  end

  private def right
    self.class.right
  end

  class_methods do
    def task(name, receiver: self)
      right.<<({name: name, receiver: receiver || self})
    end

    def error(name, receiver: self, catch: StandardError)
      left.<<({name: name, receiver: receiver || self, catch: catch || StandardError})
    end

    def call(raw)
      new(raw: raw).call
    end

    def right
      @right ||= Array.new
    end

    def left
      @left ||= Array.new
    end
  end
end