Add latest changes from gitlab-org/gitlab@16-0-stable-ee

15 files changed, 768 insertions, 24 deletions

diff --git a/lib/bitbucket_server/representation/pull_request.rb b/lib/bitbucket_server/representation/pull_request.rb
index 2f377bdced2..9a8eec67bd7 100644
--- a/lib/bitbucket_server/representation/pull_request.rb
+++ b/lib/bitbucket_server/representation/pull_request.rb
@@ -68,6 +68,23 @@ module BitbucketServer
         raw.dig('toRef', 'latestCommit')
       end
 
+      def to_hash
+        {
+          iid: iid,
+          author: author,
+          author_email: author_email,
+          author_username: author_username,
+          description: description,
+          created_at: created_at,
+          updated_at: updated_at,
+          state: state,
+          title: title,
+          source_branch_name: source_branch_name,
+          target_branch_name: target_branch_name,
+          target_branch_sha: target_branch_sha
+        }
+      end
+
       private
 
       def created_date
diff --git a/lib/gitlab/bitbucket_server_import/importer.rb b/lib/gitlab/bitbucket_server_import/importer.rb
index ea9b79c12fd..6b163cd1b2d 100644
--- a/lib/gitlab/bitbucket_server_import/importer.rb
+++ b/lib/gitlab/bitbucket_server_import/importer.rb
@@ -3,9 +3,10 @@
 module Gitlab
   module BitbucketServerImport
     class Importer
+      include Loggable
+
       attr_reader :recover_missing_commits
       attr_reader :project, :project_key, :repository_slug, :client, :errors, :users, :already_imported_cache_key
-      attr_accessor :logger
 
       BATCH_SIZE = 100
       # The base cache key to use for tracking already imported objects.
@@ -38,7 +39,6 @@ module Gitlab
         @errors = []
         @users = {}
         @temp_branches = []
-        @logger = Gitlab::Import::Logger.build
         @already_imported_cache_key = ALREADY_IMPORTED_CACHE_KEY %
           { project: project.id, collection: collection_method }
       end
@@ -427,26 +427,6 @@ module Gitlab
         }
       end
 
-      def log_debug(details)
-        logger.debug(log_base_data.merge(details))
-      end
-
-      def log_info(details)
-        logger.info(log_base_data.merge(details))
-      end
-
-      def log_warn(details)
-        logger.warn(log_base_data.merge(details))
-      end
-
-      def log_base_data
-        {
-          class: self.class.name,
-          project_id: project.id,
-          project_path: project.full_path
-        }
-      end
-
       def metrics
         @metrics ||= Gitlab::Import::Metrics.new(:bitbucket_server_importer, @project)
       end
