Plugins

EasyOp has an opt-in plugin system. Plugins extend the operation lifecycle via the plugin DSL. They are not required automatically — require each plugin file before using it.

Activation

require "easyop/plugins/instrumentation"
require "easyop/plugins/recording"
require "easyop/plugins/async"
require "easyop/plugins/transactional"

# Domain event plugins (opt-in separately):
require "easyop/events/event"
require "easyop/events/bus"
require "easyop/events/bus/memory"
require "easyop/events/registry"
require "easyop/plugins/events"
require "easyop/plugins/event_handlers"

class ApplicationOperation
  include Easyop::Operation

  plugin Easyop::Plugins::Instrumentation
  plugin Easyop::Plugins::Recording,    model: OperationLog
  plugin Easyop::Plugins::Async,        queue: "operations"
  plugin Easyop::Plugins::Transactional
end

# All subclasses inherit these plugins automatically:
class CreateUser < ApplicationOperation
  def call; ctx.user = User.create!(email: ctx.email); end
end

Plugin Execution Order

Plugins wrap _easyop_run via Ruby's prepend. The last installed plugin is the outermost wrapper:

# Installation order: Instrumentation → Recording → Async → Transactional
#
# Execution order (outermost first):
#
#  Transactional (outermost — last installed)
#    Recording::RunWrapper
#      Instrumentation::RunWrapper
#        prepare { before → call → after }  (innermost)
#
# This means:
# - The DB transaction opens first and closes last (wraps everything)
# - Recording measures wall time including transaction overhead
# - The Instrumentation event fires inside the transaction + recording window
# - Async only adds .call_async — it does NOT wrap _easyop_run

Plugin: Instrumentation

Emits an ActiveSupport::Notifications event after every operation call. Requires ActiveSupport (included with Rails).

require "easyop/plugins/instrumentation"

class ApplicationOperation
  include Easyop::Operation
  plugin Easyop::Plugins::Instrumentation
end

Event

Event name: "easyop.operation.call"

Built-in log subscriber

# config/initializers/easyop.rb
Easyop::Plugins::Instrumentation.attach_log_subscriber
# Output: "[EasyOp] Users::Register ok (4.2ms)"
# Output: "[EasyOp] Users::Authenticate FAILED (1.1ms) — Invalid email or password"

Manual subscribe

ActiveSupport::Notifications.subscribe("easyop.operation.call") do |event|
  p = event.payload
  MyAPM.record_span(
    p[:operation],
    success:  p[:success],
    error:    p[:error],
    duration: p[:duration]
  )
end

Plugin: Recording

Persists every operation execution to an ActiveRecord model. Useful for audit trails, debugging, and performance monitoring.

require "easyop/plugins/recording"

class ApplicationOperation
  include Easyop::Operation
  plugin Easyop::Plugins::Recording, model: OperationLog
end

Options

Migration

create_table :operation_logs do |t|
  t.string   :operation_name, null: false
  t.boolean  :success,        null: false
  t.string   :error_message
  t.text     :params_data         # JSON — INPUT keys only; sensitive values replaced with [FILTERED]
  t.float    :duration_ms
  t.datetime :performed_at,   null: false

  # Optional — add when using the record_result DSL to capture output data:
  t.text     :result_data         # JSON — ctx snapshot after the call (computed values included)
end

Flow Tracing (optional)

Add optional columns to reconstruct the full call tree when nested flows run. Missing columns are silently skipped — fully backward-compatible.

# Migration to add flow tracing to an existing table:
add_column :operation_logs, :root_reference_id,     :string
add_column :operation_logs, :reference_id,          :string
add_column :operation_logs, :parent_operation_name, :string
add_column :operation_logs, :parent_reference_id,   :string

add_index :operation_logs, :root_reference_id
add_index :operation_logs, :reference_id, unique: true
add_index :operation_logs, :parent_reference_id

Execution Index (optional)

Add an execution_index integer column to record the 1-based call order among siblings. Root operations store nil; each child's counter resets independently under its own parent — so two children of different parents can both be index 1.

# Migration to add execution order to an existing table:
add_column :operation_logs, :execution_index, :integer

