process items from the firehose in the worker

app/import_worker.rb

···
 require_relative 'init'
 require_relative 'import_manager'
 require_relative 'models/import'
+require_relative 'post_downloader'
 require_relative 'reports/basic_report'
 
 class ImportWorker
···
   def run
     @user_threads = []
 
+    @firehose_thread = Thread.new { process_firehose_items }
+    @downloader = PostDownloader.new
+
     loop do
       @user_threads.delete_if { |t| !t.alive? }
 
-      ActiveRecord::Base.transaction do
-        if user = User.with_unfinished_imports.where.not(id: @user_threads.map(&:user_id)).first
-          collections = user.imports.unfinished.map(&:collection)
-          thread = UserThread.new(user, collections, @verbose)
-          @user_threads << thread
-        end
+      if user = User.with_unfinished_imports.where.not(id: @user_threads.map(&:user_id)).first
+        collections = user.imports.unfinished.map(&:collection)
+        thread = UserThread.new(user, collections, @verbose)
+        @user_threads << thread
       end
 
       # possible future enhancement: use LISTEN/UNLISTEN/NOTIFY and wait_for_notify
       sleep 5
+    end
+  end
+
+  def process_firehose_items
+    loop do
+      items = [Like, Repost, Pin, Quote]
+        .map { |type| type.in_queue(:firehose).order('time').limit(25) }
+        .flatten
+        .sort_by(&:time)
+        .first(25)
+
+      if items.length > 0
+        @downloader.process_items(items)
+      else
+        sleep 5
+      end
     end
   end
 end

app/post_downloader.rb

···
 
       @report&.update(queue: { length: queue.length })
 
+      process_items(items)
+    end
+  end
+
+  def process_items(items)
+    #
       existing_posts = Post.where(rkey: items.map { |x| AT_URI(x.post_uri).rkey }).to_a
 
       items.dup.each do |item|
···
         end
       end
 
-      next if items.empty?
+      return if items.empty?
 
       begin
         response = @sky.get_request('app.bsky.feed.getPosts', { uris: items.map(&:post_uri).uniq })
···
       rescue StandardError => e
         puts "Error in PostDownloader: #{e.class}: #{e}"
       end
-    end
+    #
   end
 
   def save_post(post_uri, record)