diff --git a/lib/gitlab/bitbucket_server_import/importers/lfs_object_importer.rb b/lib/gitlab/bitbucket_server_import/importers/lfs_object_importer.rb
new file mode 100644
index 00000000000..67b8eefc351
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/lfs_object_importer.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class LfsObjectImporter
+        include Loggable
+
+        def initialize(project, lfs_attributes)
+          @project = project
+          @lfs_download_object = LfsDownloadObject.new(**lfs_attributes.symbolize_keys)
+        end
+
+        def execute
+          log_info(import_stage: 'import_lfs_object', message: 'starting', oid: lfs_download_object.oid)
+
+          Projects::LfsPointers::LfsDownloadService.new(project, lfs_download_object).execute
+
+          log_info(import_stage: 'import_lfs_object', message: 'finished', oid: lfs_download_object.oid)
+        end
+
+        private
+
+        attr_reader :project, :lfs_download_object
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/lfs_objects_importer.rb b/lib/gitlab/bitbucket_server_import/importers/lfs_objects_importer.rb
new file mode 100644
index 00000000000..d568f60f4fc
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/lfs_objects_importer.rb
@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class LfsObjectsImporter
+        include ParallelScheduling
+
+        def execute
+          log_info(import_stage: 'import_lfs_objects', message: 'starting')
+
+          download_service = Projects::LfsPointers::LfsObjectDownloadListService.new(project)
+
+          begin
+            queue_workers(download_service) if project&.lfs_enabled?
+          rescue StandardError => e
+            track_import_failure!(project, exception: e)
+          end
+
+          log_info(import_stage: 'import_lfs_objects', message: 'finished')
+
+          job_waiter
+        end
+
+        def sidekiq_worker_class
+          ImportLfsObjectWorker
+        end
+
+        def collection_method
+          :lfs_objects
+        end
+
+        def id_for_already_processed_cache(lfs_download_object)
+          lfs_download_object.oid
+        end
+
+        private
+
+        def queue_workers(download_service)
+          download_service.each_list_item do |lfs_download_object|
+            # Needs to come before `already_processed?` as `jobs_remaining` resets to zero when the job restarts and
+            # jobs_remaining needs to be the total amount of enqueued jobs
+            job_waiter.jobs_remaining += 1
+
+            next if already_processed?(lfs_download_object)
+
+            job_delay = calculate_job_delay(job_waiter.jobs_remaining)
+
+            sidekiq_worker_class.perform_in(job_delay, project.id, lfs_download_object.as_json, job_waiter.key)
+
+            mark_as_processed(lfs_download_object)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/notes_importer.rb b/lib/gitlab/bitbucket_server_import/importers/notes_importer.rb
new file mode 100644
index 00000000000..07ee9569ab1
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/notes_importer.rb
@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class NotesImporter
+        include ParallelScheduling
+
+        def execute
+          project.merge_requests.find_each do |merge_request|
+            # Needs to come before `already_processed?` as `jobs_remaining` resets to zero when the job restarts and
+            # jobs_remaining needs to be the total amount of enqueued jobs
+            job_waiter.jobs_remaining += 1
+
+            next if already_processed?(merge_request)
+
+            job_delay = calculate_job_delay(job_waiter.jobs_remaining)
+
+            sidekiq_worker_class.perform_in(job_delay, project.id, { iid: merge_request.iid }, job_waiter.key)
+
+            mark_as_processed(merge_request)
+          end
+
+          job_waiter
+        end
+
+        private
+
+        attr_reader :project
+
+        def sidekiq_worker_class
+          ImportPullRequestNotesWorker
+        end
+
+        def id_for_already_processed_cache(merge_request)
+          merge_request.iid
+        end
+
+        def collection_method
+          :notes
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/pull_request_importer.rb b/lib/gitlab/bitbucket_server_import/importers/pull_request_importer.rb
new file mode 100644
index 00000000000..5d306f98980
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/pull_request_importer.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class PullRequestImporter
+        include Loggable
+
+        def initialize(project, hash)
+          @project = project
+          @formatter = Gitlab::ImportFormatter.new
+          @user_finder = UserFinder.new(project)
+
+          # Object should behave as a object so we can remove object.is_a?(Hash) check
+          # This will be fixed in https://gitlab.com/gitlab-org/gitlab/-/issues/412328
+          @object = hash.with_indifferent_access
+        end
+
+        def execute
+          log_info(import_stage: 'import_pull_request', message: 'starting', iid: object[:iid])
+
+          description = ''
+          description += author_line
+          description += object[:description] if object[:description]
+
+          attributes = {
+            iid: object[:iid],
+            title: object[:title],
+            description: description,
+            source_project_id: project.id,
+            source_branch: Gitlab::Git.ref_name(object[:source_branch_name]),
+            source_branch_sha: object[:source_branch_sha],
+            target_project_id: project.id,
+            target_branch: Gitlab::Git.ref_name(object[:target_branch_name]),
+            target_branch_sha: object[:target_branch_sha],
+            state_id: MergeRequest.available_states[object[:state]],
+            author_id: user_finder.author_id(object),
+            created_at: object[:created_at],
+            updated_at: object[:updated_at]
+          }
+
+          creator = Gitlab::Import::MergeRequestCreator.new(project)
+
+          creator.execute(attributes)
+
+          log_info(import_stage: 'import_pull_request', message: 'finished', iid: object[:iid])
+        end
+
+        private
+
+        attr_reader :object, :project, :formatter, :user_finder
+
+        def author_line
+          return '' if user_finder.uid(object)
+
+          formatter.author_line(object[:author])
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/pull_request_notes_importer.rb b/lib/gitlab/bitbucket_server_import/importers/pull_request_notes_importer.rb
new file mode 100644
index 00000000000..69de47e2006
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/pull_request_notes_importer.rb
@@ -0,0 +1,183 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class PullRequestNotesImporter
+        include Loggable
+
+        def initialize(project, hash)
+          @project = project
+          @formatter = Gitlab::ImportFormatter.new
+          @client = BitbucketServer::Client.new(project.import_data.credentials)
+          @project_key = project.import_data.data['project_key']
+          @repository_slug = project.import_data.data['repo_slug']
+          @user_finder = UserFinder.new(project)
+
+          # TODO: Convert object into a object instead of using it as a hash
+          @object = hash.with_indifferent_access
+        end
+
+        def execute
+          log_info(import_stage: 'import_pull_request_notes', message: 'starting', iid: object[:iid])
+
+          merge_request = project.merge_requests.find_by(iid: object[:iid]) # rubocop: disable CodeReuse/ActiveRecord
+
+          if merge_request
+            activities = client.activities(project_key, repository_slug, merge_request.iid)
+
+            comments, other_activities = activities.partition(&:comment?)
+
+            merge_event = other_activities.find(&:merge_event?)
+            import_merge_event(merge_request, merge_event) if merge_event
+
+            inline_comments, pr_comments = comments.partition(&:inline_comment?)
+
+            import_inline_comments(inline_comments.map(&:comment), merge_request)
+            import_standalone_pr_comments(pr_comments.map(&:comment), merge_request)
+          end
+
+          log_info(import_stage: 'import_pull_request_notes', message: 'finished', iid: object[:iid])
+        end
+
+        private
+
+        attr_reader :object, :project, :formatter, :client, :project_key, :repository_slug, :user_finder
+
+        # rubocop: disable CodeReuse/ActiveRecord
+        def import_merge_event(merge_request, merge_event)
+          log_info(import_stage: 'import_merge_event', message: 'starting', iid: merge_request.iid)
+
+          committer = merge_event.committer_email
+
+          user_id = user_finder.find_user_id(by: :email, value: committer) || project.creator_id
+          timestamp = merge_event.merge_timestamp
+          merge_request.update({ merge_commit_sha: merge_event.merge_commit })
+          metric = MergeRequest::Metrics.find_or_initialize_by(merge_request: merge_request)
+          metric.update(merged_by_id: user_id, merged_at: timestamp)
+
+          log_info(import_stage: 'import_merge_event', message: 'finished', iid: merge_request.iid)
+        end
+        # rubocop: enable CodeReuse/ActiveRecord
+
+        def import_inline_comments(inline_comments, merge_request)
+          log_info(import_stage: 'import_inline_comments', message: 'starting', iid: merge_request.iid)
+
+          inline_comments.each do |comment|
+            position = build_position(merge_request, comment)
+            parent = create_diff_note(merge_request, comment, position)
+
+            next unless parent&.persisted?
+
+            discussion_id = parent.discussion_id
+
+            comment.comments.each do |reply|
+              create_diff_note(merge_request, reply, position, discussion_id)
+            end
+          end
+
+          log_info(import_stage: 'import_inline_comments', message: 'finished', iid: merge_request.iid)
+        end
+
+        def create_diff_note(merge_request, comment, position, discussion_id = nil)
+          attributes = pull_request_comment_attributes(comment)
+          attributes.merge!(position: position, type: 'DiffNote')
+          attributes[:discussion_id] = discussion_id if discussion_id
+
+          note = merge_request.notes.build(attributes)
+
+          if note.valid?
+            note.save
+            return note
+          end
+
+          log_info(import_stage: 'create_diff_note', message: 'creating fallback DiffNote', iid: merge_request.iid)
+
+          # Bitbucket Server supports the ability to comment on any line, not just the
+          # line in the diff. If we can't add the note as a DiffNote, fallback to creating
+          # a regular note.
+          create_fallback_diff_note(merge_request, comment, position)
+        rescue StandardError => e
+          Gitlab::ErrorTracking.log_exception(
+            e,
+            import_stage: 'create_diff_note', comment_id: comment.id, error: e.message
+          )
+
+          nil
+        end
+
+        def create_fallback_diff_note(merge_request, comment, position)
+          attributes = pull_request_comment_attributes(comment)
+          note = "*Comment on"
+
+          note += " #{position.old_path}:#{position.old_line} -->" if position.old_line
+          note += " #{position.new_path}:#{position.new_line}" if position.new_line
+          note += "*\n\n#{comment.note}"
+
+          attributes[:note] = note
+          merge_request.notes.create!(attributes)
+        end
+
+        def build_position(merge_request, pr_comment)
+          params = {
+            diff_refs: merge_request.diff_refs,
+            old_path: pr_comment.file_path,
+            new_path: pr_comment.file_path,
+            old_line: pr_comment.old_pos,
+            new_line: pr_comment.new_pos
+          }
+
+          Gitlab::Diff::Position.new(params)
+        end
+
+        def import_standalone_pr_comments(pr_comments, merge_request)
+          log_info(import_stage: 'import_standalone_pr_comments', message: 'starting', iid: merge_request.iid)
+
+          pr_comments.each do |comment|
+            merge_request.notes.create!(pull_request_comment_attributes(comment))
+
+            comment.comments.each do |replies|
+              merge_request.notes.create!(pull_request_comment_attributes(replies))
+            end
+          rescue StandardError => e
+            Gitlab::ErrorTracking.log_exception(
+              e,
+              import_stage: 'import_standalone_pr_comments',
+              merge_request_id: merge_request.id,
+              comment_id: comment.id,
+              error: e.message
+            )
+          ensure
+            log_info(import_stage: 'import_standalone_pr_comments', message: 'finished', iid: merge_request.iid)
+          end
+        end
+
+        def pull_request_comment_attributes(comment)
+          author = user_finder.uid(comment)
+          note = ''
+
+          unless author
+            author = project.creator_id
+            note = "*By #{comment.author_username} (#{comment.author_email})*\n\n"
+          end
+
+          note +=
+            # Provide some context for replying
+            if comment.parent_comment
+              "> #{comment.parent_comment.note.truncate(80)}\n\n#{comment.note}"
+            else
+              comment.note
+            end
+
+          {
+            project: project,
+            note: note,
+            author_id: author,
+            created_at: comment.created_at,
+            updated_at: comment.updated_at
+          }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/pull_requests_importer.rb b/lib/gitlab/bitbucket_server_import/importers/pull_requests_importer.rb
new file mode 100644
index 00000000000..92ec10bf037
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/pull_requests_importer.rb
@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class PullRequestsImporter
+        include ParallelScheduling
+
+        def execute
+          page = 1
+
+          loop do
+            log_info(
+              import_stage: 'import_pull_requests', message: "importing page #{page} using batch-size #{BATCH_SIZE}"
+            )
+
+            pull_requests = client.pull_requests(
+              project_key, repository_slug, page_offset: page, limit: BATCH_SIZE
+            ).to_a
+
+            break if pull_requests.empty?
+
+            pull_requests.each do |pull_request|
+              # Needs to come before `already_processed?` as `jobs_remaining` resets to zero when the job restarts and
+              # jobs_remaining needs to be the total amount of enqueued jobs
+              job_waiter.jobs_remaining += 1
+
+              next if already_processed?(pull_request)
+
+              job_delay = calculate_job_delay(job_waiter.jobs_remaining)
+
+              sidekiq_worker_class.perform_in(job_delay, project.id, pull_request.to_hash, job_waiter.key)
+
+              mark_as_processed(pull_request)
+            end
+
+            page += 1
+          end
+
+          job_waiter
+        end
+
+        private
+
+        def sidekiq_worker_class
+          ImportPullRequestWorker
+        end
+
+        def collection_method
+          :pull_requests
+        end
+
+        def id_for_already_processed_cache(object)
+          object.iid
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/importers/repository_importer.rb b/lib/gitlab/bitbucket_server_import/importers/repository_importer.rb
new file mode 100644
index 00000000000..cd09ac40e9f
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/importers/repository_importer.rb
@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Importers
+      class RepositoryImporter
+        include Loggable
+
+        def initialize(project)
+          @project = project
+        end
+
+        def execute
+          log_info(import_stage: 'import_repository', message: 'starting import')
+
+          if project.empty_repo?
+            project.repository.import_repository(project.import_url)
+            project.repository.fetch_as_mirror(project.import_url, refmap: refmap)
+
+            update_clone_time
+          end
+
+          log_info(import_stage: 'import_repository', message: 'finished import')
+
+          true
+        rescue ::Gitlab::Git::CommandError => e
+          Gitlab::ErrorTracking.log_exception(
+            e, import_stage: 'import_repository', message: 'failed import', error: e.message
+          )
+
+          # Expire cache to prevent scenarios such as:
+          # 1. First import failed, but the repo was imported successfully, so +exists?+ returns true
+          # 2. Retried import, repo is broken or not imported but +exists?+ still returns true
+          project.repository.expire_content_cache if project.repository_exists?
+
+          raise
+        end
+
+        private
+
+        attr_reader :project
+
+        def refmap
+          # We omit :heads and :tags since these are fetched in the import_repository
+          ['+refs/pull-requests/*/to:refs/merge-requests/*/head']
+        end
+
+        def update_clone_time
+          project.touch(:last_repository_updated_at)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/loggable.rb b/lib/gitlab/bitbucket_server_import/loggable.rb
new file mode 100644
index 00000000000..e74c33dacdb
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/loggable.rb
@@ -0,0 +1,41 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module Loggable
+      def log_debug(messages)
+        logger.debug(log_data(messages))
+      end
+
+      def log_info(messages)
+        logger.info(log_data(messages))
+      end
+
+      def log_warn(messages)
+        logger.warn(log_data(messages))
+      end
+
+      def log_error(messages)
+        logger.error(log_data(messages))
+      end
+
+      private
+
+      def logger
+        Gitlab::BitbucketServerImport::Logger
+      end
+
+      def log_data(messages)
+        messages.merge(log_base_data)
+      end
+
+      def log_base_data
+        {
+          class: self.class.name,
+          project_id: project.id,
+          project_path: project.full_path
+        }
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/logger.rb b/lib/gitlab/bitbucket_server_import/logger.rb
new file mode 100644
index 00000000000..f2294efe0fe
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/logger.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    class Logger < ::Gitlab::Import::Logger
+      def default_attributes
+        super.merge(import_type: :bitbucket_server)
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/parallel_importer.rb b/lib/gitlab/bitbucket_server_import/parallel_importer.rb
new file mode 100644
index 00000000000..355944ed350
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/parallel_importer.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    class ParallelImporter
+      def self.async?
+        true
+      end
+
+      def self.imports_repository?
+        true
+      end
+
+      def self.track_start_import(project)
+        Gitlab::Import::Metrics.new(:bitbucket_server_importer, project).track_start_import
+      end
+
+      def initialize(project)
+        @project = project
+      end
+
+      def execute
+        Gitlab::Import::SetAsyncJid.set_jid(project.import_state)
+
+        Stage::ImportRepositoryWorker
+          .with_status
+          .perform_async(project.id)
+
+        true
+      end
+
+      private
+
+      attr_reader :project
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/parallel_scheduling.rb b/lib/gitlab/bitbucket_server_import/parallel_scheduling.rb
new file mode 100644
index 00000000000..7fd61bbb206
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/parallel_scheduling.rb
@@ -0,0 +1,101 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    module ParallelScheduling
+      include Loggable
+
+      attr_reader :project, :already_processed_cache_key, :job_waiter_cache_key
+
+      # The base cache key to use for tracking already processed objects.
+      ALREADY_PROCESSED_CACHE_KEY =
+        'bitbucket-server-importer/already-processed/%{project}/%{collection}'
+
+      # The base cache key to use for storing job waiter key
+      JOB_WAITER_CACHE_KEY =
+        'bitbucket-server-importer/job-waiter/%{project}/%{collection}'
+
+      BATCH_SIZE = 100
+
+      # project - An instance of `Project`.
+      def initialize(project)
+        @project = project
+
+        @already_processed_cache_key =
+          format(ALREADY_PROCESSED_CACHE_KEY, project: project.id, collection: collection_method)
+        @job_waiter_cache_key =
+          format(JOB_WAITER_CACHE_KEY, project: project.id, collection: collection_method)
+      end
+
+      private
+
+      def client
+        @client ||= BitbucketServer::Client.new(project.import_data.credentials)
+      end
+
+      def project_key
+        @project_key ||= project.import_data.data['project_key']
+      end
+
+      def repository_slug
+        @repository_slug ||= project.import_data.data['repo_slug']
+      end
+
+      # Returns the ID to use for the cache used for checking if an object has
+      # already been processed or not.
+      #
+      # object - The object we may want to import.
+      def id_for_already_processed_cache(object)
+        raise NotImplementedError
+      end
+
+      # The Sidekiq worker class used for scheduling the importing of objects in
+      # parallel.
+      def sidekiq_worker_class
+        raise NotImplementedError
+      end
+
+      # The name of the method to call to retrieve the data to import.
+      def collection_method
+        raise NotImplementedError
+      end
+
+      def job_waiter
+        @job_waiter ||= begin
+          key = Gitlab::Cache::Import::Caching.read(job_waiter_cache_key)
+          key ||= Gitlab::Cache::Import::Caching.write(job_waiter_cache_key, JobWaiter.generate_key)
+
+          JobWaiter.new(0, key)
+        end
+      end
+
+      def already_processed?(object)
+        id = id_for_already_processed_cache(object)
+
+        Gitlab::Cache::Import::Caching.set_includes?(already_processed_cache_key, id)
+      end
+
+      # Marks the given object as "already processed".
+      def mark_as_processed(object)
+        id = id_for_already_processed_cache(object)
+
+        Gitlab::Cache::Import::Caching.set_add(already_processed_cache_key, id)
+      end
+
+      def calculate_job_delay(job_index)
+        multiplier = (job_index / BATCH_SIZE)
+
+        (multiplier * 1.minute) + 1.second
+      end
+
+      def track_import_failure!(project, exception:, **args)
+        Gitlab::Import::ImportFailureService.track(
+          project_id: project.id,
+          error_source: self.class.name,
+          exception: exception,
+          **args
+        )
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/bitbucket_server_import/user_finder.rb b/lib/gitlab/bitbucket_server_import/user_finder.rb
new file mode 100644
index 00000000000..f96454eb2cc
--- /dev/null
+++ b/lib/gitlab/bitbucket_server_import/user_finder.rb
@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BitbucketServerImport
+    # Class that can be used for finding a GitLab user ID based on a BitBucket user
+
+    class UserFinder
+      attr_reader :project
+
+      CACHE_KEY = 'bitbucket_server-importer/user-finder/%{project_id}/%{by}/%{value}'
+      CACHE_USER_ID_NOT_FOUND = -1
+
+      # project - An instance of `Project`
+      def initialize(project)
+        @project = project
+      end
+
+      def author_id(object)
+        uid(object) || project.creator_id
+      end
+
+      # Object should behave as a object so we can remove object.is_a?(Hash) check
+      # This will be fixed in https://gitlab.com/gitlab-org/gitlab/-/issues/412328
+      def uid(object)
+        # We want this to only match either username or email depending on the flag state.
+        # There should be no fall-through.
+        if Feature.enabled?(:bitbucket_server_user_mapping_by_username)
+          find_user_id(by: :username, value: object.is_a?(Hash) ? object[:author_username] : object.author_username)
+        else
+          find_user_id(by: :email, value: object.is_a?(Hash) ? object[:author_email] : object.author_email)
+        end
+      end
+
+      def find_user_id(by:, value:)
+        return unless value
+
+        cache_key = build_cache_key(by, value)
+        cached_id = cache.read_integer(cache_key)
+
+        return if cached_id == CACHE_USER_ID_NOT_FOUND
+        return cached_id if cached_id
+
+        user = if by == :email
+                 User.find_by_any_email(value, confirmed: true)
+               else
+                 User.find_by_username(value)
+               end
+
+        user&.id.tap do |id|
+          cache.write(cache_key, id || CACHE_USER_ID_NOT_FOUND)
+        end
+      end
+
+      private
+
+      def cache
+        Cache::Import::Caching
+      end
+
+      def build_cache_key(by, value)
+        format(CACHE_KEY, project_id: project.id, by: by, value: value)
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/import_sources.rb b/lib/gitlab/import_sources.rb
index 77b0df765c4..37bcc53019f 100644
--- a/lib/gitlab/import_sources.rb
+++ b/lib/gitlab/import_sources.rb
@@ -12,7 +12,7 @@ module Gitlab
     IMPORT_TABLE = [
       ImportSource.new('github',           'GitHub',            Gitlab::GithubImport::ParallelImporter),
       ImportSource.new('bitbucket',        'Bitbucket Cloud',   Gitlab::BitbucketImport::Importer),
-      ImportSource.new('bitbucket_server', 'Bitbucket Server',  Gitlab::BitbucketServerImport::Importer),
+      ImportSource.new('bitbucket_server', 'Bitbucket Server',  Gitlab::BitbucketServerImport::ParallelImporter),
       ImportSource.new('fogbugz',          'FogBugz',           Gitlab::FogbugzImport::Importer),
       ImportSource.new('git',              'Repository by URL', nil),
       ImportSource.new('gitlab_project',   'GitLab export',     Gitlab::ImportExport::Importer),
@@ -20,6 +20,9 @@ module Gitlab
       ImportSource.new('manifest',         'Manifest file',     nil)
     ].freeze
 
+    LEGACY_IMPORT_TABLE = IMPORT_TABLE.deep_dup
+    LEGACY_IMPORT_TABLE[2].importer = Gitlab::BitbucketServerImport::Importer
+
     class << self
       prepend_mod_with('Gitlab::ImportSources') # rubocop: disable Cop/InjectEnterpriseEditionModule
 
@@ -44,7 +47,9 @@ module Gitlab
       end
 
       def import_table
-        IMPORT_TABLE
+        return IMPORT_TABLE if Feature.enabled?(:bitbucket_server_parallel_importer)
+
+        LEGACY_IMPORT_TABLE
       end
     end
   end