# Composite index — fetch siblings in call order:
add_index :operation_logs, [:parent_reference_id, :execution_index],
          name: 'index_operation_logs_on_parent_ref_and_exec_index'

FullCheckout  (execution_index: nil — root)
  ├─ ValidateCart    (execution_index: 1)
  ├─ ApplyDiscount   (execution_index: 2)
  │   ├─ LookupCode  (execution_index: 1)
  │   └─ DeductAmt   (execution_index: 2)
  └─ CreateOrder     (execution_index: 3)

A nested flow like FullCheckout → AuthAndValidate → AuthenticateUser produces one record per operation, all sharing the same root_reference_id, with parent_* fields linking each operation to its caller.

Easyop::Flow automatically forwards the parent-tracing ctx to child steps. For the flow itself to appear in the tree as the root entry, inherit from your recorded base class and add transactional false (so step-level transactions are not shadowed by an outer one):

class ProcessCheckout < ApplicationOperation
  include Easyop::Flow
  transactional false   # EasyOp handles soft rollback; each step owns its AR transaction

  flow ValidateCart, ChargePayment, CreateOrder
end

# Result in operation_logs:
# ProcessCheckout  root=aaa  ref=bbb  parent=nil         idx=nil
#   ValidateCart   root=aaa  ref=ccc  parent=bbb   idx=1
#   ChargePayment  root=aaa  ref=ddd  parent=bbb   idx=2
#   CreateOrder    root=aaa  ref=eee  parent=bbb   idx=3

# Useful model scopes / methods:
scope :for_tree,    ->(id) { where(root_reference_id: id).order(:performed_at) }
scope :children_of, ->(ref) { where(parent_reference_id: ref).order(:execution_index) }
def root?; parent_reference_id.nil?; end

# Fetch all logs for a given execution tree:
root_log = OperationLog.find_by(operation_name: "FullCheckout", parent_reference_id: nil)
OperationLog.for_tree(root_log.root_reference_id)

# Fetch direct children of a node in call order:
OperationLog.children_of(root_log.reference_id)

Controlling params_data (record_params)

By default, params_data records only the keys that were present in ctx when the operation was called (inputs). Values computed during the call body — like ctx.user = User.create!(...) — are excluded from params_data and should be captured with record_result instead. Use the record_params DSL to further customize or suppress params recording:

# Disable entirely — no params_data written
class Users::Authenticate < ApplicationOperation
  record_params false
end

# Selective keys — only these ctx attrs are recorded
class Tickets::GenerateTickets < ApplicationOperation
  record_params attrs: %i[event_id seat_count]
end

# Block — full control over the extracted hash
class Reports::GeneratePdf < ApplicationOperation
  record_params { |ctx| { report_type: ctx.report_type, page_count: ctx.pages } }
end

# Symbol — delegates to a private method on the instance
class Payments::ChargeCard < ApplicationOperation
  record_params :safe_params
  private
  def safe_params = { user_id: ctx.user.id, amount_cents: ctx.amount_cents }
end

FILTERED_KEYS are always applied to the extracted hash regardless of form. Plugin install-level record_params: accepts the same forms as the DSL.

Result Data (optional)

Add an optional result_data :text column to capture ctx output after the operation runs. Use the record_result DSL — four forms supported.

# Migration:
add_column :operation_logs, :result_data, :text  # stored as JSON

# True form — full ctx snapshot (FILTERED_KEYS applied, internal keys excluded):
class Users::Register < ApplicationOperation
  record_result true
end

# Attrs form — one or more ctx keys:
class PlaceOrder < ApplicationOperation
  record_result attrs: :order_id
end

class ProcessPayment < ApplicationOperation
  record_result attrs: [:charge_id, :amount_cents]
end

# Block form — custom extraction:
class GenerateReport < ApplicationOperation
  record_result { |ctx| { rows: ctx.rows.count, format: ctx.format } }
end

# Symbol form — delegates to a private instance method:
class BuildInvoice < ApplicationOperation
  record_result :build_result
  private
  def build_result = { invoice_id: ctx.invoice.id, total: ctx.total }
end

# Plugin-level default — inherited by all subclasses, overridable per class:
plugin Easyop::Plugins::Recording, model: OperationLog,
       record_result: true
