Add extension for ROM transactions

We add a `Dry::Operation::Extensions::ROM` module that, when included,
gives access to a `#transaction` method. This method wraps the yielded
steps in a ROM [1] transaction, rolling back in case one of them returns
a failure.

We lean on a new `Dry::Operation#intercepting_failure` method, which
allows running a callback before the failure is re-thrown again to be
managed by the wrapping `#steps` call. Besides providing clarity, this
method will be reused by future extensions.

The extension expects the including class to define a `#rom` method
giving access to the ROM container.

```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
```

The extension uses the `:default` gateway by default, but it can be
changed both at include time with `include
Dry::Operation::Extensions::ROM[gateway: :my_gateway]`, and at runtime
with `#transaction(gateway: :my_gateway)`.

This commit also establishes the dry-operation's convention for database
transactions. Instead of wrapping the whole flow, we require the user to
be conscious of the transaction boundaries (not including, e.g.,
external requests or notifications). That encourages using individual
operations when thinking about composition instead of the whole flow.

[1] - https://rom-rb.org

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 expected interface for an extension is not implemented
+    class InterfaceNotImplementedError < ::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::InterfaceNotImplementedError] 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::InterfaceNotImplementedError, <<~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::InterfaceNotImplementedError,
+        /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