Feat: Restructuring MigrateImport, Dispatching a per product job for import

src/Catalog/Services/SkuBackfillService.php

@@ -8,15 +8,16 @@ use Lunar\Models\ProductVariant;
  * Generates a SKU for every ProductVariant missing one — extracted out of
  * Command\BackfillMissingSkusCommand (which becomes a thin CLI wrapper
  * around this, keeping --dry-run/progress-bar concerns out of the
- * reusable logic) so MigrateImport\RunMigrateImportJob can also call it
+ * reusable logic) so MigrateImport\Shopify\Services\ShopifyExportImporter can
- * directly, right after a Shopify import, with no CLI concerns at all.
+ * also call it directly, once every product job in its import batch has
+ * finished (see that class's own import()), with no CLI concerns at all.
  *
  * Format is "SKU-P{product_id}-V{variant_id}": deterministic and
  * guaranteed unique without a uniqueness check, since product_id/
  * variant_id already are. Only variants with a null `sku` are touched —
  * not an importer bug when one shows up after a Shopify import, the
  * source CSV rows genuinely had no `Variant SKU` value (see
- * MigrateImport\Shopify\ShopifyExportImporter).
+ * MigrateImport\Shopify\Services\ShopifyExportImporter).
  */
 class SkuBackfillService
 {

src/Command/BackfillMissingSkusCommand.php

@@ -10,7 +10,7 @@ use Modules\Core\Catalog\Services\SkuBackfillService;
  * CLI wrapper (--dry-run, a progress bar) around Catalog\Services\
  * SkuBackfillService — see that class's own docblock for the actual
  * backfill logic, also called automatically after a Shopify import (see
- * MigrateImport\RunMigrateImportJob).
+ * MigrateImport\Jobs\RunMigrateImportJob).
  */
 class BackfillMissingSkusCommand extends Command
 {

src/Command/MigrateImportCommand.php

@@ -4,8 +4,8 @@ namespace Modules\Core\Command;
 
 use Illuminate\Console\Command;
 use Lunar\Models\Language;
-use Modules\Core\MigrateImport\ImportSpec;
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
-use Modules\Core\MigrateImport\RunMigrateImportJob;
+use Modules\Core\MigrateImport\Jobs\RunMigrateImportJob;
 
 class MigrateImportCommand extends Command
 {
@@ -67,7 +67,7 @@ class MigrateImportCommand extends Command
         // (HTML)/etc. column per row, no per-locale columns at all — so
         // its text is necessarily written in exactly one language, and
         // there is no reliable way to detect which one from the file
-        // itself. Modules\Core\MigrateImport\ImportLocale::code() used to
+        // itself. Modules\Core\MigrateImport\Services\ImportLocale::code() used to
         // (as its former name, DefaultLocale, admits) assume it always
         // matched this store's own Lunar\Models\
         // Language::getDefault(), which is often wrong (a store's default

src/Command/WipeCatalogCommand.php

@@ -142,7 +142,7 @@ class WipeCatalogCommand extends Command
      * for the skipped products) whose 'image' ImportMapping rows then
      * caused a LATER Shopify re-import to silently reuse those now-
      * orphaned Media objects instead of importing fresh ones — see
-     * MigrateImport\Shopify\ShopifyExportImporter::resolveOrImportImage()'s
+     * MigrateImport\Shopify\Services\ShopifyExportImporter::resolveOrImportImage()'s
      * own docblock for that half of the same incident. Always re-querying
      * the first N remaining rows (never advancing an id cursor) guarantees
      * every product is actually visited exactly once, however many are

src/MigrateImport/Contracts/Importer.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace Modules\Core\MigrateImport\Contracts;
+
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
+
+interface Importer
+{
+    public function import(ImportSpec $spec): void;
+}

src/MigrateImport/DTOs/ImportSpec.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Modules\Core\MigrateImport;
+namespace Modules\Core\MigrateImport\DTOs;
 
 class ImportSpec
 {

src/MigrateImport/Importer.php

@@ -1,8 +0,0 @@
-<?php
-
-namespace Modules\Core\MigrateImport;
-
-interface Importer
-{
-    public function import(ImportSpec $spec): void;
-}

src/MigrateImport/Jobs/RunMigrateImportJob.php

@@ -0,0 +1,45 @@
+<?php
+
+namespace Modules\Core\MigrateImport\Jobs;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Log;
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
+use Modules\Core\MigrateImport\Services\ImporterFactory;
+
+class RunMigrateImportJob implements ShouldQueue
+{
+    use Dispatchable;
+    use InteractsWithQueue;
+    use Queueable;
+    use SerializesModels;
+
+    public function __construct(
+        public readonly ImportSpec $spec,
+    ) {
+    }
+
+    /**
+     * Just hands off to the right Importer and returns — for Shopify,
+     * that importer dispatches a job batch instead of importing inline
+     * (see Shopify\ShopifyExportImporter::import()) and this job's own
+     * work is done the moment that batch is queued, well before the
+     * batch's jobs actually run. The SKU backfill that used to happen
+     * right here, after import() returned, now happens in that batch's
+     * own then() callback instead — running it here would fire before a
+     * single product had actually been imported.
+     */
+    public function handle(): void
+    {
+        Log::info('Import started', ['source' => $this->spec->source, 'type' => $this->spec->type, 'file' => $this->spec->filePath]);
+
+        $importer = ImporterFactory::make($this->spec);
+        $importer->import($this->spec);
+
+        Log::info('Import job complete', ['source' => $this->spec->source]);
+    }
+}

src/MigrateImport/JudgeMe/Resolvers/ProductResolver.php

@@ -3,7 +3,6 @@
 namespace Modules\Core\MigrateImport\JudgeMe\Resolvers;
 
 use Lunar\Models\Product;
-use Lunar\Models\Url;
 
 class ProductResolver
 {

src/MigrateImport/JudgeMe/Services/JudgeMeCsvReader.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Modules\Core\MigrateImport\JudgeMe;
+namespace Modules\Core\MigrateImport\JudgeMe\Services;
 
 use RuntimeException;
 

src/MigrateImport/JudgeMe/Services/JudgeMeExportImporter.php

@@ -1,12 +1,12 @@
 <?php
 
-namespace Modules\Core\MigrateImport\JudgeMe;
+namespace Modules\Core\MigrateImport\JudgeMe\Services;
 
 use Throwable;
 use Illuminate\Support\Carbon;
 use Illuminate\Support\Facades\Log;
-use Modules\Core\MigrateImport\Importer;
+use Modules\Core\MigrateImport\Contracts\Importer;
-use Modules\Core\MigrateImport\ImportSpec;
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
 use Modules\Core\MigrateImport\JudgeMe\Resolvers\ProductResolver;
 use Modules\Core\MigrateImport\Models\ImportMapping;
 use Modules\Core\Review\Models\ProductReview;
@@ -22,11 +22,24 @@ class JudgeMeExportImporter implements Importer
 
     public function import(ImportSpec $spec): void
     {
-        foreach ($this->csvReader->read($spec->filePath) as $row) {
+        $rows = $this->csvReader->read($spec->filePath);
+        $total = count($rows);
+        $imported = 0;
+
+        Log::info('JudgeMe import: starting', ['total' => $total]);
+
+        foreach ($rows as $row) {
             $this->importReview($row);
+            $imported++;
+
+            if ($imported % 100 === 0) {
+                Log::info('JudgeMe import: progress', ['imported' => $imported, 'total' => $total]);
             }
         }
 
+        Log::info('JudgeMe import: finished', ['imported' => $imported, 'total' => $total]);
+    }
+
     private function importReview(array $row): void
     {
         $handle = trim((string) ($row['product_handle'] ?? ''));

src/MigrateImport/RunMigrateImportJob.php

@@ -1,39 +0,0 @@
-<?php
-
-namespace Modules\Core\MigrateImport;
-
-use Illuminate\Bus\Queueable;
-use Illuminate\Contracts\Queue\ShouldQueue;
-use Illuminate\Foundation\Bus\Dispatchable;
-use Illuminate\Queue\InteractsWithQueue;
-use Illuminate\Queue\SerializesModels;
-use Modules\Core\Catalog\Services\SkuBackfillService;
-
-class RunMigrateImportJob implements ShouldQueue
-{
-    use Dispatchable;
-    use InteractsWithQueue;
-    use Queueable;
-    use SerializesModels;
-
-    public function __construct(
-        public readonly ImportSpec $spec,
-    ) {
-    }
-
-    public function handle(SkuBackfillService $skuBackfill): void
-    {
-        $importer = ImporterFactory::make($this->spec);
-        $importer->import($this->spec);
-
-        // Only Shopify's importer creates ProductVariant rows at all (see
-        // Shopify\ShopifyExportImporter) — JudgeMe never touches products,
-        // so running this for that source would just be a guaranteed
-        // no-op query every time. Source CSV rows genuinely can have no
-        // `Variant SKU` value; see SkuBackfillService's own docblock for
-        // why that's synthesized rather than treated as an importer bug.
-        if ($this->spec->source === 'shopify') {
-            $skuBackfill->backfill();
-        }
-    }
-}

src/MigrateImport/Services/ImportLocale.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Modules\Core\MigrateImport;
+namespace Modules\Core\MigrateImport\Services;
 
 use Lunar\Models\Language;
 

src/MigrateImport/Services/ImporterFactory.php

@@ -1,10 +1,12 @@
 <?php
 
-namespace Modules\Core\MigrateImport;
+namespace Modules\Core\MigrateImport\Services;
 
 use InvalidArgumentException;
-use Modules\Core\MigrateImport\JudgeMe\JudgeMeExportImporter;
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
-use Modules\Core\MigrateImport\Shopify\ShopifyExportImporter;
+use Modules\Core\MigrateImport\Contracts\Importer;
+use Modules\Core\MigrateImport\JudgeMe\Services\JudgeMeExportImporter;
+use Modules\Core\MigrateImport\Shopify\Services\ShopifyExportImporter;
 
 class ImporterFactory
 {

src/MigrateImport/Shopify/DTOs/ProductGroup.php

@@ -1,6 +1,6 @@
 <?php
 
-namespace Modules\Core\MigrateImport\Shopify;
+namespace Modules\Core\MigrateImport\Shopify\DTOs;
 
 class ProductGroup
 {

src/MigrateImport/Shopify/Jobs/ImportShopifyProductJob.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Modules\Core\MigrateImport\Shopify\Jobs;
+
+use Illuminate\Bus\Batchable;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use Lunar\Models\CollectionGroup;
+use Lunar\Models\Currency;
+use Modules\Core\MigrateImport\Services\ImportLocale;
+use Modules\Core\MigrateImport\Shopify\DTOs\ProductGroup;
+use Modules\Core\MigrateImport\Shopify\Services\ShopifyExportImporter;
+
+/**
+ * One product per job — ShopifyExportImporter::import() used to loop over
+ * every ProductGroup inline, inside RunMigrateImportJob's own single
+ * process. A large export (hundreds of products, each with variants,
+ * media downloads and conversions) grew that one process's memory past
+ * queue:work's --memory limit (see docker-compose.yml), which kills the
+ * worker mid-job; the container then restarts and the WHOLE import
+ * re-runs from row one, never actually finishing. Splitting into one
+ * job per product means memory resets between jobs (a fresh worker
+ * process picks up each one), and a restart only repeats whichever
+ * single product was in flight — ImportMapping's own per-handle
+ * resolve()/record() already makes re-importing the same product cheap
+ * and idempotent.
+ */
+class ImportShopifyProductJob implements ShouldQueue
+{
+    use Batchable;
+    use Dispatchable;
+    use InteractsWithQueue;
+    use Queueable;
+    use SerializesModels;
+
+    public int $tries = 3;
+
+    public function __construct(
+        private readonly ProductGroup $group,
+        private readonly string $imagesPath,
+        private readonly CollectionGroup $collectionGroup,
+        private readonly Currency $currency,
+        private readonly ?string $locale,
+    ) {
+    }
+
+    public function handle(ShopifyExportImporter $importer): void
+    {
+        if ($this->locale !== null) {
+            ImportLocale::set($this->locale);
+        }
+
+        try {
+            $importer->importProduct($this->group, $this->imagesPath, $this->collectionGroup, $this->currency);
+        } finally {
+            // A queue worker process outlives a single job — this must
+            // not leak into whichever product the same worker picks up
+            // next.
+            ImportLocale::reset();
+        }
+    }
+}

src/MigrateImport/Shopify/Resolvers/CollectionResolver.php

@@ -7,7 +7,7 @@ use Lunar\FieldTypes\TranslatedText;
 use Lunar\Models\Collection;
 use Lunar\Models\CollectionGroup;
 use Lunar\Models\Product;
-use Modules\Core\MigrateImport\ImportLocale;
+use Modules\Core\MigrateImport\Services\ImportLocale;
 
 class CollectionResolver
 {

src/MigrateImport/Shopify/Resolvers/ImportAttributeResolver.php

@@ -8,7 +8,7 @@ use Lunar\Models\Attribute;
 use Lunar\Models\AttributeGroup;
 use Lunar\Models\Product;
 use Lunar\Models\ProductType;
-use Modules\Core\MigrateImport\ImportLocale;
+use Modules\Core\MigrateImport\Services\ImportLocale;
 
 class ImportAttributeResolver
 {

src/MigrateImport/Shopify/Resolvers/ProductAttributeResolver.php

@@ -6,7 +6,7 @@ use Lunar\FieldTypes\Number;
 use Lunar\FieldTypes\Text;
 use Lunar\FieldTypes\TranslatedText;
 use Lunar\Models\ProductType;
-use Modules\Core\MigrateImport\ImportLocale;
+use Modules\Core\MigrateImport\Services\ImportLocale;
 
 class ProductAttributeResolver
 {

src/MigrateImport/Shopify/Resolvers/ProductOptionResolver.php

@@ -5,7 +5,7 @@ namespace Modules\Core\MigrateImport\Shopify\Resolvers;
 use Illuminate\Support\Str;
 use Lunar\Models\ProductOption;
 use Lunar\Models\ProductOptionValue;
-use Modules\Core\MigrateImport\ImportLocale;
+use Modules\Core\MigrateImport\Services\ImportLocale;
 
 class ProductOptionResolver
 {

src/MigrateImport/Shopify/Services/ShopifyCsvReader.php

@@ -1,8 +1,9 @@
 <?php
 
-namespace Modules\Core\MigrateImport\Shopify;
+namespace Modules\Core\MigrateImport\Shopify\Services;
 
 use RuntimeException;
+use Modules\Core\MigrateImport\Shopify\DTOs\ProductGroup;
 
 class ShopifyCsvReader
 {

src/MigrateImport/Shopify/Services/ShopifyExportImporter.php

@@ -1,9 +1,10 @@
 <?php
 
-namespace Modules\Core\MigrateImport\Shopify;
+namespace Modules\Core\MigrateImport\Shopify\Services;
 
 use Lunar\Models\TaxClass;
 use Lunar\Models\ProductOption;
+use Illuminate\Support\Facades\Bus;
 use Illuminate\Support\Facades\Log;
 use Lunar\Models\Collection;
 use Lunar\Models\CollectionGroup;
@@ -12,10 +13,13 @@ use Lunar\Models\Language;
 use Lunar\Models\Product;
 use Lunar\Models\ProductVariant;
 use Lunar\Models\Url;
-use Modules\Core\MigrateImport\ImportLocale;
+use Modules\Core\Catalog\Services\SkuBackfillService;
-use Modules\Core\MigrateImport\ImportSpec;
+use Modules\Core\MigrateImport\Services\ImportLocale;
-use Modules\Core\MigrateImport\Importer;
+use Modules\Core\MigrateImport\DTOs\ImportSpec;
+use Modules\Core\MigrateImport\Shopify\DTOs\ProductGroup;
+use Modules\Core\MigrateImport\Contracts\Importer;
 use Modules\Core\MigrateImport\Models\ImportMapping;
+use Modules\Core\MigrateImport\Shopify\Jobs\ImportShopifyProductJob;
 use Modules\Core\MigrateImport\Shopify\Resolvers\AssetResolver;
 use Modules\Core\MigrateImport\Shopify\Resolvers\BrandResolver;
 use Modules\Core\MigrateImport\Shopify\Resolvers\CollectionResolver;
@@ -47,14 +51,21 @@ class ShopifyExportImporter implements Importer
     ) {
     }
 
+    /**
+     * Dispatches one ImportShopifyProductJob per ProductGroup instead of
+     * importing them inline — see that job's own docblock for why (a
+     * single process holding every group in memory for the whole run
+     * kept exceeding queue:work's --memory limit on a large export,
+     * which kills the worker mid-run and restarts the entire import from
+     * scratch). Bus::batch()'s then() is what SkuBackfillService used to
+     * run right after this loop — now deferred until every product job
+     * in the batch has actually finished, since dispatching a batch
+     * itself returns immediately.
+     */
     public function import(ImportSpec $spec): void
     {
-        if ($spec->locale !== null) {
-            ImportLocale::set($spec->locale);
-        }
-
-        try {
         $groups = $this->csvReader->read($spec->filePath);
+        $total = count($groups);
         $imagesPath = dirname($spec->filePath).'/files';
         $collectionGroup = CollectionGroup::firstOrCreate(
             ['handle' => 'shopify'],
@@ -62,17 +73,30 @@ class ShopifyExportImporter implements Importer
         );
         $currency = Currency::getDefault();
 
-            foreach ($groups as $group) {
+        Log::info('Shopify import: dispatching product jobs', ['total' => $total]);
-                $this->importProduct($group, $imagesPath, $collectionGroup, $currency);
+
-            }
+        $jobs = collect($groups)->map(fn (ProductGroup $group) => new ImportShopifyProductJob(
-        } finally {
+            $group,
-            // A queue worker process outlives a single import run — this
+            $imagesPath,
-            // must not leak into whatever's imported next.
+            $collectionGroup,
-            ImportLocale::reset();
+            $currency,
-        }
+            $spec->locale,
+        ))->all();
+
+        Bus::batch($jobs)
+            ->name("Shopify import: {$spec->filePath}")
+            ->then(function () use ($total) {
+                Log::info('Shopify import: all product jobs finished, backfilling missing SKUs', ['total' => $total]);
+                app(SkuBackfillService::class)->backfill();
+                Log::info('Shopify import: finished', ['total' => $total]);
+            })
+            ->catch(function ($batch, $e) {
+                Log::error('Shopify import: batch failed', ['error' => $e->getMessage()]);
+            })
+            ->dispatch();
     }
 
-    private function importProduct(
+    public function importProduct(
         ProductGroup $group,
         string $imagesPath,
         CollectionGroup $collectionGroup,