simplify migration generator, add created_at to outbox events, bulk insert events

effron · effron · commit 1c4d40d2f95f · 2025-11-14T15:52:13.000-05:00
diff --git a/README.md b/README.md
@@ -186,10 +186,10 @@ Journaled includes a built-in Outbox-style delivery adapter with horizontally sc
 
 This feature requires creating database tables and is completely optional. Existing users are unaffected.
 
-1. **Generate migrations:**
+1. **Install migrations:**
 
 ```bash
-rails generate journaled:database_events
+rake journaled:install:migrations
 rails db:migrate
 ```
 
@@ -269,7 +269,6 @@ 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)
diff --git a/app/models/journaled/event.rb b/app/models/journaled/event.rb
@@ -18,7 +18,7 @@ def event_type
   end
 
   def created_at
-    @created_at ||= Time.current
+    @created_at ||= Time.zone.now
   end
 
   # Event metadata and configuration (not serialized)
diff --git a/app/models/journaled/outbox/event.rb b/app/models/journaled/outbox/event.rb
@@ -24,7 +24,6 @@ class Event < Journaled.outbox_base_class_name.constantize
       }
 
       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
       #
@@ -48,28 +47,11 @@ def requeue!
         )
       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)
+        ready_to_process.order(:id).limit(1).pick(:created_at)
       end
     end
   end
diff --git a/db/migrate/1_install_uuid_generate_v7.rb b/db/migrate/1_install_uuid_generate_v7.rb
@@ -1,11 +1,13 @@
-class InstallUuidGenerateV7 < ActiveRecord::Migration<%= migration_version %>
+# frozen_string_literal: true
+
+class InstallUuidGenerateV7 < ActiveRecord::Migration[7.2]
   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
+    execute <<-SQL.squish
       CREATE OR REPLACE FUNCTION uuid_generate_v7()
       RETURNS uuid
       LANGUAGE plpgsql
diff --git a/db/migrate/2_create_journaled_events.rb b/db/migrate/2_create_journaled_events.rb
@@ -1,4 +1,6 @@
-class CreateJournaledEvents < ActiveRecord::Migration<%= migration_version %>
+# frozen_string_literal: true
+
+class CreateJournaledEvents < ActiveRecord::Migration[7.2]
   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|
@@ -11,6 +13,7 @@ def change
       t.text :failure_reason
 
       t.datetime :failed_at
+      t.datetime :created_at, null: false, default: -> { "clock_timestamp()" }
     end
 
     # Index for querying failed events
diff --git a/journaled.gemspec b/journaled.gemspec
@@ -17,7 +17,7 @@ Gem::Specification.new do |s|
   s.license     = "MIT"
   s.metadata['rubygems_mfa_required'] = 'true'
 
-  s.files = Dir["{app,config,lib,journaled_schemas}/**/*", "LICENSE", "Rakefile", "README.md"]
+  s.files = Dir["{app,config,db,lib,journaled_schemas}/**/*", "LICENSE", "Rakefile", "README.md"]
 
   s.required_ruby_version = ">= 3.2"
 
diff --git a/lib/generators/journaled/database_events/database_events_generator.rb b/lib/generators/journaled/database_events/database_events_generator.rb
diff --git a/lib/generators/journaled/database_events/templates/README b/lib/generators/journaled/database_events/templates/README
diff --git a/lib/journaled.rb b/lib/journaled.rb
@@ -31,9 +31,8 @@ module Journaled
   mattr_writer(:transactional_batching_enabled) { true }
 
   # Worker configuration (for Outbox-style event processing)
-  mattr_accessor(:worker_batch_size) { 500 } # Kinesis PutRecords API max
+  mattr_accessor(:worker_batch_size) { 500 }
   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
diff --git a/lib/journaled/engine.rb b/lib/journaled/engine.rb
@@ -2,6 +2,8 @@
 
 module Journaled
   class Engine < ::Rails::Engine
+    engine_name 'journaled'
+
     config.after_initialize do
       ActiveSupport.on_load(:active_job) do
         Journaled.delivery_adapter.validate_configuration! unless Journaled.development_or_test?
diff --git a/lib/journaled/outbox/adapter.rb b/lib/journaled/outbox/adapter.rb
@@ -23,40 +23,44 @@ class TableNotFoundError < StandardError; end
       def self.deliver(events:, **)
         check_table_exists!
 
-        events.each do |event|
+        records = events.map do |event|
           # Exclude the application-level id - the database will generate its own using uuid_generate_v7()
           event_data = event.journaled_attributes.except(:id)
 
-          Event.create!(
+          {
             event_type: event.journaled_attributes[:event_type],
             event_data:,
             partition_key: event.journaled_partition_key,
             stream_name: event.journaled_stream_name,
-          )
+          }
         end
+
+        # rubocop:disable Rails/SkipsModelValidations
+        Event.insert_all(records) if records.any?
+        # rubocop:enable Rails/SkipsModelValidations
       end
 
       # Check if the required database table exists
       #
       # @raise [TableNotFoundError] if the table doesn't exist
       def self.check_table_exists!
