generalize code to allow different types of imports

app/importer.rb

···
 require 'minisky'
 
 require_relative 'at_uri'
-require_relative 'models/import'
-require_relative 'models/like'
 require_relative 'models/post'
 require_relative 'models/user'
 
 class Importer
-  attr_accessor :like_queue, :report
+  attr_accessor :item_queue, :report
 
   def initialize(user_did)
     @did = DID.new(user_did)
     @user = User.find_or_create_by!(did: user_did)
 
     @uid_cache = { user_did => @user.id }
+    @imported_count = 0
   end
 
-  def run_import(time_limit = nil)
-    @minisky = Minisky.new(@did.document.pds_host, nil)
+  def importer_name
+    self.class.name
+  end
 
-    import_likes(time_limit)
+  def collection
+    importer_name.gsub(/Importer$/, '').downcase
   end
 
-  def import_likes(requested_time_limit = nil)
-    import = @user.import || @user.create_import!
+  def run_import(requested_time_limit = nil)
+    @minisky = Minisky.new(@did.document.pds_host, nil)
+    @import = @user.imports.find_by(collection: collection) || @user.imports.create!(collection: collection)
 
-    params = { repo: @did, collection: 'app.bsky.feed.like', limit: 100 }
-
-    if import.cursor
-      params[:cursor] = import.cursor
-    else
-      import.update!(started_from: Time.now) unless requested_time_limit
+    if @import.cursor.nil?
+      @import.update!(started_from: Time.now) unless requested_time_limit
     end
 
-    count = 0
-    time_limit = requested_time_limit || import.last_completed
+    @time_limit = requested_time_limit || @import.last_completed
+    puts "Fetching until: #{@time_limit}" if @time_limit
 
-    puts "Fetching until: #{time_limit}" if time_limit
+    import_items
 
-    loop do
-      response = @minisky.get_request('com.atproto.repo.listRecords', params)
+    @import.update!(last_completed: @import.started_from) unless requested_time_limit
+    @import.update!(cursor: nil, started_from: nil)
+    @report&.update(importers: { importer_name => { :finished => true }})
+  end
 
-      records = response['records']
-      cursor = response['cursor']
+  def import_items
+    raise NotImplementedError
+  end
 
-      count += records.length
-      @report&.update(importer: { imported_likes: count })
-      @report&.update(importer: { oldest_date: Time.parse(records.last['value']['createdAt']) }) unless records.empty?
+  def create_item_for_post(uri)
+    post_uri = AT_URI(uri)
+    return if post_uri.collection != 'app.bsky.feed.post'
 
-      process_likes(records)
-      params[:cursor] = cursor
+    post = Post.find_by(
+      user_id: user_id_for_did(post_uri.repo),
+      rkey: post_uri.rkey
+    )
 
-      import.update!(cursor: cursor)
+    if post
+      yield({ post: post })
+    else
+      item_stub = yield({ post_uri: post_uri })
 
-      break if !cursor
-      break if time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < time_limit }
+      if @item_queue
+        @item_queue.push(item_stub)
+        @report&.update(queue: { length: @item_queue.length })
+      end
     end
-
-    import.update!(cursor: nil, started_from: nil)
-    import.update!(last_completed: import.started_from) unless requested_time_limit
-
-    @report&.update(importer: { finished: true })
   end
 
-  def process_likes(likes)
-    likes.each do |record|
-      begin
-        like_rkey = AT_URI(record['uri']).rkey
-        next if @user.likes.where(rkey: like_rkey).exists?
-
-        like_time = Time.parse(record['value']['createdAt'])
-
-        post_uri = AT_URI(record['value']['subject']['uri'])
-        next if post_uri.collection != 'app.bsky.feed.post'
-
-        post_did = post_uri.repo
-
-        if @uid_cache[post_did].nil?
-          post_author = User.find_or_create_by!(did: post_did)
-          @uid_cache[post_did] = post_author.id
-        end
-
-        post = Post.find_by(user_id: @uid_cache[post_did], rkey: post_uri.rkey)
-
-        if post
-          @user.likes.create!(rkey: like_rkey, time: like_time, post: post)
-        else
-          like_stub = @user.likes.create!(rkey: like_rkey, time: like_time, post_uri: post_uri)
-
-          if @like_queue
-            @like_queue.push(like_stub)
-            @report&.update(queue: { length: @like_queue.length })
-          end
-        end
-      rescue StandardError => e
-        puts "Error in Importer#process_likes: #{record['uri']}: #{e}"
-      end
-    end
+  def user_id_for_did(did)
+    @uid_cache[did] ||= User.find_or_create_by!(did: did)
   end
 end

app/importers/likes_importer.rb

