improved handling of invalid records (hi retr0id 👋)

app/errors.rb

···
+class InvalidRecordError < StandardError
+end

app/firehose_client.rb

···
     elsif op.action == :delete
       @current_user.likes.where(rkey: op.rkey).delete_all
     end
+  rescue StandardError => e
+    log "Error in #process_like (#{msg.seq}, #{op.uri}): #{e}"
+    log e.backtrace.reject { |x| x.include?('/ruby/') }
+    sleep 5 if e.is_a?(ActiveRecord::ConnectionFailed)
   end
 
   def process_repost(msg, op)
···
     elsif op.action == :delete
       @current_user.reposts.where(rkey: op.rkey).delete_all
     end
+  rescue StandardError => e
+    log "Error in #process_repost (#{msg.seq}, #{op.uri}): #{e}"
+    log e.backtrace.reject { |x| x.include?('/ruby/') }
+    sleep 5 if e.is_a?(ActiveRecord::ConnectionFailed)
   end
 
   def process_post(msg, op)
···
         post.destroy
       end
     end
+  rescue StandardError => e
+    log "Error in #process_post (#{msg.seq}, #{op.uri}): #{e}"
+    log e.backtrace.reject { |x| x.include?('/ruby/') }
+    sleep 5 if e.is_a?(ActiveRecord::ConnectionFailed)
   end
 
   def log(text)

app/importers/base_importer.rb

···
 require 'didkit'
 require 'minisky'
+require 'time'
 
 require_relative '../at_uri'
+require_relative '../errors'
 require_relative '../models/post'
 require_relative '../models/user'
 
···
 
   def import_items
     raise NotImplementedError
+  end
+
+  def created_at(record)
+    Time.parse(record['createdAt'])
+  rescue StandardError
+    raise InvalidRecordError
   end
 end

app/importers/likes_importer.rb

···
-require 'time'
 require_relative 'base_importer'
 
 class LikesImporter < BaseImporter
···
 
       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?
+      oldest_date = nil
 
       records.each do |record|
         begin
           like = @user.likes.import_from_record(record['uri'], record['value'], queue: :import)
+
+          record_date = like&.time || created_at(record['value'])
+          oldest_date = [oldest_date, record_date].compact.min
 
           if like && like.pending? && @item_queue
             @item_queue.push(like)
             @report&.update(queue: { length: @item_queue.length })
           end
-        rescue StandardError => e
+        rescue InvalidRecordError => e
           puts "Error in LikesImporter: #{record['uri']}: #{e}"
         end
       end
 
+      @imported_count += records.length
+      @report&.update(importers: { importer_name => { :imported_items => @imported_count }})
+      @report&.update(importers: { importer_name => { :oldest_date => oldest_date }}) if oldest_date
+
       params[:cursor] = cursor
       @import.update!(cursor: cursor)
 
       break if !cursor
-      break if @time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < @time_limit }
+      break if @time_limit && oldest_date && oldest_date < @time_limit
     end
   end
 end

app/importers/posts_importer.rb

···
-require 'time'
 require_relative 'base_importer'
 
 class PostsImporter < BaseImporter
···
 
       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?
+      oldest_date = nil
 
       records.each do |record|
         begin
           quote = @user.quotes.import_from_record(record['uri'], record['value'], queue: :import)
           pin = @user.pins.import_from_record(record['uri'], record['value'], queue: :import)
+
+          record_date = quote&.time || pin&.time || created_at(record['value'])
+          oldest_date = [oldest_date, record_date].compact.min
 
           if @item_queue
             if quote && quote.pending?
···
 
             @report&.update(queue: { length: @item_queue.length })
           end
-        rescue StandardError => e
+        rescue InvalidRecordError => e
           puts "Error in PostsImporter: #{record['uri']}: #{e}"
         end
       end
 
+      @imported_count += records.length
+      @report&.update(importers: { importer_name => { :imported_items => @imported_count }})
+      @report&.update(importers: { importer_name => { :oldest_date => oldest_date }}) if oldest_date
+
       params[:cursor] = cursor
       @import.update!(cursor: cursor)
 
       break if !cursor
-      break if @time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < @time_limit }
+      break if @time_limit && oldest_date && oldest_date < @time_limit
     end
   end
 end

app/importers/reposts_importer.rb

···
-require 'time'
 require_relative 'base_importer'
 
 class RepostsImporter < BaseImporter
···
 
       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?
+      oldest_date = nil
 
       records.each do |record|
         begin
           repost = @user.reposts.import_from_record(record['uri'], record['value'], queue: :import)
+
+          record_date = repost&.time || created_at(record['value'])
+          oldest_date = [oldest_date, record_date].compact.min
 
           if repost && repost.pending? && @item_queue
             @item_queue.push(repost)
             @report&.update(queue: { length: @item_queue.length })
           end
-        rescue StandardError => e
+        rescue InvalidRecordError => e
           puts "Error in RepostsImporter: #{record['uri']}: #{e}"
         end
       end
 
+      @imported_count += records.length
+      @report&.update(importers: { importer_name => { :imported_items => @imported_count }})
+      @report&.update(importers: { importer_name => { :oldest_date => oldest_date }}) if oldest_date
+
       params[:cursor] = cursor
       @import.update!(cursor: cursor)
 
       break if !cursor
-      break if @time_limit && records.any? { |x| Time.parse(x['value']['createdAt']) < @time_limit }
+      break if @time_limit && oldest_date && oldest_date < @time_limit
     end
   end
 end

app/models/user_importable.rb

···
+require_relative '../errors'
+
 module UserImportable
   def import_from_record(item_uri, record, **args)
-    item = self.new_from_record(item_uri, record)
+    item = try_build_from_record(item_uri, record)
     return nil if item.nil? || already_imported?(item)
 
     item.import_item!(args)
+  end
+
+  def try_build_from_record(item_uri, record)
+    self.new_from_record(item_uri, record)
+  rescue StandardError
+    raise InvalidRecordError
   end
 
   def already_imported?(item)

app/post_downloader.rb

···
             puts "Invalid post #{item.post_uri}: #{post.errors.full_messages.join("; ")}"
             invalidate_item(item)
           end
-        rescue StandardError => e
+        rescue InvalidRecordError => e
           puts "Error in PostDownloader: #{item.post_uri}: #{e.class}: #{e}"
+
+          item = items.detect { |x| x.post_uri == data['uri'] }
+          item.update!(queue: nil)
+          items.delete(item)
         end
       end
 
···
   def save_post(post_uri, record)
     did, _, rkey = AT_URI(post_uri)
 
-    text = record.delete('text')
-    created = record.delete('createdAt')
-    record.delete('$type')
-
     author = User.find_or_create_by!(did: did)
 
     if post = Post.find_by(user: author, rkey: rkey)
       return post
+    else
+      post = build_post(author, rkey, record)
+      post.save
+      post
     end
+  end
 
-    Post.create(
+  def build_post(author, rkey, record)
+    text = record.delete('text')
+    created = record.delete('createdAt')
+
+    record.delete('$type')
+
+    Post.new(
       user: author,
       rkey: rkey,
       time: Time.parse(created),
       text: text,
       data: JSON.generate(record)
     )
+  rescue StandardError
+    raise InvalidRecordError
   end
 
   def update_item(item, post)