Improve uniquness checks on import

app/jobs/account/data_import_job.rb

@@ -8,13 +8,13 @@ def perform(import)
     step :check do |step|
       import.check \
         start: step.cursor,
-        callback: ->(record_set:, file:) { step.set!([ record_set.model.name, file ]) }
+        callback: ->(record_set:, file:) { step.set!([ record_set.cursor_key, file ]) }
     end
 
     step :process do |step|
       import.process \
         start: step.cursor,
-        callback: ->(record_set:, files:) { step.set!([ record_set.model.name, files.last ]) }
+        callback: ->(record_set:, files:) { step.set!([ record_set.cursor_key, files.last ]) }
     end
   end
 end

app/models/account/data_transfer/action_text/rich_text_record_set.rb

@@ -51,7 +51,12 @@ def check_record(file_path)
         raise IntegrityError, "#{file_path} is missing required fields: #{missing.join(', ')}"
       end
 
+      if ::ActionText::RichText.exists?(id: data["id"])
+        raise ConflictError, "ActionTextRichText record with ID #{data['id']} already exists"
+      end
+
       check_associations_dont_exist(data)
+      check_unique_values_arent_duplicated(data)
     end
 
     def transform_body_for_export(content)

app/models/account/data_transfer/active_storage/file_record_set.rb

@@ -21,7 +21,7 @@ def export_record(blob)
     end
 
     def files
-      zip.glob("storage/*")
+      zip.glob("storage/**/*")
     end
 
     def import_batch(files)

app/models/account/data_transfer/board/publication_record_set.rb

@@ -0,0 +1,15 @@
+class Account::DataTransfer::Board::PublicationRecordSet < Account::DataTransfer::RecordSet
+  def initialize(account)
+    super(account: account, model: ::Board::Publication)
+  end
+
+  private
+    def check_record(file_path)
+      super
+
+      data = load(file_path)
+      if !data["key"].nil? && model.exists?(key: data["key"])
+        raise ConflictError, "#{model} record with key #{data['key']} already exists"
+      end
+    end
+end

app/models/account/data_transfer/manifest.rb

@@ -9,16 +9,18 @@ def each_record_set(start: nil)
     raise ArgumentError, "No block given" unless block_given?
 
     started = start.nil?
-    record_class, last_id = start if start
+    cursor_key, last_id = start if start
 
     record_sets.each do |record_set|
       if started
         yield record_set
-      elsif record_set.model.name == record_class
+      elsif record_set.cursor_key == cursor_key
         started = true
         yield record_set, last_id
       end
     end
+
+    raise ArgumentError, "Unknown record set cursor: #{cursor_key}" unless started
   end
 
   private
@@ -33,8 +35,8 @@ def record_sets
           ::Column
         ),
         Account::DataTransfer::EntropyRecordSet.new(account),
