Add database-backed event processing with Outbox pattern

Introduce an optional Outbox pattern implementation as an alternative to
ActiveJob-based delivery. Events can now be stored in database tables
and processed by background workers, providing better reliability and
observability for event delivery.

This implementation includes:

Models & Storage:
- Outbox::Base model with common validations
- Outbox::Event model with distributed locking for concurrent workers
- Outbox::DeadLetter model for failed events with requeue support
- Migration generator for creating required database tables

Processing Infrastructure:
- Outbox::Worker daemon with graceful shutdown and signal handling
- Outbox::BatchProcessor with retry logic and dead letter queue
- Outbox::Adapter implementing the DeliveryAdapter interface
- Rake task for running workers (rake journaled:work)

Configuration & Documentation:
- New configuration options: outbox_base_class_name, worker_batch_size,
  worker_poll_interval, worker_lock_timeout, worker_max_attempts
- Comprehensive README documentation with setup instructions
- Examples for monitoring and DLQ handling

Testing:
- Complete test coverage for all Outbox components
- Integration tests for end-to-end event processing
- Test database schema with Outbox tables

The default behavior remains unchanged (ActiveJobAdapter). Users can
opt into database-backed processing by configuring the Outbox::Adapter.

Author: effron

README.md

@@ -22,9 +22,11 @@ durable, eventually consistent record that discrete events happened.
 
 ## Installation
 
