extracted import code into models

app/at_uri.rb

···
   attr_reader :repo, :collection, :rkey
 
   def initialize(uri)
+    uri = uri.to_s
     raise InvalidURIError, "Invalid AT URI: #{uri}" if uri.include?(' ') || !uri.start_with?('at://')
 
     parts = uri.split('/')
···
 end
 
 def AT_URI(uri)
+  return uri if uri.is_a?(AT_URI)
   AT_URI.new(uri)
 end

app/importers/base_importer.rb

···
   def import_items
     raise NotImplementedError
   end
-
-  def create_item_for_post(uri)
-    post_uri = AT_URI(uri)
-    return unless post_uri.is_post?
-
-    post = Post.find_by_at_uri(post_uri)
-
-    if post
-      yield({ post: post })
-    else
-      item_stub = yield({ post_uri: post_uri })
-
-      if @item_queue
-        @item_queue.push(item_stub)
-        @report&.update(queue: { length: @item_queue.length })
-      end
-    end
-  end
 end

app/importers/likes_importer.rb

···
 require 'time'
-
-require_relative '../at_uri'
 require_relative 'base_importer'
 
 class LikesImporter < BaseImporter
···
 
       records.each do |record|
         begin
-          like_rkey = AT_URI(record['uri']).rkey
-          next if @user.likes.where(rkey: like_rkey).exists?
+          like = @user.likes.import_from_record(record['uri'], record['value'])
 
-          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))
+          if like && like.pending? && @item_queue
+            @item_queue.push(like)
+            @report&.update(queue: { length: @item_queue.length })
           end
         rescue StandardError => e
           puts "Error in LikesImporter: #{record['uri']}: #{e}"

app/importers/posts_importer.rb

···
 require 'time'
-
-require_relative '../at_uri'
 require_relative 'base_importer'
 
 class PostsImporter < BaseImporter
···
 
       records.each do |record|
         begin
-          if record['value']['embed'] && record['value']['embed']['record']
-            save_quote(record)
-          end
+          quote = @user.quotes.import_from_record(record['uri'], record['value'])
+          pin = @user.pins.import_from_record(record['uri'], record['value'])
 
-          if record['value']['reply'] && record['value']['text'].include?('📌')
-            save_pin(record)
+          if @item_queue
+            if quote && quote.pending?
+              @item_queue.push(quote)
+            end
+
+            if pin && pin.pending?
+              @item_queue.push(pin)
+            end
+
+            @report&.update(queue: { length: @item_queue.length })
           end
         rescue StandardError => e
           puts "Error in LikesImporter: #{record['uri']}: #{e}"
···
 
       break if !cursor
       break if @time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < @time_limit }
-    end
-  end
-
-  def save_quote(record)
-    post_rkey = AT_URI(record['uri']).rkey
-    return if @user.quotes.where(rkey: post_rkey).exists?
-
-    post_time = Time.parse(record['value']['createdAt'])
-
-    quoted_post_uri = case record['value']['embed']['$type']
-    when 'app.bsky.embed.record'
-      record['value']['embed']['record']['uri']
-    when 'app.bsky.embed.recordWithMedia'
-      record['value']['embed']['record']['record']['uri']
-    else
-      return
-    end
-
-    create_item_for_post(quoted_post_uri) do |args|
-      @user.quotes.create!(args.merge(rkey: post_rkey, time: post_time, quote_text: record['value']['text']))
-    end
-  end
-
-  def save_pin(record)
-    post_rkey = AT_URI(record['uri']).rkey
-    return if @user.pins.where(rkey: post_rkey).exists?
-
-    post_time = Time.parse(record['value']['createdAt'])
-    parent_post_uri = record['value']['reply']['parent']['uri']
-
-    create_item_for_post(parent_post_uri) do |args|
-      @user.pins.create!(args.merge(rkey: post_rkey, time: post_time, pin_text: record['value']['text']))
     end
   end
 end

app/importers/reposts_importer.rb

···
 require 'time'
-
-require_relative '../at_uri'
 require_relative 'base_importer'
 
 class RepostsImporter < BaseImporter
···
 
       records.each do |record|
         begin
-          repost_rkey = AT_URI(record['uri']).rkey
-          next if @user.reposts.where(rkey: repost_rkey).exists?
+          repost = @user.reposts.import_from_record(record['uri'], record['value'])
 
-          repost_time = Time.parse(record['value']['createdAt'])
-          post_uri = record['value']['subject']['uri']
-
-          create_item_for_post(post_uri) do |args|
-            @user.reposts.create!(args.merge(rkey: repost_rkey, time: repost_time))
+          if repost && repost.pending? && @item_queue
+            @item_queue.push(repost)
+            @report&.update(queue: { length: @item_queue.length })
           end
         rescue StandardError => e
           puts "Error in RepostsImporter: #{record['uri']}: #{e}"

app/models/importable.rb