# or: record_result: { attrs: :metadata }
# or: record_result: ->(ctx) { { id: ctx.record_id } }
# or: record_result: :build_result

Class-level record_result overrides the plugin-level default. Missing ctx keys produce nil — no error. ActiveRecord objects → { "id" => 42, "class" => "User" }. Serialization errors are swallowed. The result_data column is silently skipped when absent from the model table — fully backward-compatible.

Filtered keys

Sensitive keys are kept in params_data but their value is replaced with "[FILTERED]", so audit logs show which fields were passed without exposing their values. All filter layers are additive — none replaces the built-in list:

1. Built-in FILTERED_KEYS — always applied: :password, :password_confirmation, :token, :secret, :api_key
2. Global config — Easyop.configure { |c| c.recording_filter_keys = [:api_token, /token/i] }
3. filter_keys: plugin option — plugin Easyop::Plugins::Recording, model: OperationLog, filter_keys: [:stripe_secret]
4. filter_params DSL — per-class, inheritable, stackable: filter_params :card_number, /access.?key/i

class ApplicationOperation < ...
  filter_params :api_token, /secret/i
end

class Payments::ChargeCard < ApplicationOperation
  filter_params :card_number  # stacks on top of parent's list
end

Internal tracing keys (__recording_*) are always fully removed from params_data. ActiveRecord objects in ctx are serialized as { "id" => 42, "class" => "User" }.

Opting out

# Disable recording for this class only:
class Newsletter::SendBroadcast < ApplicationOperation
  recording false
end

# Re-enable in a subclass when parent has recording disabled:
class ImportantOp < BaseWithNoRecording
  recording true
end

# Skip params serialization for high-frequency or sensitive operations:
class Users::TrackPageView < ApplicationOperation
  plugin Easyop::Plugins::Recording, model: OperationLog, record_params: false
end

Recording failures are silently swallowed — a failed database write never breaks the operation.

Plugin: Async

Adds .call_async to any operation class, enqueuing execution as an ActiveJob. Requires ActiveJob (included with Rails).

require "easyop/plugins/async"

class Reports::GeneratePDF < ApplicationOperation
  plugin Easyop::Plugins::Async, queue: "reports"
end

Enqueueing

# Enqueue immediately (default queue):
Reports::GeneratePDF.call_async(report_id: 123)

# Override queue at call time:
Reports::GeneratePDF.call_async(report_id: 123, queue: "low_priority")

# Delay execution:
Reports::GeneratePDF.call_async(report_id: 123, wait: 10.minutes)

# Schedule at a specific time:
Reports::GeneratePDF.call_async(report_id: 123, wait_until: Date.tomorrow.noon)

Scheduling options

Queue DSL

Use the queue class method to declare or override the default queue directly on a class without re-declaring the plugin. Accepts Symbol or String. The setting is inherited by subclasses and can be overridden at any level.

class Weather::BaseOperation < ApplicationOperation
  queue :weather   # all Weather ops use the "weather" queue
end

class Weather::FetchForecast < Weather::BaseOperation
  # inherits queue :weather automatically
end

class Weather::CleanupExpiredDays < Weather::BaseOperation
  queue :low_priority   # override just for this class
end

Weather::FetchForecast._async_default_queue      # => "weather"
Weather::CleanupExpiredDays._async_default_queue # => "low_priority"

Queue inheritance

class BaseOp
  include Easyop::Operation
  plugin Easyop::Plugins::Async, queue: "operations"
end

class ChildOp < BaseOp
  def call; end
end

ChildOp._async_default_queue  # => "operations"  (inherited from parent)

ActiveRecord serialization

# AR objects are serialized as { "__ar_class" => "Order", "__ar_id" => 42 }
# and re-fetched via Order.find(42) in the job:
Orders::SendConfirmation.call_async(order: @order, user: current_user)

# Only pass serializable values:
# String, Integer, Float, Boolean, nil, Hash, Array, ActiveRecord::Base

Job class

# Easyop::Plugins::Async::Job is created lazily on first .call_async call
# This means you can require the plugin before ActiveJob loads
Easyop::Plugins::Async::Job  # => the ActiveJob subclass (after first call_async)