-        return if table_exists_cache
+        return if @table_exists
 
         unless Event.table_exists?
           raise TableNotFoundError, <<~ERROR
             Journaled::Outbox::Adapter requires the 'journaled_outbox_events' table.
 
             To create the required tables, run:
 
-              rails generate journaled:database_events
+              rake journaled:install:migrations
               rails db:migrate
 
             For more information, see the README:
             https://github.com/Betterment/journaled#outbox-style-event-processing-optional
           ERROR
         end
 
-        self.table_exists_cache = true
+        @table_exists = true
       end
 
       # Returns the database connection to use for transactional batching
@@ -87,12 +91,6 @@ def self.validate_configuration!
           and row-level locking for distributed worker coordination. Other databases are not supported.
         ERROR
       end
-
-      # Cache table existence check to avoid repeated queries
-      class << self
-        attr_accessor :table_exists_cache
-      end
-      self.table_exists_cache = false
     end
   end
 end
diff --git a/lib/journaled/outbox/batch_processor.rb b/lib/journaled/outbox/batch_processor.rb
@@ -34,7 +34,7 @@ def initialize
       def process_batch
         ActiveRecord::Base.transaction do
           events = Event.fetch_batch_for_update
-          Rails.logger.info("Processing batch of #{events.count} events")
+          Rails.logger.info("[journaled] Processing batch of #{events.count} events")
 
           result = batch_sender.send_batch(events)
 
@@ -45,7 +45,7 @@ def process_batch
           mark_events_as_failed(result[:failed]) if result[:failed].any?
 
           Rails.logger.info(
-            "Batch complete: #{result[:succeeded].count} succeeded, " \
+            "[journaled] Batch complete: #{result[:succeeded].count} succeeded, " \
             "#{result[:failed].count} marked as failed",
           )
 
diff --git a/lib/journaled/outbox/metric_emitter.rb b/lib/journaled/outbox/metric_emitter.rb
@@ -30,14 +30,13 @@ def emit_queue_metrics
         emit_metric('journaled.worker.queue_total_count', value: metrics[:total_count])
         emit_metric('journaled.worker.queue_workable_count', value: metrics[:workable_count])
         emit_metric('journaled.worker.queue_erroring_count', value: metrics[:erroring_count])
-
-        emit_metric('journaled.worker.queue_oldest_age_seconds', value: metrics[:oldest_age_seconds] || 0)
+        emit_metric('journaled.worker.queue_oldest_age_seconds', value: metrics[:oldest_age_seconds])
 
         Rails.logger.info(
           "Queue metrics: total=#{metrics[:total_count]}, " \
           "workable=#{metrics[:workable_count]}, " \
           "erroring=#{metrics[:erroring_count]}, " \
-          "oldest_age=#{metrics[:oldest_age_seconds]&.round(2)}s",
+          "oldest_age=#{metrics[:oldest_age_seconds].round(2)}s",
         )
       end
 
@@ -71,13 +70,12 @@ def calculate_queue_metrics
 
         # Oldest non-failed event timestamp
         oldest_timestamp = Event.oldest_non_failed_timestamp
-        oldest_age_seconds = oldest_timestamp ? Time.current - oldest_timestamp : nil
+        oldest_age_seconds = oldest_timestamp ? Time.current - oldest_timestamp : 0
 
         {
           total_count:,
           workable_count:,
           erroring_count:,
-          oldest_non_failed_timestamp: oldest_timestamp,
           oldest_age_seconds:,
         }
       end
diff --git a/lib/journaled/outbox/worker.rb b/lib/journaled/outbox/worker.rb
@@ -107,7 +107,7 @@ def setup_signal_handlers
 
       # Emit metrics if the interval has elapsed
       def emit_metrics_if_needed
-        return unless Time.current - last_metrics_emission >= 60 # 60 seconds = 1 minute
+        return unless Time.current - last_metrics_emission >= 60
 
         # Collect and emit metrics in a background thread to avoid blocking the main loop
         Thread.new do
diff --git a/spec/dummy/db/schema.rb b/spec/dummy/db/schema.rb
@@ -52,6 +52,7 @@
     t.string "stream_name", null: false
     t.text "failure_reason"
     t.datetime "failed_at"
+    t.datetime "created_at", null: false, default: -> { "clock_timestamp()" }
     t.index ["failed_at"], name: "index_journaled_outbox_events_on_failed_at"
   end
 end
diff --git a/spec/lib/journaled/outbox/adapter_spec.rb b/spec/lib/journaled/outbox/adapter_spec.rb
@@ -21,7 +21,7 @@
 
   before do
     # Reset table existence cache
-    described_class.table_exists_cache = false
+    described_class.instance_variable_set(:@table_exists, false)
   end
 
   describe '.validate_configuration!' do
@@ -127,7 +127,7 @@
           described_class.deliver(events:, enqueue_opts:)
         }.to raise_error(
           Journaled::Outbox::Adapter::TableNotFoundError,
-          /rails generate journaled:database_events/,
+          /rake journaled:install:migrations/,
         )
       end
     end
diff --git a/spec/models/journaled/outbox/event_spec.rb b/spec/models/journaled/outbox/event_spec.rb