···
+require 'time'
+
+require_relative '../at_uri'
+require_relative '../importer'
+
+class LikesImporter < Importer
+  def import_items
+    params = { repo: @did, collection: 'app.bsky.feed.like', limit: 100 }
+    params[:cursor] = @import.cursor if @import.cursor
+
+    loop do
+      response = @minisky.get_request('com.atproto.repo.listRecords', params)
+
+      records = response['records']
+      cursor = response['cursor']
+
+      @imported_count += records.length
+      @report&.update(importers: { importer_name => { :imported_items => @imported_count }})
+      @report&.update(importers: { importer_name => { :oldest_date => Time.parse(records.last['value']['createdAt']) }}) unless records.empty?
+
+      records.each do |record|
+        begin
+          like_rkey = AT_URI(record['uri']).rkey
+          next if @user.likes.where(rkey: like_rkey).exists?
+
+          like_time = Time.parse(record['value']['createdAt'])
+          post_uri = record['value']['subject']['uri']
+
+          create_item_for_post(post_uri) do |args|
+            @user.likes.create!(args.merge(rkey: like_rkey, time: like_time))
+          end
+        rescue StandardError => e
+          puts "Error in LikesImporter: #{record['uri']}: #{e}"
+        end
+      end
+
+      params[:cursor] = cursor
+      @import.update!(cursor: cursor)
+
+      break if !cursor
+      break if @time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < @time_limit }
+    end
+  end
+end

app/like_queue.rb app/item_queue.rb

···
-class LikeQueue
+class ItemQueue
   BATCH_SIZE = 25
 
-  def initialize(likes = [])
+  def initialize(items = [])
     @mutex = Mutex.new
-    @queue = likes
+    @queue = items
   end
 
-  def push(like)
+  def push(item)
     @mutex.synchronize {
-      @queue << like
+      @queue << item
     }
   end
 

app/models/import.rb

···
 
 class Import < ActiveRecord::Base
   belongs_to :user
+
+  validates_inclusion_of :collection, in: %w(likes)
+  validates_uniqueness_of :collection, scope: :user_id
 end

app/models/user.rb

···
 
   has_many :posts
   has_many :likes, foreign_key: 'actor_id'
-  has_one :import
+  has_many :imports
 end

db/migrate/20250903215014_add_import_collections.rb

···
+class AddImportCollections < ActiveRecord::Migration[7.2]
+  def change
+    add_column :imports, :collection, :string, limit: 20, null: true
+
+    reversible do |dir|
+      dir.up do
+        execute "UPDATE imports SET collection = 'likes'"
+      end
+    end
+
+    change_column_null :imports, :collection, false
+
+    remove_index :imports, :user_id, unique: true
+    add_index :imports, [:user_id, :collection], unique: true
+
+    reversible do |dir|
+      dir.down do
+        execute "DELETE FROM imports WHERE collection != 'likes'"
+      end
+    end
+  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_08_31_210930) do
+ActiveRecord::Schema[7.2].define(version: 2025_09_03_215014) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
 
···
     t.string "cursor"
     t.datetime "started_from"
     t.datetime "last_completed"
-    t.index ["user_id"], name: "index_imports_on_user_id", unique: true
+    t.string "collection", limit: 20, null: false
+    t.index ["user_id", "collection"], name: "index_imports_on_user_id_and_collection", unique: true
   end
 
   create_table "likes", force: :cascade do |t|

lib/tasks/import.rake

···
-require_relative '../../app/importer'
-require_relative '../../app/like_queue'
+require_relative '../../app/importers/likes_importer'
+require_relative '../../app/item_queue'
 require_relative '../../app/post_downloader'
 
 class ImportReport
···
   end
 
   def update(data)
-    data.each do |k, v|
-      @data[k] ||= {}
-      @data[k].update(v)
+    deep_merge(@data, data)
+    render
+  end
+
+  def deep_merge(target, updates)
+    updates.each do |k, v|
+      if v.is_a?(Hash)
+        target[k] ||= {}
+        deep_merge(target[k], v)
+      else
+        target[k] = v
+      end
     end
-
-    render
   end
 
   def render
     print " " * 80 + "\r"
     puts "Elapsed time: #{(Time.now - @start).to_i} s"
 
-    print " " * 80 + "\r"
-    puts "Importer: imported likes = #{@data.dig(:importer, :imported_likes) || 0} (until: #{@data.dig(:importer, :oldest_date)})" +
-      "#{" (DONE)" if @data.dig(:importer, :finished)}"
+    importers = @data[:importers] || {}
+
+    importers.each do |name, data|
+      print " " * 80 + "\r"
+      puts "#{name}: imported items = #{data[:imported_items] || 0} (until: #{data[:oldest_date]}) #{" (DONE)" if data[:finished]}"
+    end
 
     print " " * 80 + "\r"
     puts "Downloader: imported posts = #{@data.dig(:downloader, :downloaded_posts) || 0} (until: #{@data.dig(:downloader, :oldest_date)})"
···
     print " " * 80 + "\r"
     puts "Queue size: #{@data.dig(:queue, :length) || 0}"
 
-    print "\e[4A"
+    print "\e[#{3 + importers.length}A"
   end
 end
 
···
     raise "Required USER parameter missing"
   end
 
-  queue = LikeQueue.new(Like.where(post: nil).to_a)
+  case ENV['COLLECTION']
+  when 'likes'
+    queue = ItemQueue.new(Like.where(post: nil).to_a)
+    importer = LikesImporter.new(ENV['USER'])
+  when nil
+    raise "Required COLLECTION parameter missing"
+  else
+    raise "Invalid collection: #{ENV['COLLECTION']}"
+  end    
+
   report = ImportReport.new
 
-  importer = Importer.new(ENV['USER'])
-  importer.like_queue = queue
+  importer.item_queue = queue
   importer.report = report
 
   downloader = PostDownloader.new