added worker for background imports

app/firehose_client.rb

···
 require 'skyfall'
 
 require_relative 'init'
+require_relative 'models/import_job'
 require_relative 'models/post'
 require_relative 'models/subscription'
 require_relative 'models/user'
···
       @replaying = true
       @last_update = Time.now
 
-      @timer ||= EM::PeriodicTimer.new(20) do
+      @live_check_timer ||= EM::PeriodicTimer.new(20) do
         now = Time.now
         diff = now - @last_update
 
···
           log "Timer: last update #{sprintf('%.1f', diff)}s ago"
         end
       end
+
+      @jobs_timer ||= EM::PeriodicTimer.new(3) do
+        ImportJob.all.each do |job|
+          @active_users[job.user.did] = job.user
+          job.create_imports
+          job.destroy
+        end
+      end
     }
 
     @sky.on_disconnect {
···
     if msg.status == :deleted
       if user = User.find_by(did: msg.repo)
         user.destroy
+        @active_users.delete_if { |u| u.id == user.id }
       end
     end
   end
···
   end
 
   def inspect
-    vars = instance_variables - [:@timer]
+    vars = instance_variables - [:@jobs_timer, :@live_check_timer]
     values = vars.map { |v| "#{v}=#{instance_variable_get(v).inspect}" }.join(", ")
     "#<#{self.class}:0x#{object_id} #{values}>"
   end

app/import_worker.rb

···
+require 'active_record'
+
+require_relative 'init'
+require_relative 'import_manager'
+require_relative 'models/import'
+
+class ImportWorker
+  class UserThread < Thread
+    def initialize(user, collections)
+      @user = user
+      super { run(collections) }
+    end
+
+    def user_id
+      @user.id
+    end
+
+    def run(collections)
+      import = ImportManager.new(@user)
+      import.start(collections, false)
+    end
+  end
+
+  def run
+    @user_threads = []
+
+    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)
+          @user_threads << thread
+        end
+      end
+
+      # possible future enhancement: use LISTEN/UNLISTEN/NOTIFY and wait_for_notify
+      sleep 5
+    end
+  end
+end

app/models/import.rb

···
 
   validates_inclusion_of :collection, in: %w(likes reposts posts)
   validates_uniqueness_of :collection, scope: :user_id
+
+  scope :unfinished, -> { where('(started_from IS NOT NULL) OR (last_completed IS NULL)') }
 end

app/models/import_job.rb

···
+require 'active_record'
+
+require_relative 'user'
+
+class ImportJob < ActiveRecord::Base
+  belongs_to :user
+
+  def create_imports
+    %w(likes reposts posts).each do |group|
+      self.user.imports.find_or_create_by!(collection: group)
+    end
+  end
+end

app/models/user.rb

···
 require 'active_record'
 
 require_relative 'import'
+require_relative 'import_job'
 require_relative 'like'
 require_relative 'quote'
 require_relative 'pin'
···
 
   has_many :posts
   has_many :imports, dependent: :delete_all
+  has_one :import_job, dependent: :delete
 
   before_destroy :delete_posts_cascading
 
···
 
   def self.active
     self.joins(:imports).distinct
+  end
+
+  def self.with_unfinished_imports
+    self.joins(:imports).merge(Import.unfinished).distinct
+  end
+
+  def active?
+    imports.exists?
   end
 
   def all_pending_items

bin/worker

···
+#!/usr/bin/env ruby
+
+require 'bundler/setup'
+require_relative '../app/import_worker'
+
+$stdout.sync = true
+
+if ENV['ARLOG'] == '1'
+  ActiveRecord::Base.logger = Logger.new(STDOUT)
+else
+  ActiveRecord::Base.logger = nil
+end
+
+worker = ImportWorker.new
+
+trap("SIGINT") {
+  puts "Stopping..."
+  exit
+}
+
+trap("SIGTERM") {
+  puts "Shutting down the service..."
+  exit
+}
+
+worker.run

db/migrate/20250920182018_add_import_jobs.rb

···
+class AddImportJobs < ActiveRecord::Migration[7.2]
+  def change
+    create_table :import_jobs do |t|
+      t.integer "user_id", null: false
+    end
+
+    add_index :import_jobs, :user_id, unique: true
+  end
+end

db/schema.rb

···
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.2].define(version: 2025_09_18_024627) do
+ActiveRecord::Schema[7.2].define(version: 2025_09_20_182018) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
+
+  create_table "import_jobs", force: :cascade do |t|
+    t.integer "user_id", null: false
+    t.index ["user_id"], name: "index_import_jobs_on_user_id", unique: true
+  end
 
   create_table "imports", force: :cascade do |t|
     t.integer "user_id", null: false

lib/tasks/import.rake

···
 require_relative '../../app/models/user'
 require_relative '../../app/post_downloader'
 
+task :enqueue_user do
+  unless ENV['DID']
+    raise "Required DID parameter missing"
+  end
+
+  user = User.find_or_create_by!(did: ENV['DID'])
+
+  if user.import_job
+    puts "Import for #{user.did} is already scheduled."
+  elsif user.active?
+    puts "Import for #{user.did} has already started."
+  else
+    user.create_import_job!
+    puts "Import for #{user.did} scheduled ✓"
+  end
+end
+
 task :import_user do
   unless ENV['DID']
     raise "Required DID parameter missing"