Add database-backed event processing with Outbox pattern

This commit introduces a database-backed outbox pattern for reliable event
processing as an alternative to direct Kinesis delivery. This enables
transactional event guarantees and better reliability.

Key changes:
- Add Journaled::Outbox::Adapter for database-backed event delivery
- Add Journaled::Outbox::Event model with UUID v7 primary keys
- Add Journaled::Outbox::Worker for processing events from the database
- Add Journaled::Outbox::BatchProcessor for batch delivery to Kinesis
- Add Journaled::Outbox::MetricEmitter for monitoring queue health
- Add generators for creating database tables and UUID v7 support
- Add rake task for running the worker process
- Update KinesisBatchSender to work with database events
- Add comprehensive test coverage for all outbox components

The outbox pattern stores events in PostgreSQL with guaranteed ordering
using UUID v7 timestamps, then processes them asynchronously with
configurable retry logic and detailed metrics emission.

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,127 @@ 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.
+
+**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
+```
+
+**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
+
+# Emitted once per minute with queue statistics
+ActiveSupport::Notifications.subscribe('journaled.worker.queue_metrics') do |name, start, finish, id, payload|
+  Statsd.gauge('journaled.worker.queue.total', payload[:total_count], tags: ["worker:#{payload[:worker_id]}"])
+  Statsd.gauge('journaled.worker.queue.workable', payload[:workable_count], tags: ["worker:#{payload[:worker_id]}"])
+  Statsd.gauge('journaled.worker.queue.erroring', payload[:erroring_count], tags: ["worker:#{payload[:worker_id]}"])
+  Statsd.gauge('journaled.worker.queue.oldest_age_seconds', payload[:oldest_age_seconds], tags: ["worker:#{payload[:worker_id]}"]) if payload[:oldest_age_seconds]
+end
+```
+
+Queue metrics payload includes:
+- `total_count` - Total number of events in the queue (including failed)
+- `workable_count` - Events ready to be processed (not failed)
+- `erroring_count` - Events with errors but not yet marked as permanently failed
+- `oldest_non_failed_timestamp` - Timestamp of the oldest non-failed event (extracted from UUID v7)
+- `oldest_age_seconds` - Age in seconds of the oldest non-failed event
+
+Note: Metrics are collected in a background thread to avoid blocking the main worker loop.
+
+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/event.rb

@@ -18,7 +18,7 @@ def event_type
   end
 
   def created_at
-    @created_at ||= Time.zone.now
+    @created_at ||= Time.current
   end
 
   # Event metadata and configuration (not serialized)

app/models/journaled/outbox/event.rb

@@ -0,0 +1,76 @@
+# 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'
+
+      skip_audit_log
+
+      attribute :event_data, :json
+
+      validates :event_type, :event_data, :partition_key, :stream_name, presence: true
+
+      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
+      #
+      # @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')
+          .to_a
+      end
+
+      # Requeue a failed event for processing
+      #
+      # Clears failure information so the event can be retried.
+      #
+      # @return [Boolean] Whether the requeue was successful
+      def requeue!
+        update!(
+          failed_at: nil,
+          failure_reason: nil,
+        )
+      end
+
+      # Extract timestamp from UUID v7 id
+      #
+      # UUID v7 embeds a timestamp in the first 48 bits (milliseconds since Unix epoch)
+      #
+      # @return [Time] The timestamp embedded in the UUID
+      def self.timestamp_from_uuid(uuid)
+        # Remove dashes and take first 12 hex characters (48 bits)
+        hex_timestamp = uuid.to_s.delete('-')[0, 12]
+        # Convert from hex to milliseconds since epoch
+        milliseconds = hex_timestamp.to_i(16)
+        # Convert to Time object
+        Time.zone.at(milliseconds / 1000.0)
+      end
+
+      # Get the oldest non-failed event's timestamp
+      #
+      # @return [Time, nil] The timestamp of the oldest event, or nil if no events exist
+      def self.oldest_non_failed_timestamp
+        oldest = ready_to_process.order(:id).limit(1).pick(:id)
+        return nil unless oldest
+
+        timestamp_from_uuid(oldest)
+      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,30 @@
+===============================================================================
+
+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
+
+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,19 @@
+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
+
+      t.text :failure_reason
+
+      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