-1. If you haven't already,
-[configure ActiveJob](https://guides.rubyonrails.org/active_job_basics.html)
-to use one of the following queue adapters:
+1. **Configure a queue adapter** (only required if using the default ActiveJob delivery adapter):
+
+    If you haven't already,
+    [configure ActiveJob](https://guides.rubyonrails.org/active_job_basics.html)
+    to use one of the following queue adapters:
 
     - `:delayed_job` (via `delayed_job_active_record`)
     - `:que`
@@ -52,6 +54,8 @@ to use one of the following queue adapters:
 
     This configuration isn't necessary for applications running Rails 8+.
 
+    **Note:** If you plan to use the [Database-Backed Event Processing](#database-backed-event-processing-optional) (Outbox adapter), you can skip this step entirely, as the Outbox adapter does not use ActiveJob.
+
 2. To integrate Journaled into your application, simply include the gem in your
 app's Gemfile.
 
@@ -129,6 +133,37 @@ Journaling provides a number of different configuation options that can be set i
 
   The number of seconds before the :http_handler should timeout while waiting for a HTTP response.
 
+#### `Journaled.delivery_adapter` (default: `Journaled::DeliveryAdapters::ActiveJobAdapter`)
+
+  Determines how events are delivered to Kinesis. Two options are available:
+
+  - **`Journaled::DeliveryAdapters::ActiveJobAdapter`** (default) - Enqueues events to ActiveJob. Requires a DB-backed queue adapter (see Installation).
+
+  - **`Journaled::Outbox::Adapter`** - Stores events in a database table and processes them via separate worker daemons. See [Database-Backed Event Processing](#database-backed-event-processing-optional) for setup instructions.
+
+  Example:
+  ```ruby
+  # Use the database-backed Outbox adapter
+  Journaled.delivery_adapter = Journaled::Outbox::Adapter
+  ```
+
+#### `Journaled.outbox_base_class_name` (default: `'ActiveRecord::Base'`)
+
+  **Only relevant when using `Journaled::Outbox::Adapter`.**
+
+  Specifies which ActiveRecord base class the Outbox event storage model (`Journaled::Outbox::Event`) should use for its database connection. This is useful for multi-database setups where you want to store events in a separate database.
+
+  Example:
+  ```ruby
+  # Store outbox events in a separate database
+  class EventsRecord < ActiveRecord::Base
+    self.abstract_class = true
+    connects_to database: { writing: :events, reading: :events }
+  end
+
+  Journaled.outbox_base_class_name = 'EventsRecord'
+  ```
+
 #### ActiveJob `set` options
 
 Both model-level directives accept additional options to be passed into ActiveJob's `set` method:
@@ -143,6 +178,117 @@ has_audit_log enqueue_with: { priority: 30 }
 # Or for custom journaling:
 journal_attributes :email, enqueue_with: { priority: 20, queue: 'journaled' }
 ```
+##### Database-Backed Event Processing (Optional)
+
+Journaled includes a built-in database-backed delivery adapter with horizontally scalable workers. This is useful if you want to:
+
+- **Decouple from Delayed Job/ActiveJob**: Process events independently of your existing job queue
+- **Horizontal scaling**: Run multiple worker daemons across different servers/containers
+- **Batch API efficiency**: Use Kinesis `put_records` (batch) API instead of `put_record` (single), reducing API calls by up to 500x
+- **Custom retry logic**: Fine-tune retry behavior and failed event handling
+- **Visibility**: Events are stored in database tables where you can inspect, query, and monitor them
+
+**Setup:**
+
+This feature requires creating database tables and is completely optional. Existing users are unaffected.
+
+1. **Generate migrations:**
+
+```bash
+rails generate journaled:database_events
+rails db:migrate
+```
+
+This creates a table for storing events:
+- `journaled_outbox_events` - Queue of events to be processed (includes `failed_at` column for tracking failures)
+
+2. **Configure to use the database adapter:**
+
+```ruby
+# config/initializers/journaled.rb
+
+# Use the database-backed Outbox adapter instead of ActiveJob
+Journaled.delivery_adapter = Journaled::Outbox::Adapter
+
+# Optional: Customize worker behavior (these are the defaults)
+Journaled.worker_batch_size = 500        # Max events per Kinesis batch (Kinesis API limit)
+Journaled.worker_poll_interval = 5       # Seconds between polls
+Journaled.worker_max_attempts = 3        # Retries before marking as failed
+```
+
+**Note:** When using the Outbox adapter, you do **not** need to configure an ActiveJob queue adapter (skip step 1 of Installation). The Outbox adapter uses the `journaled_outbox_events` table for event storage and its own worker daemons for processing, making it independent of ActiveJob. Transactional batching still works seamlessly with the Outbox adapter.
+
+3. **Start worker daemon(s):**
+
+```bash
+bundle exec rake journaled_worker:work
+```
+
+4. **Monitoring:**
+
+The system emits `ActiveSupport::Notifications` events:
+
+```ruby
+# config/initializers/journaled.rb
+
+# Emitted for every batch processed (regardless of outcome)
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_process') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.batches', tags: ["worker:#{payload[:worker_id]}"])
+end
+
+# Emitted for successfully sent events
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_sent') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.events_sent', payload[:event_count], tags: ["worker:#{payload[:worker_id]}"])
+end
+
+# Emitted for permanently failed events (marked as failed in database)
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_failed') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.events_failed', payload[:event_count], tags: ["worker:#{payload[:worker_id]}"])
+end
+
+# Emitted for transiently failed events (will be retried)
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_errored') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.events_errored', payload[:event_count], tags: ["worker:#{payload[:worker_id]}"])
+end
+```
+
+5. **Failed Events:**
+
+Inspect and requeue failed events:
+
+```ruby
+# Find failed events
+Journaled::Outbox::Event.failed.where(stream_name: 'my_stream')
+Journaled::Outbox::Event.failed_since(1.hour.ago)
+
+# Requeue a failed event (clears failure info and resets attempts)
+failed_event = Journaled::Outbox::Event.failed.find(123)
+failed_event.requeue!
+```
+
+6. **Production Deployment:**
+
+Deploy workers as separate services (systemd, Docker, Kubernetes, etc.):
+
+```yaml
+# Example Kubernetes deployment
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: journaled-worker
+spec:
+  replicas: 3
+  template:
+    spec:
+      containers:
+      - name: worker
+        image: myapp:latest
+        command: ["bundle", "exec", "rake", "journaled_worker:work"]
+        env:
+        - name: AWS_DEFAULT_REGION
+          value: "us-east-1"
+        # ... other env vars
+```
 
 ### Attribution
 

app/models/journaled/outbox/event.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module Journaled
+  module Outbox
+    # ActiveRecord model for database-backed event processing
+    #
+    # This model is only used when the Outbox::Adapter is configured.
+    # Events are stored in the database and processed by worker daemons.
+    #
+    # Successfully delivered events are deleted immediately.
+    # Failed events are marked with failed_at and can be queried or requeued.
+    class Event < Journaled.outbox_base_class_name.constantize
+      self.table_name = 'journaled_outbox_events'
+
+      attribute :event_data, :json
+
+      validates :event_type, :event_data, :partition_key, :stream_name, presence: true
+      validates :attempts, numericality: { only_integer: true, greater_than_or_equal_to: 0 }
+
+      scope :ready_to_process, -> {
+        where(failed_at: nil)
+          .order(:id)
+      }
+
+      scope :failed, -> { where.not(failed_at: nil) }
+      scope :failed_since, ->(time) { where('failed_at >= ?', time) }
+
+      # Fetch a batch of events for processing using SELECT FOR UPDATE SKIP LOCKED
+      #
+      # @return [Array<Journaled::Outbox::Event>] Events locked for processing
+      def self.fetch_batch_for_update
+        ready_to_process
+          .limit(Journaled.worker_batch_size)
+          .lock('FOR UPDATE SKIP LOCKED')
+          .to_a
+      end
+
+      # Requeue a failed event for processing
+      #
+      # Clears failure information and resets attempts so the event can be retried.
+      #
+      # @return [Boolean] Whether the requeue was successful
+      def requeue!
+        update!(
+          failed_at: nil,
+          attempts: 0,
+          last_error: nil,
+        )
+      end
+    end
+  end
+end

lib/generators/journaled/database_events/database_events_generator.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require 'rails/generators'
+require 'rails/generators/active_record'
+
+module Journaled
+  module Generators
+    class DatabaseEventsGenerator < Rails::Generators::Base
+      include ActiveRecord::Generators::Migration
+
+      source_root File.expand_path('templates', __dir__)
+
+      desc "Generates migration for journaled database-backed event processing"
+
+      def install_uuid_generate_v7_migration
+        migration_template(
+          "install_uuid_generate_v7.rb.erb",
+          "db/migrate/install_uuid_generate_v7.rb",
+          migration_version:,
+        )
+      end
+
+      def create_journaled_events_migration
+        migration_template(
+          "create_journaled_events.rb.erb",
+          "db/migrate/create_journaled_events.rb",
+          migration_version:,
+        )
+      end
+
+      def show_readme
+        readme "README" if behavior == :invoke
+      end
+
+      private
+
+      def migration_version
+        "[#{ActiveRecord::VERSION::MAJOR}.#{ActiveRecord::VERSION::MINOR}]"
+      end
+    end
+  end
+end

lib/generators/journaled/database_events/templates/README

@@ -0,0 +1,31 @@
+===============================================================================
+
+Database-backed event processing migrations have been generated!
+
+Next steps:
+
+1. Run migrations:
+
+   rails db:migrate
+
+2. Configure Journaled to use the outbox adapter in an initializer:
+
+   # config/initializers/journaled.rb
+   Journaled.delivery_adapter = Journaled::Outbox::Adapter
+
+   # Optional configuration:
+   Journaled.worker_batch_size = 500  # default: 500 (Kinesis max)
+   Journaled.worker_poll_interval = 5  # default: 5 seconds
+   Journaled.worker_max_attempts = 3  # default: 3
+
+3. Start the worker daemon:
+
+   bundle exec rake journaled_worker:work
+
+4. (Optional) For horizontal scaling, run multiple workers on different
+   servers/containers. They will coordinate using database locks.
+
+For more information, see the README:
+https://github.com/Betterment/journaled#database-backed-event-processing-optional
+
+===============================================================================

lib/generators/journaled/database_events/templates/create_journaled_events.rb.erb

@@ -0,0 +1,22 @@
+class CreateJournaledEvents < ActiveRecord::Migration<%= migration_version %>
+  def change
+    # UUID v7 primary key (auto-generated by database using uuid_generate_v7())
+    create_table :journaled_outbox_events, id: :uuid, default: -> { "uuid_generate_v7()" } do |t|
+      # Event identification and data
+      t.string :event_type, null: false
+      t.jsonb :event_data, null: false
+      t.string :partition_key, null: false
+      t.string :stream_name, null: false
+
+      # Retry logic
+      t.integer :attempts, default: 0, null: false
+      t.text :last_error
+
+      # Failure tracking (instead of separate DLQ table)
+      t.datetime :failed_at
+    end
+
+    # Index for querying failed events
+    add_index :journaled_outbox_events, :failed_at
+  end
+end

lib/generators/journaled/database_events/templates/install_uuid_generate_v7.rb.erb

@@ -0,0 +1,25 @@
+class InstallUuidGenerateV7 < ActiveRecord::Migration<%= migration_version %>
+  def up
+    # Enable pgcrypto extension for gen_random_bytes()
+    enable_extension 'pgcrypto'
+
+    # Install UUID v7 generation function
+    # Source: https://github.com/Betterment/postgresql-uuid-generate-v7
+    execute <<-SQL
+      CREATE OR REPLACE FUNCTION uuid_generate_v7()
+      RETURNS uuid
+      LANGUAGE plpgsql
+      PARALLEL SAFE
+      AS $$
+      DECLARE
+        unix_time_ms CONSTANT bytea NOT NULL DEFAULT substring(int8send((extract(epoch FROM clock_timestamp()) * 1000)::bigint) from 3);
+        buffer bytea NOT NULL DEFAULT unix_time_ms || gen_random_bytes(10);
+      BEGIN
+        buffer = set_byte(buffer, 6, (b'0111' || get_byte(buffer, 6)::bit(4))::bit(8)::int);
+        buffer = set_byte(buffer, 8, (b'10' || get_byte(buffer, 8)::bit(6))::bit(8)::int);
+        RETURN encode(buffer, 'hex');
+      END
+      $$;
+    SQL
+  end
+end

lib/journaled.rb

@@ -10,8 +10,11 @@
 require 'journaled/connection'
 require 'journaled/delivery_adapter'
 require 'journaled/delivery_adapters/active_job_adapter'
+require 'journaled/outbox/adapter'
 require 'journaled/kinesis_client_factory'
 require 'journaled/kinesis_batch_sender'
+require 'journaled/outbox/batch_processor'
+require 'journaled/outbox/worker'
 
 module Journaled
   SUPPORTED_QUEUE_ADAPTERS = %w(delayed delayed_job good_job que).freeze
@@ -22,9 +25,15 @@ module Journaled
   mattr_accessor(:http_open_timeout) { 2 }
   mattr_accessor(:http_read_timeout) { 60 }
   mattr_accessor(:job_base_class_name) { 'ActiveJob::Base' }
+  mattr_accessor(:outbox_base_class_name) { 'ActiveRecord::Base' }
   mattr_accessor(:delivery_adapter) { Journaled::DeliveryAdapters::ActiveJobAdapter }
   mattr_writer(:transactional_batching_enabled) { true }
 
+  # Worker configuration (for database-backed event processing)
+  mattr_accessor(:worker_batch_size) { 500 } # Kinesis PutRecords API max
+  mattr_accessor(:worker_poll_interval) { 5 } # seconds
+  mattr_accessor(:worker_max_attempts) { 3 }
+
   def self.transactional_batching_enabled?
     Thread.current[:journaled_transactional_batching_enabled] || @@transactional_batching_enabled
   end