1 files changed, 36 insertions, 13 deletions
diff --git a/app/workers/bulk_imports/entity_worker.rb b/app/workers/bulk_imports/entity_worker.rb
index 5b41ccbdea1..7f173b738cf 100644
--- a/app/workers/bulk_imports/entity_worker.rb
+++ b/app/workers/bulk_imports/entity_worker.rb
@@ -10,24 +10,47 @@ module BulkImports
 
     worker_has_external_dependencies!
 
-    def perform(entity_id)
-      entity = BulkImports::Entity.with_status(:started).find_by_id(entity_id)
+    def perform(entity_id, current_stage = nil)
+      return if stage_running?(entity_id, current_stage)
+
+      logger.info(
+        worker: self.class.name,
+        entity_id: entity_id,
+        current_stage: current_stage
+      )
+
+      next_pipeline_trackers_for(entity_id).each do |pipeline_tracker|
+        BulkImports::PipelineWorker.perform_async(
+          pipeline_tracker.id,
+          pipeline_tracker.stage,
+          entity_id
+        )
+      end
+    rescue => e
+      logger.error(
+        worker: self.class.name,
+        entity_id: entity_id,
+        current_stage: current_stage,
+        error_message: e.message
+      )
+
+      Gitlab::ErrorTracking.track_exception(e, entity_id: entity_id)
+    end
 
-      if entity
-        entity.update!(jid: jid)
+    private
 
-        BulkImports::Importers::GroupImporter.new(entity).execute
-      end
+    def stage_running?(entity_id, stage)
+      return unless stage
 
-    rescue => e
-      extra = {
-        bulk_import_id: entity&.bulk_import&.id,
-        entity_id: entity&.id
-      }
+      BulkImports::Tracker.stage_running?(entity_id, stage)
+    end
 
-      Gitlab::ErrorTracking.track_exception(e, extra)
+    def next_pipeline_trackers_for(entity_id)
+      BulkImports::Tracker.next_pipeline_trackers_for(entity_id)
+    end
 
-      entity&.fail_op
+    def logger
+      @logger ||= Gitlab::Import::Logger.build
     end
   end
 end