Add a basic batching mechanism

Inspired by AWS Event Source Mapping, it support batch on size or time_window

Author: ramsayleung

Procfile.dev

@@ -1,2 +1,3 @@
 web: bin/rails server
 css: bin/rails tailwindcss:watch
+job: bin/jobs

app/jobs/classifier_trainer_job.rb

@@ -3,7 +3,10 @@ class ClassifierTrainerJob < ApplicationJob
   queue_as :training
 
   def perform(trained_messages)
-    Rails.logger.info "Retrain all the classifiers for public"
+    Rails.logger.info "Retraining classifiers."
+    trained_message_ids = trained_messages.map { |data| data["id"] }
+    trained_messages = TrainedMessage.where(id: trained_message_ids)
+
     # Separate messages by their training target
     user_name_messages     = trained_messages.select(&:user_name?)
     message_content_messages = trained_messages.select(&:message_content?)

app/jobs/process_pending_batches_job.rb

@@ -0,0 +1,7 @@
+class ProcessPendingBatchesJob < ApplicationJob
+  queue_as :batching
+
+  def perform
+    BatchProcessor.process_pending_batches
+  end
+end

app/models/batch_processor.rb

@@ -0,0 +1,76 @@
+class BatchProcessor < ApplicationRecord
+  validates :batch_key, presence: true, uniqueness: true
+  validates :job_class, presence: true
+  validates :pending_count, presence: true, numericality: { greater_than_or_euqal_to: 0 }
+
+  # Generic batching configuration
+  DEFAULT_BATCH_SIZE = 100
+  DEFAULT_BATCH_WINDOW_IN_SECONDS = 30.seconds
+
+  # JSON serialization for SQLite compatibility
+  def shared_args
+    JSON.parse(shared_args_json || "{}")
+  end
+
+  def shared_args=(value)
+    self.shared_args_json = value.to_json
+  end
+
+  def pending_items
+    JSON.parse(pending_items_json || "[]")
+  end
+
+  def pending_items=(value)
+    self.pending_items_json = value.to_json
+  end
+
+  def self.add_to_batch(batch_key, job_class, item_data, shared_args = {}, batch_size: DEFAULT_BATCH_SIZE, batch_window: DEFAULT_BATCH_WINDOW_IN_SECONDS)
+    batch = find_or_create_by(batch_key: batch_key) do |b|
+      b.job_class = job_class
+      b.shared_args = shared_args
+      b.pending_items = []
+      b.pending_count = 0
+      b.batch_size = batch_size
+      b.batch_window_in_seconds = batch_window.to_i
+      b.last_processed_at = Time.current
+    end
+
+    current_items = batch.pending_items
+    current_items << item_data
+    batch.pending_items = current_items
+    batch.pending_count = current_items.size
+    batch.save!
+
+    # Trigger processing ONLY if the batch is full
+    if batch.pending_count >= batch.batch_size
+      process_batch(batch)
+    end
+  end
+
+  def self.process_pending_batches
+    where("pending_count > 0")
+      .find_each do |batch|
+      if batch.last_processed_at < batch.batch_window_in_seconds.seconds.ago
+        process_batch(batch)
+      end
+    end
+  end
+
+  private
+  def self.should_process_batch?(batch)
+    batch.pending_count >= batch.batch_size || batch.last_processed_at < batch.batch_window_in_seconds.seconds.ago
+  end
+
+  def self.process_batch(batch)
+    return if batch.pending_items.blank?
+
+    job_class = batch.job_class.constantize
+    job_class.perform_later(batch.pending_items, **batch.shared_args.symbolize_keys)
+
+    batch.update!(
+      pending_items: [],
+      pending_count: 0,
+      last_processed_at: Time.current
+    )
+  end
+end

app/models/trained_message.rb

@@ -54,10 +54,18 @@ def should_ban_user
       )
     end
   end
+
   def retrain_classifier
     return if untrained?
 
-    ClassifierTrainerJob.perform_later([ self ])
+    BatchProcessor.add_to_batch(
+      "classifier_training_batch_key",
+      "ClassifierTrainerJob",
+      self,                     # pass trainedMessage as item_data
+      {},                        # no shared_args needed
+      batch_size: 100,
+      batch_window: 5.minutes
+    )
   end
 
   private

config/database.yml

@@ -10,16 +10,41 @@ default: &default
   timeout: 5000
 
 development:
-  <<: *default
-  database: storage/development.sqlite3
+  primary:
+    <<: *default
+    database: storage/development.sqlite3
+  cache:
+    <<: *default
+    database: storage/development_cache.sqlite3
+    migrations_paths: db/cache_migrate
+  queue:
+    <<: *default
+    database: storage/development_queue.sqlite3
+    migrations_paths: db/queue_migrate
+  cable:
+    <<: *default
+    database: storage/development_cable.sqlite3
+    migrations_paths: db/cable_migrate
 
 # Warning: The database defined as "test" will be erased and
 # re-generated from your development database when you run "rake".
 # Do not set this db to the same as development or production.
 test:
-  <<: *default
-  database: storage/test.sqlite3
-
+  primary:
+    <<: *default
+    database: storage/test.sqlite3
+  cache:
+    <<: *default
+    database: storage/test_cache.sqlite3
+    migrations_paths: db/cache_migrate
+  queue:
+    <<: *default
+    database: storage/test_queue.sqlite3
+    migrations_paths: db/queue_migrate
+  cable:
+    <<: *default
+    database: storage/test_cable.sqlite3
+    migrations_paths: db/cable_migrate
 
 # Store production database in the storage/ directory, which by default
 # is mounted as a persistent Docker volume in config/deploy.yml.

config/environments/development.rb

@@ -55,6 +55,12 @@
   # Highlight code that enqueued background job in logs.
   config.active_job.verbose_enqueue_logs = true
 
+  # Configure Active Job to use Solid Queue
+  config.active_job.queue_adapter = :solid_queue
+
+  # Configure Solid Queue to use the queue database
+  config.solid_queue.connects_to = { database: { writing: :queue } }
+
   # Raises error for missing translations.
   # config.i18n.raise_on_missing_translations = true
 

config/recurring.yml

@@ -13,3 +13,12 @@ production:
   clear_solid_queue_finished_jobs:
     command: "SolidQueue::Job.clear_finished_in_batches(sleep_between_batches: 0.3)"
     schedule: every hour at minute 12
+
+  process_pending_batches:
+    class: ProcessPendingBatchesJob
+    schedule: every 3 seconds
+
+development:
+  process_pending_batches:
+    class: ProcessPendingBatchesJob
+    schedule: every 3 seconds

db/migrate/20250912031239_create_batch_processors.rb

@@ -0,0 +1,17 @@
+class CreateBatchProcessors < ActiveRecord::Migration[8.0]
+  def change
+    create_table :batch_processors do |t|
+      t.string :batch_key, null: false
+      t.string :job_class, null: false
+      t.text :shared_args_json, default: "{}"
+      t.text :pending_items_json, default: "[]"
+      t.integer :pending_count, default: 0
+      t.integer :batch_size, default: 100
+      t.integer :batch_window_in_seconds, default: 30
+
+      t.timestamps
+    end
+    add_index :batch_processors, :batch_key, unique: true
+    add_index :batch_processors, :updated_at
+  end
+end

db/migrate/20250912081630_add_last_processed_at_to_batch_processors.rb

@@ -0,0 +1,5 @@
+class AddLastProcessedAtToBatchProcessors < ActiveRecord::Migration[8.0]
+  def change
+    add_column :batch_processors, :last_processed_at, :datetime
+  end
+end