added queue field to items to mark where they're from

app/firehose_client.rb

···
     return unless @current_user
 
     if op.action == :create
-      @current_user.likes.import_from_record(op.uri, op.raw_record)
+      @current_user.likes.import_from_record(op.uri, op.raw_record, queue: :firehose)
     elsif op.action == :delete
       @current_user.likes.where(rkey: op.rkey).delete_all
     end
···
     return unless @current_user
 
     if op.action == :create
-      @current_user.reposts.import_from_record(op.uri, op.raw_record)
+      @current_user.reposts.import_from_record(op.uri, op.raw_record, queue: :firehose)
     elsif op.action == :delete
       @current_user.reposts.where(rkey: op.rkey).delete_all
     end
···
   def process_post(msg, op)
     if op.action == :create
       if @current_user
-        @current_user.quotes.import_from_record(op.uri, op.raw_record)
-        @current_user.pins.import_from_record(op.uri, op.raw_record)
+        @current_user.quotes.import_from_record(op.uri, op.raw_record, queue: :firehose)
+        @current_user.pins.import_from_record(op.uri, op.raw_record, queue: :firehose)
       end
     elsif op.action == :delete
       if @current_user

app/import_manager.rb

···
     @user = user
   end
 
-  def start(sets, include_pending)
+  def start(sets)
     queued_items = []
     importers = []
     sets = [sets] unless sets.is_a?(Array)
···
     sets.each do |set|
       case set
       when 'likes'
-        queued_items += @user.likes.pending.to_a if include_pending
+        queued_items += @user.likes.in_queue(:import).to_a
         importers << LikesImporter.new(@user)
       when 'reposts'
-        queued_items += @user.reposts.pending.to_a if include_pending
+        queued_items += @user.reposts.in_queue(:import).to_a
         importers << RepostsImporter.new(@user)
       when 'posts'
-        queued_items += @user.quotes.pending.to_a + @user.pins.pending.to_a if include_pending
+        queued_items += @user.quotes.in_queue(:import).to_a + @user.pins.in_queue(:import).to_a
         importers << PostsImporter.new(@user)
       when 'all'