Plugin: Transactional

Wraps every operation call in a database transaction. Supports ActiveRecord and Sequel — adapter is detected automatically.

require "easyop/plugins/transactional"

# On a specific operation:
class TransferFunds < ApplicationOperation
  plugin Easyop::Plugins::Transactional

  def call
    ctx.from_account.debit!(ctx.amount)
    ctx.to_account.credit!(ctx.amount)
    ctx.transaction_id = SecureRandom.uuid
  end
end

# Globally — all subclasses get transactions:
class ApplicationOperation
  include Easyop::Operation
  plugin Easyop::Plugins::Transactional
end

Opting out

class ReadOnlyReport < ApplicationOperation
  transactional false   # no transaction overhead for read-only ops
end

Include style (also works)

class LegacyOp
  include Easyop::Operation
  include Easyop::Plugins::Transactional
end

Lifecycle scope

The transaction wraps the entire prepare chain — before hooks, call, and after hooks all run inside the same transaction. If ctx.fail! is called (raising Ctx::Failure), the transaction rolls back.

With Flow

# Applied per-step: each step gets its own transaction
class ProcessOrder
  include Easyop::Flow
  flow ValidateCart, ChargePayment, CreateOrder  # each step has its own transaction
end

# For a flow-wide transaction, apply it to the Flow class itself:
class ProcessOrder
  include Easyop::Flow
  plugin Easyop::Plugins::Transactional  # entire flow in one transaction

  flow ValidateCart, ChargePayment, CreateOrder
end

Plugin: Events (producer)

Emit domain events after an operation completes. Events fire in an ensure block, so they publish even when call! raises Ctx::Failure. Individual publish failures are swallowed per-declaration and never crash the operation.

require "easyop/events/event"
require "easyop/events/bus"
require "easyop/events/bus/memory"
require "easyop/events/registry"
require "easyop/plugins/events"

class PlaceOrder < ApplicationOperation
  plugin Easyop::Plugins::Events

  emits "order.placed",    on: :success, payload: [:order_id, :total]
  emits "order.failed",    on: :failure, payload: ->(ctx) { { error: ctx.error } }
  emits "order.attempted", on: :always
  emits "vip.order.placed", on: :success, guard: ->(ctx) { ctx.total > 1_000 }

  def call
    ctx.order_id = Order.create!(ctx.slice(:user_id, :items)).id
    ctx.total    = ctx.items.sum { |i| i[:price] }
  end
end

emits options

Event object (Easyop::Events::Event)

event.to_h  # => { name:, payload:, source:, metadata:, timestamp: }

Inheritance

Subclasses inherit all parent emits declarations and can add their own:

class PlaceSubscriptionOrder < PlaceOrder
  emits "subscription.placed", on: :success  # added on top of inherited declarations
end

Per-class bus override

plugin Easyop::Plugins::Events, bus: Easyop::Events::Bus::Memory.new

Plugin: EventHandlers (subscriber)

Register an operation as a handler for domain events. Registration happens at class-load time via Easyop::Events::Registry. Supports exact names and glob patterns.

require "easyop/plugins/event_handlers"

# Sync handler — ctx.event is an Easyop::Events::Event object,
# payload keys are merged directly into ctx:
class SendOrderConfirmation < ApplicationOperation
  plugin Easyop::Plugins::EventHandlers

  on "order.placed"

  def call
    event    = ctx.event        # Easyop::Events::Event
    order_id = ctx.order_id     # from event.payload
    OrderMailer.confirm(order_id).deliver_later
  end
end

# Async handler — requires Plugins::Async on the same class.
# ctx.event_data is a plain Hash (serializable for ActiveJob):
class IndexOrder < ApplicationOperation
  plugin Easyop::Plugins::Async,         queue: "indexing"
  plugin Easyop::Plugins::EventHandlers

  on "order.*",      async: true             # one-segment wildcard
  on "inventory.**", async: true, queue: "low"  # any-depth wildcard + queue override

  def call
    SearchIndex.reindex(ctx.order_id)
  end
end

Pattern syntax

Dispatch context

Events Bus

