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

effron · effron · commit 21bf3db80991 · 2025-11-14T11:38:22.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/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/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/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
@@ -31,79 +31,53 @@
     end
   end
 
-  describe '.timestamp_from_uuid' do
-    it 'extracts timestamp from UUID v7' do
-      # UUID v7 with known timestamp: 2024-01-01 00:00:00 UTC (1704067200000 milliseconds)
-      # In hex: 0x18CC251F400 = 018cc251f400
-      uuid_with_known_timestamp = '018cc251-f400-7000-8000-000000000000'
-
-      timestamp = described_class.timestamp_from_uuid(uuid_with_known_timestamp)
-
-      expect(timestamp).to be_within(1.second).of(Time.zone.parse('2024-01-01 00:00:00 UTC'))
-    end
-
-    it 'handles UUID without dashes' do
-      uuid_without_dashes = '018cc251f40070008000000000000000'
-
-      timestamp = described_class.timestamp_from_uuid(uuid_without_dashes)
-
-      expect(timestamp).to be_within(1.second).of(Time.zone.parse('2024-01-01 00:00:00 UTC'))
-    end
-  end
-
   describe '.oldest_non_failed_timestamp' do
     it 'returns nil when no events exist' do
       expect(described_class.oldest_non_failed_timestamp).to be_nil
     end
 
     it 'returns timestamp of oldest non-failed event' do
-      # Create events with different timestamps embedded in their UUIDs
+      # Create events - the database will generate UUIDs, which are ordered chronologically
       older_event = described_class.create!(
-        id: '018cc251-f400-7000-8000-000000000000', # 2024-01-01
         event_type: 'test_event',
         event_data: { test: 'data' },
         partition_key: 'key1',
         stream_name: 'test_stream',
       )
 
       described_class.create!(
-        id: '018cc252-0000-7000-8000-000000000000', # slightly newer
         event_type: 'test_event',
         event_data: { test: 'data' },
         partition_key: 'key2',
         stream_name: 'test_stream',
       )
 
       timestamp = described_class.oldest_non_failed_timestamp
-      expected_timestamp = described_class.timestamp_from_uuid(older_event.id)
 
-      expect(timestamp).to be_within(1.second).of(expected_timestamp)
+      expect(timestamp).to be_within(1.second).of(older_event.created_at)
     end
 
     it 'ignores failed events' do
-      # Create a failed event with older timestamp
+      # Create a failed event
       described_class.create!(
-        id: '018cc251-f400-7000-8000-000000000000', # 2024-01-01
         event_type: 'test_event',
         event_data: { test: 'data' },
         partition_key: 'key1',
         stream_name: 'test_stream',
         failed_at: Time.current,
       )
 
-      # Create a non-failed event with newer timestamp
+      # Create a non-failed event
       newer_event = described_class.create!(
-        id: '018cc252-0000-7000-8000-000000000000', # slightly newer
         event_type: 'test_event',
         event_data: { test: 'data' },
         partition_key: 'key2',
         stream_name: 'test_stream',
       )
 
       timestamp = described_class.oldest_non_failed_timestamp
-      expected_timestamp = described_class.timestamp_from_uuid(newer_event.id)
 
-      expect(timestamp).to be_within(1.second).of(expected_timestamp)
+      expect(timestamp).to be_within(1.second).of(newer_event.created_at)
     end
   end
 end