+        Account::DataTransfer::Board::PublicationRecordSet.new(account),
         *record_sets_for(
-          ::Board::Publication,
           ::Webhook,
           ::Access,
           ::Card,

app/models/account/data_transfer/record_set.rb

@@ -15,6 +15,10 @@ def initialize(account:, model:, attributes: nil, importable_model_names: nil)
     @importable_model_names = importable_model_names || [ model.name ]
   end
 
+  def cursor_key
+    "#{self.class.name}/#{model.name}"
+  end
+
   def export(to:, start: nil)
     with_zip(to) do
       block = lambda do |record|
@@ -40,13 +44,16 @@ def import(from:, start: nil, callback: nil)
   def check(from:, start: nil, callback: nil)
     with_zip(from) do
       file_list = files
+      check_file_names_arent_ambiguous(file_list)
       file_list = skip_to(file_list, start) if start
 
       file_list.each do |file_path|
         check_record(file_path)
         callback&.call(record_set: self, file: file_path)
       end
     end
+  ensure
+    clear_duplicate_detector
   end
 
   private
@@ -83,6 +90,14 @@ def import_batch(files)
       model.insert_all!(batch_data)
     end
 
+    def check_file_names_arent_ambiguous(file_list)
+      duplicates = file_list.map(&:downcase).tally.filter_map { |name, count| name if count > 1 }
+
+      if duplicates.any?
+        raise IntegrityError, "#{model} has multiple files for the same record: #{duplicates.join(', ')}"
+      end
+    end
+
     def check_record(file_path)
       data = load(file_path)
       expected_id = File.basename(file_path, ".json")
@@ -101,6 +116,7 @@ def check_record(file_path)
       end
 
       check_associations_dont_exist(data)
+      check_unique_values_arent_duplicated(data)
     end
 
     def check_associations_dont_exist(data)
@@ -134,6 +150,30 @@ def verify_model_type(type_name)
       end
     end
 
+    def check_unique_values_arent_duplicated(data)
+      # Remap the data being imported to the account we are importing it to.
+      # This ensures that uniquness checks are valid, else someone could change
+      # the account_id in the export to point to different accounts and circumvent
+      # this check.
+      data = data.merge("account_id" => account.id)
+
+      if columns = duplicate_detector.detect(data)
+        raise IntegrityError, "#{model} has multiple records with the same #{columns.to_sentence}: #{data.values_at(*columns).join(', ')}"
+      end
+    end
+
+    def duplicate_detector
+      @duplicate_detector ||= DuplicateDetector.new(unique_key_sets)
+    end
+
+    def clear_duplicate_detector
+      @duplicate_detector = nil
+    end
+
+    def unique_key_sets
+      @unique_key_sets ||= model.connection.indexes(model.table_name).select(&:unique).map(&:columns)
+    end
+
     def skip_to(file_list, last_id)
       index = file_list.index(last_id)
 

app/models/account/data_transfer/record_set/duplicate_detector.rb

@@ -0,0 +1,27 @@
+class Account::DataTransfer::RecordSet::DuplicateDetector
+  def initialize(key_sets)
+    @key_sets = key_sets
+    @seen = Hash.new { |seen, columns| seen[columns] = Set.new }
+  end
+
+  def detect(data)
+    key_sets.find do |columns|
+      values = data.values_at(*columns)
+      values.none?(&:nil?) && !seen[columns].add?(digest(values))
+    end
+  end
+
+  private
+    attr_reader :key_sets, :seen
+
+    # NOTE: The digest is here to avoid storing whole row values in memory, which could be quite large.
+    # (remeber, the value is user controlled, someone could craft a malicious import to baloon memory)
+    # This is an improvement, but it still could be a problem if the number of unique values is very large.
+    # In that case, we may need to use a more memory-efficient approach, such as storing the digests in a
+    # temporary file, or we could switch to a Bloom filter, or a Merkel tree.
+    # MySQL's default collation compares strings case-insensitively, and UUID columns
+    # cast their values through base36, so values differing only in case are duplicates.
+    def digest(values)
+      Digest::SHA256.digest(values.map { |value| value.is_a?(String) ? value.downcase : value }.to_json)
+    end
+end

app/models/account/data_transfer/user_record_set.rb

@@ -57,5 +57,19 @@ def check_record(file_path)
       unless (ATTRIBUTES - data.keys).empty?
         raise IntegrityError, "#{file_path} is missing required fields"
       end
+
+      if User.exists?(id: data["id"])
+        raise ConflictError, "User record with ID #{data['id']} already exists"
+      end
+
+      check_unique_values_arent_duplicated data.merge("email_address" => normalized_email_address(data))
+    end
+
+    def normalized_email_address(data)
+      Identity.normalize_value_for(:email_address, data["email_address"])
+    end
+
+    def unique_key_sets
+      super + [ %w[ email_address ] ]
     end
 end

app/models/account/import.rb

@@ -23,8 +23,9 @@ def check(start: nil, callback: nil)
     processing!
 
     ZipFile.read_from(file.blob) do |zip|
-      Account::DataTransfer::Manifest.new(account).each_record_set(start: start) do |record_set, last_id|
-        record_set.check(from: zip, start: last_id, callback: callback)
+      Account::DataTransfer::Manifest.new(account).each_record_set(start: start) do |record_set|
+        # The last_id is ignored so that we can check for record uniquness when interrupted
+        record_set.check(from: zip, callback: callback)
       end
     end
   rescue Account::DataTransfer::RecordSet::ConflictError => e

app/models/zip_file/reader.rb

@@ -2,6 +2,8 @@ class ZipFile::Reader
   def initialize(io)
     @io = io
     @reader = ZipKit::FileReader.read_zip_structure(io: io)
+
+    ensure_unique_entry_names
   rescue ZipKit::FileReader::ReadError, ZipKit::FileReader::MissingEOCD, ZipKit::FileReader::UnsupportedFeature => e
     raise ZipFile::InvalidFileError, e.message
   end
@@ -19,10 +21,21 @@ def read(file_path)
   end
 
   def glob(pattern)
-    @reader.map(&:filename).select { |name| File.fnmatch(pattern, name) }.sort
+    @reader.map(&:filename).select { |name| File.fnmatch(pattern, name, File::FNM_PATHNAME) }.sort
   end
 
   def exists?(file_path)
     @reader.any? { |e| e.filename == file_path }
   end
+
+  private
+    # ZIP files can technically contian duplicate entries in their manifest.
+    # But that doesn't produce a valid filesystem, so we flat-out reject them here.
+    def ensure_unique_entry_names
+      duplicates = @reader.map(&:filename).tally.filter_map { |name, count| name if count > 1 }
+
+      if duplicates.any?
+        raise ZipFile::InvalidFileError, "Duplicate entries in zip: #{duplicates.join(', ')}"
+      end
+    end
 end

test/models/account/data_transfer/manifest_test.rb

@@ -0,0 +1,40 @@
+require "test_helper"
+
+class Account::DataTransfer::ManifestTest < ActiveSupport::TestCase
+  setup do
+    @manifest = Account::DataTransfer::Manifest.new(accounts("37s"))
+  end
+
+  test "each_record_set resumes at the record set identified by the cursor" do
+    file_record_set = record_sets.find { |record_set| record_set.is_a?(Account::DataTransfer::ActiveStorage::FileRecordSet) }
+
+    yielded = []
+    @manifest.each_record_set(start: [ file_record_set.cursor_key, "storage/somekey" ]) do |record_set, last_id|
+      yielded << record_set
+    end
+
+    assert_equal 1, yielded.size
+    assert_instance_of Account::DataTransfer::ActiveStorage::FileRecordSet, yielded.first
+  end
+
+  test "each_record_set distinguishes record sets sharing a model" do
+    blob_record_set = record_sets.find { |record_set| record_set.is_a?(Account::DataTransfer::ActiveStorage::BlobRecordSet) }
+    file_record_set = record_sets.find { |record_set| record_set.is_a?(Account::DataTransfer::ActiveStorage::FileRecordSet) }
+
+    assert_equal blob_record_set.model, file_record_set.model
+    assert_not_equal blob_record_set.cursor_key, file_record_set.cursor_key
+  end
+
+  test "each_record_set rejects a cursor that matches no record set" do
+    assert_raises(ArgumentError) do
+      @manifest.each_record_set(start: [ "Nonexistent/RecordSet", "somefile" ]) { }
+    end
+  end
+
+  private
+    def record_sets
+      [].tap do |sets|
+        @manifest.each_record_set { |record_set| sets << record_set }
+      end
+    end
+end