-        queued_items += @user.all_pending_items if include_pending
+        queued_items += @user.all_items_in_queue(:import)
         importers += [
           LikesImporter.new(@user),
           RepostsImporter.new(@user),

app/import_worker.rb

···
     def run(collections)
       import = ImportManager.new(@user)
       import.report = BasicReport.new if @verbose
-      import.start(collections, false)
+      import.start(collections)
     end
   end
 

app/importers/likes_importer.rb

···
 
       records.each do |record|
         begin
-          like = @user.likes.import_from_record(record['uri'], record['value'])
+          like = @user.likes.import_from_record(record['uri'], record['value'], queue: :import)
 
           if like && like.pending? && @item_queue
             @item_queue.push(like)

app/importers/posts_importer.rb

···
 
       records.each do |record|
         begin
-          quote = @user.quotes.import_from_record(record['uri'], record['value'])
-          pin = @user.pins.import_from_record(record['uri'], record['value'])
+          quote = @user.quotes.import_from_record(record['uri'], record['value'], queue: :import)
+          pin = @user.pins.import_from_record(record['uri'], record['value'], queue: :import)
 
           if @item_queue
             if quote && quote.pending?

app/importers/reposts_importer.rb

···
 
       records.each do |record|
         begin
-          repost = @user.reposts.import_from_record(record['uri'], record['value'])
+          repost = @user.reposts.import_from_record(record['uri'], record['value'], queue: :import)
 
           if repost && repost.pending? && @item_queue
             @item_queue.push(repost)

app/models/importable.rb

···
 
   included do
     scope :pending, -> { where(post: nil) }
+    scope :in_queue, ->(q) { where(queue: q) }
+
+    enum :queue, { firehose: 0, import: 1 }
+
+    validates_presence_of :post_uri, if: -> { post_id.nil? }
+    validate :check_queue
 
     def pending?
       post_uri != nil
     end
 
+    def check_queue
+      errors.add(:queue, 'must be nil if already processed') if queue && post
+    end
+
     def import_item!
       post_uri = AT_URI(self.post_uri)
       return nil if !post_uri.is_post?
···
       if post = Post.find_by_at_uri(post_uri)
         self.post = post
         self.post_uri = nil
+        self.queue = nil
       end
 
       self.save!

app/models/like.rb

···
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
 
-  validates_presence_of :post_uri, if: -> { post_id.nil? }
-
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
 

app/models/pin.rb

···
   validates_length_of :rkey, is: 13
   validates :pin_text, length: { minimum: 0, maximum: 1000, allow_nil: false }
 
-  validates_presence_of :post_uri, if: -> { post_id.nil? }
-
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
 

app/models/quote.rb

···
   validates_length_of :rkey, is: 13
   validates :quote_text, length: { minimum: 0, maximum: 1000, allow_nil: false }
 
-  validates_presence_of :post_uri, if: -> { post_id.nil? }
-
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
 

app/models/repost.rb

···
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
 
-  validates_presence_of :post_uri, if: -> { post_id.nil? }
-
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
 

app/models/user.rb

···
   before_destroy :delete_posts_cascading
 
   has_many :likes, foreign_key: 'actor_id', dependent: :delete_all do
-    def import_from_record(like_uri, record)
+    def import_from_record(like_uri, record, **args)
       like = self.new_from_record(like_uri, record)
       return nil if like.nil? || self.where(rkey: like.rkey).exists?
 
+      like.assign_attributes(args)
       like.import_item!
     end
   end
 
   has_many :reposts, foreign_key: 'actor_id', dependent: :delete_all do
-    def import_from_record(repost_uri, record)
+    def import_from_record(repost_uri, record, **args)
       repost = self.new_from_record(repost_uri, record)
       return nil if repost.nil? || self.where(rkey: repost.rkey).exists?
 
+      repost.assign_attributes(args)
       repost.import_item!
     end
   end
 
   has_many :quotes, foreign_key: 'actor_id', dependent: :delete_all do
-    def import_from_record(post_uri, record)
+    def import_from_record(post_uri, record, **args)
       quote = self.new_from_record(post_uri, record)
       return nil if quote.nil? || self.where(rkey: quote.rkey).exists?
 
+      quote.assign_attributes(args)
       quote.import_item!
     end
   end
 
   has_many :pins, foreign_key: 'actor_id', dependent: :delete_all do
-    def import_from_record(post_uri, record)
+    def import_from_record(post_uri, record, **args)
       pin = self.new_from_record(post_uri, record)
       return nil if pin.nil? || self.where(rkey: pin.rkey).exists?
 
+      pin.assign_attributes(args)
       pin.import_item!
     end
   end
···
 
   def all_pending_items
     [:likes, :reposts, :quotes, :pins].map { |x| self.send(x).pending.to_a }.reduce(&:+)
+  end
+
+  def all_items_in_queue(queue)
+    [:likes, :reposts, :quotes, :pins].map { |x| self.send(x).in_queue(queue).to_a }.reduce(&:+)
   end
 
   def delete_posts_cascading

app/post_downloader.rb

···
   end
 
   def update_item(item, post)
-    item.update!(post: post, post_uri: nil)
+    item.update!(post: post, post_uri: nil, queue: nil)
 
     @total_count += 1
     @oldest_imported = [@oldest_imported, item.time].min
···
           # delete reference if the account's PDS is the old bsky.social (so it must have been deleted pre Nov 2023)
           item.destroy if hostname == 'bsky.social'
         end
+      end
+
+      if !item.destroyed?
+        item.update!(queue: nil)
       end
     end
   end

db/migrate/20250923014702_add_queued_field.rb

···
+class AddQueuedField < ActiveRecord::Migration[7.2]
+  def change
+    [:likes, :reposts, :quotes, :pins].each do |table|
+      add_column(table, :queue, :smallint, null: true)
+    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_09_20_182018) do
+ActiveRecord::Schema[7.2].define(version: 2025_09_23_014702) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
 
···
     t.datetime "time", null: false
     t.bigint "post_id"
     t.string "post_uri"
+    t.integer "queue", limit: 2
     t.index ["actor_id", "rkey"], name: "index_likes_on_actor_id_and_rkey", unique: true
     t.index ["actor_id", "time", "id"], name: "index_likes_on_actor_id_and_time_and_id", order: { time: :desc, id: :desc }
   end
···
     t.text "pin_text", null: false
     t.bigint "post_id"
     t.string "post_uri"
+    t.integer "queue", limit: 2
     t.index ["actor_id", "rkey"], name: "index_pins_on_actor_id_and_rkey", unique: true
     t.index ["actor_id", "time", "id"], name: "index_pins_on_actor_id_and_time_and_id", order: { time: :desc, id: :desc }
   end
···
     t.text "quote_text", null: false
     t.bigint "post_id"
     t.string "post_uri"
+    t.integer "queue", limit: 2
     t.index ["actor_id", "rkey"], name: "index_quotes_on_actor_id_and_rkey", unique: true
     t.index ["actor_id", "time", "id"], name: "index_quotes_on_actor_id_and_time_and_id", order: { time: :desc, id: :desc }
   end
···
     t.datetime "time", null: false
     t.bigint "post_id"
     t.string "post_uri"
+    t.integer "queue", limit: 2
     t.index ["actor_id", "rkey"], name: "index_reposts_on_actor_id_and_rkey", unique: true
     t.index ["actor_id", "time", "id"], name: "index_reposts_on_actor_id_and_time_and_id", order: { time: :desc, id: :desc }
   end

lib/tasks/import.rake

···
   end
 
   user = User.find_or_create_by!(did: ENV['DID'])
-  pending = !ENV['SKIP_PENDING']
 
   unless ENV['COLLECTION']
     raise "Required COLLECTION parameter missing"
···
     exit
   }
 
-  import.start(ENV['COLLECTION'], pending)
+  import.start(ENV['COLLECTION'])
 
   puts "\n\n\n\n\n"
 end