The bus is the transport layer between producers and handlers. Configure once at boot before any EventHandlers classes are loaded.

# config/initializers/easyop.rb

# Built-in options:
Easyop::Events::Registry.bus = :memory           # default — in-process, thread-safe
Easyop::Events::Registry.bus = :active_support   # ActiveSupport::Notifications

# Custom adapter — subclass Bus::Adapter (preferred) or pass a duck-typed object:
Easyop::Events::Registry.bus = MyRabbitBus.new

# Via configure block (applied when bus is first accessed):
Easyop.configure { |c| c.event_bus = :active_support }

Built-in adapters

Test helpers

# Reset the whole registry between test examples:
Easyop::Events::Registry.reset!

# Memory-bus specific:
bus = Easyop::Events::Registry.bus   # Easyop::Events::Bus::Memory
bus.clear!                            # remove all subscriptions
bus.subscriber_count                  # => Integer

Building a Custom Bus

Subclass Easyop::Events::Bus::Adapter to build a transport-backed bus. It inherits the glob helpers from Bus::Base and adds two protected utilities:

Minimal example — logging decorator

Wrap another bus and add structured logging. No external gems required.

require "easyop/events/bus/adapter"
require "easyop/events/bus/memory"

# Decorator: wraps any inner bus and logs every publish call.
class LoggingBus < Easyop::Events::Bus::Adapter
  def initialize(inner = Easyop::Events::Bus::Memory.new)
    super()
    @inner = inner
  end

  def publish(event)
    logger.info "[bus:publish] #{event.name} source=#{event.source} payload=#{event.payload}"
    @inner.publish(event)
  end

  def subscribe(pattern, &block)
    @inner.subscribe(pattern, &block)
  end

  def unsubscribe(handle)
    @inner.unsubscribe(handle)
  end

  private

  def logger
    defined?(Rails) ? Rails.logger : Logger.new($stdout)
  end
end

# config/initializers/easyop.rb
Easyop::Events::Registry.bus = LoggingBus.new
# Or wrap the ActiveSupport bus:
Easyop::Events::Registry.bus = LoggingBus.new(Easyop::Events::Bus::ActiveSupportNotifications.new)

Full example — RabbitMQ (Bunny gem)

Uses a topic exchange so AMQP routing-key patterns map directly to EasyOp globs. AMQP * matches exactly one dot-separated word (same as EasyOp *); AMQP # matches zero-or-more words (maps from EasyOp **).

require "bunny"
require "json"
require "easyop/events/bus/adapter"

# Drop-in RabbitMQ bus for Easyop::Events.
#
#   Easyop::Events::Registry.bus = RabbitBus.new
#   Easyop::Events::Registry.bus = RabbitBus.new(ENV["AMQP_URL"])
#
class RabbitBus < Easyop::Events::Bus::Adapter
  EXCHANGE_NAME = "easyop.events"

  def initialize(amqp_url = ENV.fetch("AMQP_URL", "amqp://guest:guest@localhost"))
    super()
    @amqp_url  = amqp_url
    @mutex     = Mutex.new
    @handles   = {}   # handle.object_id => { queue:, consumer: }
  end

  # Publish +event+ to the topic exchange with routing_key = event.name.
  # The full event hash is serialised to JSON.
  def publish(event)
    exchange.publish(
      event.to_h.merge(timestamp: event.timestamp.iso8601).to_json,
      routing_key:  event.name,
      content_type: "application/json",
      persistent:   false
    )
  end

  # Bind an exclusive, auto-delete queue to the exchange and start consuming.
  # Returns a handle that can be passed to #unsubscribe.
  def subscribe(pattern, &block)
    amqp_key = _to_amqp_pattern(pattern)
    queue    = channel.queue("", exclusive: true, auto_delete: true)
    queue.bind(exchange, routing_key: amqp_key)

    consumer = queue.subscribe(manual_ack: false) do |_delivery, _props, body|
      data  = JSON.parse(body, symbolize_names: true)
      event = Easyop::Events::Event.new(
                name:      data[:name].to_s,
                payload:   data.fetch(:payload, {}),
                metadata:  data.fetch(:metadata, {}),
                source:    data[:source],
                timestamp: data[:timestamp] ? Time.parse(data[:timestamp].to_s) : Time.now
              )
      _safe_invoke(block, event)
    end

    handle = Object.new  # unique, opaque token
    @mutex.synchronize { @handles[handle.object_id] = { queue: queue, consumer: consumer } }
    handle
  end

  # Cancel the consumer and delete the queue associated with +handle+.
  def unsubscribe(handle)
    @mutex.synchronize do
      entry = @handles.delete(handle.object_id)
      return unless entry
      entry[:consumer].cancel
      entry[:queue].delete
    end
  end

  # Gracefully close the AMQP connection (call in an at_exit hook or Railtie).
  def disconnect
    @mutex.synchronize do
      @connection&.close
      @connection = nil
      @channel    = nil
      @exchange   = nil
      @handles.clear
    end
  end

  private

  # Convert EasyOp glob → AMQP topic routing-key pattern.
  #   "**"  →  "#"  (zero-or-more segments in AMQP)
  #   "*"   →  "*"  (exactly one segment — identical semantics)
  def _to_amqp_pattern(pattern)
    return pattern.source if pattern.is_a?(Regexp)
    pattern.gsub("**", "#")
  end

  def connection
    @connection ||= Bunny.new(@amqp_url, recover_from_connection_close: true).tap(&:start)
  end

  def channel
    @channel ||= connection.create_channel
  end

  def exchange
    @exchange ||= channel.topic(EXCHANGE_NAME, durable: true)
  end