···
+require 'active_support/concern'
+
+require_relative '../at_uri'
+require_relative 'post'
+
+module Importable
+  extend ActiveSupport::Concern
+
+  included do
+    scope :pending, -> { where(post: nil) }
+
+    def pending?
+      post_uri != nil
+    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
+      end
+
+      self.save!
+      self
+    end
+  end
+end

app/models/like.rb

···
 require 'active_record'
+require 'time'
 
+require_relative '../at_uri'
 require_relative 'post'
+require_relative 'importable'
 require_relative 'searchable'
 require_relative 'user'
 
 class Like < ActiveRecord::Base
   include Searchable
+  include Importable
 
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
···
 
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
+
+  def self.new_from_record(uri, record)
+    self.new(
+      rkey: AT_URI(uri).rkey,
+      time: Time.parse(record['createdAt']),
+      post_uri: record['subject']['uri']
+    )
+  end
 end

app/models/pin.rb

···
 require 'active_record'
+require 'time'
 
+require_relative '../at_uri'
 require_relative 'post'
 require_relative 'searchable'
 require_relative 'user'
 
 class Pin < ActiveRecord::Base
   include Searchable
+  include Importable
+
+  PIN_SIGN = '📌'
 
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
···
 
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
+
+  def self.new_from_record(uri, record)
+    return nil unless record['reply'] && record['text'].include?(PIN_SIGN)
+
+    self.new(
+      rkey: AT_URI(uri).rkey,
+      time: Time.parse(record['createdAt']),
+      post_uri: record['reply']['parent']['uri'],
+      pin_text: record['text']
+    )
+  end
 end

app/models/quote.rb

···
 require 'active_record'
+require 'time'
 
+require_relative '../at_uri'
 require_relative 'post'
 require_relative 'searchable'
 require_relative 'user'
 
 class Quote < ActiveRecord::Base
   include Searchable
+  include Importable
 
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
···
 
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
+
+  def self.new_from_record(uri, record)
+    return nil unless record['embed']
+
+    quoted_post_uri = case record['embed']['$type']
+    when 'app.bsky.embed.record'
+      record['embed']['record']['uri']
+    when 'app.bsky.embed.recordWithMedia'
+      record['embed']['record']['record']['uri']
+    else
+      return nil
+    end
+
+    self.new(
+      rkey: AT_URI(uri).rkey,
+      time: Time.parse(record['createdAt']),
+      post_uri: quoted_post_uri,
+      quote_text: record['text']
+    )
+  end
 end

app/models/repost.rb

···
 require 'active_record'
+require 'time'
 
+require_relative '../at_uri'
 require_relative 'post'
 require_relative 'searchable'
 require_relative 'user'
 
 class Repost < ActiveRecord::Base
   include Searchable
+  include Importable
 
   validates_presence_of :time, :rkey
   validates_length_of :rkey, is: 13
···
 
   belongs_to :user, foreign_key: 'actor_id'
   belongs_to :post, optional: true
+
+  def self.new_from_record(uri, record)
+    self.new(
+      rkey: AT_URI(uri).rkey,
+      time: Time.parse(record['createdAt']),
+      post_uri: record['subject']['uri']
+    )
+  end
 end

app/models/searchable.rb

···
       )
     }
 
-    scope :pending, -> { where(post: nil) }
-
     def cursor
       "#{self.time.to_f}:#{self.id}"
     end

app/models/user.rb

···
   validates_length_of :did, maximum: 260
 
   has_many :posts
-  has_many :likes, foreign_key: 'actor_id'
-  has_many :reposts, foreign_key: 'actor_id'
-  has_many :quotes, foreign_key: 'actor_id'
-  has_many :pins, foreign_key: 'actor_id'
   has_many :imports
+
+  has_many :likes, foreign_key: 'actor_id' do
+    def import_from_record(like_uri, record)
+      like = self.new_from_record(like_uri, record)
+      return nil if like.nil? || self.where(rkey: like.rkey).exists?
+
+      like.import_item!
+    end
+  end
+
+  has_many :reposts, foreign_key: 'actor_id' do
+    def import_from_record(repost_uri, record)
+      repost = self.new_from_record(repost_uri, record)
+      return nil if repost.nil? || self.where(rkey: repost.rkey).exists?
+
+      repost.import_item!
+    end
+  end
+
+  has_many :quotes, foreign_key: 'actor_id' do
+    def import_from_record(post_uri, record)
+      quote = self.new_from_record(post_uri, record)
+      return nil if quote.nil? || self.where(rkey: quote.rkey).exists?
+
+      quote.import_item!
+    end
+  end
+
+  has_many :pins, foreign_key: 'actor_id' do
+    def import_from_record(post_uri, record)
+      pin = self.new_from_record(post_uri, record)
+      return nil if pin.nil? || self.where(rkey: pin.rkey).exists?
+
+      pin.import_item!
+    end
+  end
 
   def all_pending_items
     [:likes, :reposts, :quotes, :pins].map { |x| self.send(x).pending.to_a }.reduce(&:+)