Add extension for ROM transactions

Gemfile

@@ -24,3 +24,8 @@ group :test do
   gem "rspec"
   gem "simplecov"
 end
+
+group :development, :test do
+  gem "rom-sql"
+  gem "sqlite3"
+end

lib/dry/operation.rb

@@ -92,6 +92,10 @@ module Dry
   #
   # The behavior configured by {ClassContext#operate_on} and {ClassContext#skip_prepending} is
   # inherited by subclasses.
+  #
+  # Some extensions are available under the `Dry::Operation::Extensions`
+  # namespace, providing additional functionality that can be included in your
+  # operation classes.
   class Operation
     def self.loader
       @loader ||= Zeitwerk::Loader.new.tap do |loader|
@@ -102,33 +106,70 @@ def self.loader
         loader.ignore(
           "#{root}/dry/operation/errors.rb"
         )
+        loader.inflector.inflect("rom" => "ROM")
       end
     end
     loader.setup
 
+    FAILURE_TAG = :halt
+    private_constant :FAILURE_TAG
+
     extend ClassContext
     include Dry::Monads::Result::Mixin
 
     # Wraps block's return value in a {Dry::Monads::Result::Success}
     #
-    # Catches :halt and returns it
+    # Catches `:halt` and returns it
     #
     # @yieldreturn [Object]
     # @return [Dry::Monads::Result::Success]
     # @see #step
     def steps(&block)
-      catch(:halt) { Success(block.call) }
+      catching_failure { Success(block.call) }
     end
 
     # Unwraps a {Dry::Monads::Result::Success}
     #
-    # Throws :halt with a {Dry::Monads::Result::Failure} on failure.
+    # Throws `:halt` with a {Dry::Monads::Result::Failure} on failure.
     #
     # @param result [Dry::Monads::Result]
     # @return [Object] wrapped value
     # @see #steps
     def step(result)
-      result.value_or { throw :halt, result }
+      result.value_or { throw_failure(result) }
+    end
+
+    # Invokes a callable in case of block's failure
+    #
+    # Throws `:halt` with a {Dry::Monads::Result::Failure} on failure.
+    #
+    # This method is useful when you want to perform some side-effect when a
+    # failure is encountered. It's meant to be used within the {#steps} block
+    # commonly wrapping a sub-set of {#step} calls.
+    #
+    # @param handler [#call] a callable that will be called when a failure is encountered
+    # @yieldreturn [Object]
+    # @return [Object] the block's return value
+    def intercepting_failure(handler, &block)
+      output = catching_failure(&block)
+
+      case output
+      when Failure
+        handler.()
+        throw_failure(output)
+      else
+        output
+      end
+    end
+
+    private
+
+    def catching_failure(&block)
+      catch(FAILURE_TAG, &block)
+    end
+
+    def throw_failure(failure)
+      throw FAILURE_TAG, failure
     end
   end
 end

lib/dry/operation/errors.rb

@@ -22,5 +22,18 @@ def initialize(methods:)
         MSG
       end
     end
+
+    # Missing dependency required by an extension
+    class MissingDependencyError < ::StandardError
+      def initialize(gem:, extension:)
+        super <<~MSG
+          To use the #{extension} extension, you first need to install the \
+          #{gem} gem. Please, add it to your Gemfile and run bundle install
+        MSG
+      end
+    end
+
+    # An error related to an extension
+    class ExtensionError < ::StandardError; end
   end
 end

lib/dry/operation/extensions/rom.rb