end

# config/initializers/easyop.rb
Easyop::Events::Registry.bus = RabbitBus.new

# Optional: disconnect cleanly on shutdown
at_exit { Easyop::Events::Registry.bus.disconnect }

Easyop::Events::Registry.bus = MyExistingBusObject.new

Building a Custom Plugin

A plugin is any object responding to .install(base_class, **options). Inherit from Easyop::Plugins::Base for a clean interface.

require "easyop/plugins/base"

module TimingPlugin
  def self.install(base, threshold_ms: 500, **_opts)
    base.prepend(RunWrapper)
    base.extend(ClassMethods)
    base.instance_variable_set(:@_timing_threshold_ms, threshold_ms)
  end

  module ClassMethods
    # Per-class DSL: `timing false` to disable
    def timing(enabled)
      @_timing_enabled = enabled
    end

    def _timing_enabled?
      # Use instance_variable_defined? (not defined?) to handle false correctly
      return @_timing_enabled if instance_variable_defined?(:@_timing_enabled)
      superclass.respond_to?(:_timing_enabled?) ? superclass._timing_enabled? : true
    end

    def _timing_threshold_ms
      @_timing_threshold_ms ||
        (superclass.respond_to?(:_timing_threshold_ms) ? superclass._timing_threshold_ms : 500)
    end
  end

  module RunWrapper
    def _easyop_run(ctx, raise_on_failure:)
      return super unless self.class._timing_enabled?

      start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      super.tap do
        ms = (Process.clock_gettime(Process::CLOCK_MONOTONIC) - start) * 1000
        if ms > self.class._timing_threshold_ms
          logger = defined?(Rails) ? Rails.logger : Logger.new($stdout)
          logger.warn "[SLOW] #{self.class.name} took #{ms.round(1)}ms"
        end
      end
    end
  end
end

# Activate:
class ApplicationOperation
  include Easyop::Operation
  plugin TimingPlugin, threshold_ms: 200
end

# Opt out on a specific class:
class Cache::Lookup < ApplicationOperation
  timing false
end

Plugin building conventions

• Prefix all internal instance/class methods with _pluginname_ (e.g. _timing_enabled?) to avoid collisions
• Use instance_variable_defined? (not defined?) when checking instance variables on class objects — correctly handles false values
• Always call super in RunWrapper#_easyop_run and return ctx
• Implement per-class opt-out using the instance_variable_defined? pattern
• Walk the inheritance chain (superclass.respond_to?) to inherit configuration

Easyop::Plugins::Base

Abstract base class. Inherit from it to document intent — it raises NotImplementedError if .install is not overridden.

class MyPlugin < Easyop::Plugins::Base
  def self.install(base, **options)
    # your implementation
  end
end