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 models (`Journaled::Outbox::Event` and `Journaled::Outbox::DeadLetter`) should use for their 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,106 @@ 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 dead letter queue 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 two tables:
+- `journaled_events` - Queue of events to be processed
+- `journaled_dead_letter_events` - Failed events after max retries
+
+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_lock_timeout = 300      # Seconds before stale lock can be reclaimed
+Journaled.worker_max_attempts = 3        # Retries before moving to DLQ
+```
+
+**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_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
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_sent') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.events_sent', payload[:succeeded])
+end
+
+ActiveSupport::Notifications.subscribe('journaled.worker.batch_failed') do |name, start, finish, id, payload|
+  Statsd.increment('journaled.worker.events_failed', payload[:failed])
+end
+```
+
+5. **Dead Letter Queue:**
+
+Inspect and requeue failed events:
+
+```ruby
+# Find failed events
+Journaled::Outbox::DeadLetter.where(stream_name: 'my_stream')
+Journaled::Outbox::DeadLetter.failed_since(1.hour.ago)
+
+# Requeue an event (creates new pending event, removes from DLQ)
+dlq_event = Journaled::Outbox::DeadLetter.find(123)
+dlq_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/base.rb

@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+module Journaled
+  module Outbox
+    # Provides common validations and scopes for event storage models.
+    class Base < Journaled.outbox_base_class_name.constantize
+      self.abstract_class = true
+
+      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 }
+    end
+  end
+end

app/models/journaled/outbox/dead_letter.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Journaled
+  module Outbox
+    # ActiveRecord model for failed events that have been moved to the dead letter queue
+    #
+    # Events end up here after reaching max retry attempts or encountering permanent failures.
+    # These can be inspected manually and potentially requeued for processing.
+    class DeadLetter < Base
+      self.table_name = 'journaled_dead_letter_events'
+
+      validates :failure_reason, :failed_at, presence: true
+
+      scope :failed_since, ->(time) { where('failed_at >= ?', time) }
+
+      # Requeue this event for processing
+      #
+      # Creates a new Event with the same attributes and removes this DLQ entry
+      #
+      # @return [Journaled::Outbox::Event] The newly created event
+      def requeue!
+        transaction do
+          new_event = Event.create!(
+            event_type:,
+            event_data:,
+            partition_key:,
+            stream_name:,
+            attempts: 0,
+          )
+
+          destroy!
+
+          new_event
+        end
+      end
+    end
+  end
+end

app/models/journaled/outbox/event.rb

@@ -0,0 +1,129 @@
+# 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.
+    class Event < Base
+      self.table_name = 'journaled_events'
+
+      scope :ready_to_process, -> {
+        where('locked_at IS NULL OR locked_at < ?', Time.current - Journaled.worker_lock_timeout.seconds)
+          .order(:created_at)
+      }
+
+      # Claim a batch of events for processing
+      #
+      # Uses SELECT FOR UPDATE SKIP LOCKED for distributed locking across multiple workers
+      #
+      # @param worker_id [String] Unique identifier for this worker (e.g., "hostname-pid")
+      # @return [Array<Journaled::Outbox::Event>] Claimed events
+      def self.claim_batch!(worker_id:)
+        if connection.adapter_name == 'PostgreSQL'
+          claim_batch_postgresql!(worker_id:)
+        else
+          claim_batch_generic!(worker_id:)
+        end
+      end
+
+      # Optimized PostgreSQL implementation using CTE with UPDATE ... FROM
+      def self.claim_batch_postgresql!(worker_id:)
+        timeout = Time.current - Journaled.worker_lock_timeout.seconds
+        batch_size = Journaled.worker_batch_size
+        locked_at = Time.current
+
+        sql = <<-SQL.squish
+          WITH locked_events AS (
+            SELECT id
+            FROM journaled_events
+            WHERE locked_at IS NULL OR locked_at < ?
+            ORDER BY created_at
+            LIMIT ?
+            FOR UPDATE SKIP LOCKED
+          )
+          UPDATE journaled_events
+          SET locked_at = ?,
+              locked_by = ?,
+              attempts = attempts + 1
+          FROM locked_events
+          WHERE journaled_events.id = locked_events.id
+          RETURNING journaled_events.*
+        SQL
+
+        find_by_sql([sql, timeout, batch_size, locked_at, worker_id])
+      end
+
+      # Generic implementation for other databases (SQLite, etc.)
+      def self.claim_batch_generic!(worker_id:)
+        transaction do
+          events = ready_to_process
+            .limit(Journaled.worker_batch_size)
+            .lock('FOR UPDATE SKIP LOCKED')
+            .to_a
+
+          event_ids = events.map(&:id)
+          locked_at = Time.current
+
+          # Use RETURNING to get updated records in one query
+          sql = <<-SQL.squish
+            UPDATE #{table_name}
+            SET locked_at = ?,
+                locked_by = ?,
+                attempts = attempts + 1
+            WHERE id IN (?)
+            RETURNING *
+          SQL
+
+          find_by_sql([sql, locked_at, worker_id, event_ids])
+        end
+      end
+
+      # Mark event as successfully delivered by deleting it
+      def mark_delivered!
+        destroy!
+      end
+
+      # Release lock on event (for transient failures, allowing retry)
+      def release_lock!(error: nil)
+        update!(
+          locked_at: nil,
+          locked_by: nil,
+          last_error: error&.message,
+        )
+      end
+
+      # Move event to dead letter queue (for permanent failures)
+      #
+      # This happens in a transaction to ensure the event is never lost.
+      # The event is moved to the DLQ and then deleted in the same transaction.
+      #
+      # @param error [Exception] The error that caused the permanent failure
+      def move_to_dlq!(error:)
+        transaction do
+          DeadLetter.create!(
+            event_type:,
+            event_data:,
+            partition_key:,
+            stream_name:,
+            attempts:,
+            failure_reason: "#{error.class}: #{error.message}",
+            failed_at: Time.current,
+            originally_created_at: created_at,
+          )
+
+          # Delete the event so it won't be processed again
+          destroy!
+        end
+      end
+
+      # Check if event has reached maximum retry attempts
+      def max_attempts_reached?
+        attempts >= Journaled.worker_max_attempts
+      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 migrations for journaled database-backed event processing"
+
+      def create_journaled_events_migration
+        migration_template(
+          "create_journaled_events.rb.erb",
+          "db/migrate/create_journaled_events.rb",
+          migration_version:,
+        )
+      end
+
+      def create_journaled_dead_letter_events_migration
+        migration_template(
+          "create_journaled_dead_letter_events.rb.erb",
+          "db/migrate/create_journaled_dead_letter_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