@@ -0,0 +1,118 @@
+# frozen_string_literal: true
+
+require "dry/operation/errors"
+
+begin
+  require "rom-sql"
+rescue LoadError
+  raise Dry::Operation::MissingDependencyError.new(gem: "rom-sql", extension: "ROM")
+end
+
+module Dry
+  class Operation
+    module Extensions
+      # Add rom transaction support to operations
+      #
+      # When this extension is included, you can use a `#transaction` method
+      # to wrap the desired steps in a rom transaction. If any of the steps
+      # returns a `Dry::Monads::Result::Failure`, the transaction will be rolled
+      # back and, as usual, the rest of the flow will be skipped.
+      #
+      # The extension expects the including class to give access to the rom
+      # container via a `#rom` method.
+      #
+      # ```ruby
+      # class MyOperation < Dry::Operation
+      #   include Dry::Operation::Extensions::ROM
+      #
+      #   attr_reader :rom
+      #
+      #   def initialize(rom:)
+      #     @rom = rom
+      #   end
+      #
+      #   def call(input)
+      #     attrs = step validate(input)
+      #     user = transaction do
+      #       new_user = step persist(attrs)
+      #       step assign_initial_role(new_user)
+      #       new_user
+      #     end
+      #     step notify(user)
+      #     user
+      #   end
+      #
+      #   # ...
+      # end
+      # ```
+      #
+      # By default, the `:default` gateway will be used. You can change this
+      # when including the extension:
+      #
+      # ```ruby
+      # include Dry::Operation::Extensions::ROM[gateway: :my_gateway]
+      # ```
+      #
+      # Or you can change it at runtime:
+      #
+      # ```ruby
+      # user = transaction(gateway: :my_gateway) do
+      #  # ...
+      # end
+      # ```
+      #
+      # @see https://rom-rb.org
+      module ROM
+        DEFAULT_GATEWAY = :default
+
+        # @!method transaction(gateway: DEFAULT_GATEWAY, &steps)
+        #  Wrap the given steps in a rom transaction.
+        #
+        #  If any of the steps returns a `Dry::Monads::Result::Failure`, the
+        #  transaction will be rolled back and `:halt` will be thrown with the
+        #  failure as its value.
+        #
+        #  @yieldreturn [Object] the result of the block
+        #  @raise [Dry::Operation::ExtensionError] if the including
+        #    class doesn't define a `#rom` method.
+        #  @see Dry::Operation#steps
+
+        def self.included(klass)
+          klass.include(self[])
+        end
+
+        # Include the extension providing a custom gateway
+        #
+        # @param gateway [Symbol] the rom gateway to use
+        def self.[](gateway: DEFAULT_GATEWAY)
+          Builder.new(gateway: gateway)
+        end
+
+        # @api private
+        class Builder < Module
+          def initialize(gateway:)
+            super()
+            @gateway = gateway
+          end
+
+          def included(klass)
+            class_exec(@gateway) do |default_gateway|
+              klass.define_method(:transaction) do |gateway: default_gateway, &steps|
+                raise Dry::Operation::ExtensionError, <<~MSG unless respond_to?(:rom)
+                  When using the ROM extension, you need to define a #rom method \
+                  that returns the ROM container
+                MSG
+
+                rom.gateways[gateway].transaction do |t|
+                  intercepting_failure(-> { raise t.rollback! }) do
+                    steps.()
+                  end
+                end
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

spec/integration/extensions/rom_spec.rb

@@ -0,0 +1,76 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+RSpec.describe Dry::Operation::Extensions::ROM do
+  include Dry::Monads[:result]
+
+  let(:rom) do
+    ROM.container(:sql, "sqlite:memory") do |config|
+      config.default.create_table(:foo) do
+        column :bar, :string
+      end
+
+      config.relation(:foo)
+    end
+  end
+
+  let(:base) do
+    Class.new(Dry::Operation) do
+      include Dry::Operation::Extensions::ROM
+
+      attr_reader :rom
+
+      def initialize(rom:)
+        @rom = rom
+        super()
+      end
+    end
+  end
+
+  it "rolls transaction back on failure" do
+    instance = Class.new(base) do
+      def call
+        transaction do
+          step create_record
+          step failure
+        end
+      end
+
+      def create_record
+        Success(rom.relations[:foo].command(:create).(bar: "bar"))
+      end
+
+      def failure
+        Failure(:failure)
+      end
+    end.new(rom: rom)
+
+    instance.()
+
+    expect(rom.relations[:foo].count).to be(0)
+  end
+
+  it "acts transparently for the regular flow" do
+    instance = Class.new(base) do
+      def call
+        transaction do
+          step create_record
+          step count_records
+        end
+      end
+
+      def create_record
+        Success(rom.relations[:foo].command(:create).(bar: "bar"))
+      end
+
+      def count_records
+        Success(rom.relations[:foo].count)
+      end
+    end.new(rom: rom)
+
+    expect(
+      instance.()
+    ).to eql(Success(1))
+  end
+end

spec/unit/extensions/rom_spec.rb

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+RSpec.describe Dry::Operation::Extensions::ROM do
+  describe "#transaction" do
+    it "raises a meaningful error when #rom method is not implemented" do
+      instance = Class.new.include(Dry::Operation::Extensions::ROM).new
+
+      expect { instance.transaction {} }.to raise_error(
+        Dry::Operation::ExtensionError,
+        /you need to define a #rom method/
+      )
+    end
+  end
+end

spec/unit/operation_spec.rb

@@ -55,4 +55,38 @@ def foo(value)
       }.to throw_symbol(:halt, failure)
     end
   end
+
+  describe "#intercepting_failure" do
+    it "forwards the block's output when it's not a failure" do
+      expect(
+        described_class.new.intercepting_failure(-> {}) { :foo }
+      ).to be(:foo)
+    end
+
+    it "doesn't call the handler when the block doesn't return a failure" do
+      called = false
+
+      catch(:halt) {
+        described_class.new.intercepting_failure(-> { called = true }) { :foo }
+      }
+
+      expect(called).to be(false)
+    end
+
+    it "throws :halt with the result when the block returns a failure" do
+      expect {
+        described_class.new.intercepting_failure(-> {}) { Failure(:foo) }
+      }.to throw_symbol(:halt, Failure(:foo))
+    end
+
+    it "calls the handler when the block returns a failure" do
+      called = false
+
+      catch(:halt) {
+        described_class.new.intercepting_failure(-> { called = true }) { Failure(:foo) }
+      }
+
+      expect(called).to be(true)
+    end
+  end
 end