gem "sbmt-outbox"

transaction do
  some_record.save!

  result = Sbmt::Outbox::CreateOutboxItem.call(
    MyOutboxItem,
    event_key: some_record.id,
    attributes: {
      payload: some_record.generate_payload,
      options: {
        key: some_record.id, # optional, may be used when producing to a Kafka topic
        headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
      }
    }
  )

  raise result.failure unless result.success?
end

transaction do
  some_record.save!
  another_record.save!

  result = Sbmt::Outbox::CreateOutboxBatch.call(
    MyOutboxItem,
    batch_attributes: [
      {
        event_key: some_record.id,
        payload: some_record.generate_payload,
        options: {
          key: some_record.id, # optional, may be used when producing to a Kafka topic
          headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
        }
      },
      {
        event_key: another_record.id,
        payload: another_record.generate_payload,
        options: {
          key: another_record.id, # optional, may be used when producing to a Kafka topic
          headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
        }
      }
    ]
  )

  raise result.failure unless result.success?
end

# frozen_string_literal: true

require_relative "config/environment"

# Comment out this line if you don't want to use a metrics exporter
Yabeda::Prometheus::Exporter.start_metrics_server!

# config/initializers/outbox.rb

Rails.application.config.outbox.tap do |config|
  config.redis = {url: ENV.fetch("REDIS_URL")} # Redis is used as a coordinator service
  config.paths << Rails.root.join("config/outbox.yml").to_s # optional; configuration file paths, deep merged at the application start, useful with Rails engines

  # optional (worker v2: default)
  config.poller = ActiveSupport::OrderedOptions.new.tap do |pc|
    # max parallel threads (per box-item, globally)
    pc.concurrency = 6
    # max threads count (per worker process)
    pc.threads_count = 1
    # maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
    pc.general_timeout = 60
    # poll buffer consists of regular items (errors_count = 0, i.e. without any processing errors) and retryable items (errors_count > 0)
    # max poll buffer size = regular_items_batch_size + retryable_items_batch_size
    pc.regular_items_batch_size = 200
    pc.retryable_items_batch_size = 100

    # poll tactic: default is optimal for most cases: rate limit + redis job-queue size threshold
    # poll tactic: aggressive is for high-intensity data: without rate limits + redis job-queue size threshold
    # poll tactic: low-priority is for low-intensity data: rate limits + redis job-queue size threshold + + redis job-queue lag threshold
    pc.tactic = "default"
    # number of batches that one thread will process per rate interval
    pc.rate_limit = 60
    # rate interval in seconds
    pc.rate_interval = 60
    # mix / max redis job queue thresholds per box-item for default / aggressive / low-priority poll tactics
    pc.min_queue_size = 10
    pc.max_queue_size = 100
    # min redis job queue time lag threshold per box-item for low-priority poll tactic (in seconds)
    pc.min_queue_timelag = 5
    # throttling delay for default / aggressive / low-priority poll tactics (in seconds)
    pc.queue_delay = 0.1
  end

  # optional (worker v2: default)
  config.processor = ActiveSupport::OrderedOptions.new.tap do |pc|
    # max threads count (per worker process)
    pc.threads_count = 4
    # maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
    pc.general_timeout = 120
    # BRPOP delay (in seconds) for polling redis job queue per box-item
    pc.brpop_delay = 2
  end

  # optional (worker v1: DEPRECATED)
  config.process_items.tap do |x|
    # maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
    x.general_timeout = 180
    # maximum batch processing time, after which the processing of the batch will be aborted in the current thread,
    # and the next thread that picks up the batch will start processing from the same place
    x.cutoff_timeout = 60
    # batch size
    x.batch_size = 200
  end

  # optional (worker v1: DEPRECATED)
  config.worker.tap do |worker|
    # number of batches that one thread will process per rate interval
    worker.rate_limit = 10
    # rate interval in seconds
    worker.rate_interval = 60
  end
end

create_table :my_outbox_items do |t|
  t.uuid :uuid, null: false
  t.string :event_name, null: false # optional, use it when you have several events per one outbox table
  t.string :event_key, null: false
  t.integer :bucket, null: false
  t.integer :status, null: false, default: 0
  t.jsonb :options
  t.binary :payload, null: false # when using mysql the column type should be mediumblob
  t.integer :errors_count, null: false, default: 0
  t.text :error_log
  t.timestamp :processed_at
  t.timestamps
end

add_index :my_outbox_items, :uuid, unique: true
add_index :my_outbox_items, [:status, :id, :bucket], algorithm: :concurrently, include: [:errors_count]
add_index :my_outbox_items, [:event_name, :event_key, :id]
add_index :my_outbox_items, :created_at

# app/models/my_outbox_item.rb
class MyOutboxItem < Sbmt::Outbox::OutboxItem
  validates :event_name, presence: true # optional
end

# app/services/import_order.rb
class ProduceMessage
  def initialize(topic:)
    @topic = topic
  end

  def call(outbox_item, payload)
    # send message to topic
    true # mark message as processed
  end
end

# app/models/my_inbox_item.rb
class MyInboxItem < Sbmt::Outbox::InboxItem
end

# app/services/import_order.rb
class ImportOrder
  def initialize(source:)
    @source = source
  end

  def call(outbox_item, payload)
    # some work to create order in the database
    true # mark message as processed
  end
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.item_process_middlewares.push(
    'MyFirstItemMiddleware', # goes first
    'MySecondItemMiddleware' # goes second
  )
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.create_item_middlewares.push(
    'MyCreateItemMiddleware'
  )
  config.create_batch_middlewares.push(
    'MyCreateBatchMiddleware'
  )
end

# my_create_item_middleware.rb
class MyCreateItemMiddleware
  def call(item_class, item_attributes)
    # your code
    yield
    # your code
  end
end

# my_create_batch_middleware.rb
class MyCreateBatchMiddleware
  def call(item_class, batch_attributes)
    # your code
    yield
    # your code
  end
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.batch_process_middlewares.push(
    'MyBatchMiddleware'
  )
end

# my_batch_middleware.rb
class MyBatchMiddleware
  def call(job)
    # your code
    yield
    # your code
  end
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.item_process_middlewares.push(
    'MyItemMiddleware'
  )
end

# my_create_item_middleware.rb
class MyItemMiddleware
  def call(item)
    # your code
    yield
    # your code
  end
end

Rails.application.routes.draw do
  mount Sbmt::Outbox::Engine => "/outbox-ui"
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.cdn_url = "https://some-cdn-url"
end

# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
  config.ui.serve_local = true
end