nshop/app/Console/Commands/MigrateAsteriaReviews.php

<?php

namespace App\Console\Commands;

use App\Jobs\MigrateReviewJob;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Schema;

/**
 * Migrates product reviews from Asteria (Magento 1.x) into this Bagisto store.
 *
 * Prerequisites
 * ─────────────
 * 1. Add the `asteria` database connection to config/database.php (see the
 *    comment block at the bottom of that file or the README).
 * 2. Add the `migrated_from_asteria_id` column to product_reviews:
 *
 *      php artisan migrate  (after creating the migration below)
 *
 *    Or run this one-off SQL directly if you prefer:
 *
 *      ALTER TABLE product_reviews
 *        ADD COLUMN migrated_from_asteria_id BIGINT UNSIGNED NULL DEFAULT NULL,
 *        ADD UNIQUE INDEX uq_migrated_asteria_id (migrated_from_asteria_id);
 *
 * 3. Ensure a queue worker is running:
 *
 *      php artisan queue:work --queue=review-migration
 *
 * Usage
 * ─────
 * php artisan reviews:migrate-asteria
 * php artisan reviews:migrate-asteria --batch-size=200
 * php artisan reviews:migrate-asteria --status=1          # approved only
 * php artisan reviews:migrate-asteria --reset-progress    # start from scratch
 * php artisan reviews:migrate-asteria --dry-run           # dispatch nothing, just show stats
 * php artisan reviews:migrate-asteria --sync              # run synchronously (no queue)
 */
class MigrateAsteriaReviews extends Command
{
    protected $signature = 'reviews:migrate-asteria
        {--batch-size=100     : Number of reviews per queue job}
        {--status=            : Filter by Magento status_id (1=approved,2=pending,3=not-approved)}
        {--reset-progress     : Ignore saved progress and start from review_id=0}
        {--dry-run            : Count records without dispatching any jobs}
        {--sync               : Process synchronously instead of via the queue}
        {--connection=asteria : Name of the Laravel DB connection for the Asteria database}';

    protected $description = 'Migrate Asteria (Magento 1.x) product reviews into Bagisto asynchronously';

    /** Cache key used to store the last successfully enqueued review_id. */
    private const PROGRESS_KEY = 'migrate_asteria_reviews_last_id';

    /** Default queue name for the migration jobs. */
    private const QUEUE_NAME = 'review-migration';

    public function handle(): int
    {
        $connection  = (string) $this->option('connection');
        $batchSize   = max(1, (int) $this->option('batch-size'));
        $statusId    = $this->option('status') !== null ? (int) $this->option('status') : null;
        $resetProgress = (bool) $this->option('reset-progress');
        $dryRun      = (bool) $this->option('dry-run');
        $sync        = (bool) $this->option('sync');

        // ── Pre-flight: make sure the connection and source tables exist ─────────
        try {
            DB::connection($connection)->getPdo();
        } catch (\Throwable $e) {
            $this->error("Cannot connect to Asteria DB (connection='{$connection}'): ".$e->getMessage());
            $this->warn('Add an "asteria" entry to config/database.php (see the comment at the end of that file).');

            return self::FAILURE;
        }

        if (! $this->ensureMigrationColumnExists()) {
            return self::FAILURE;
        }

        // ── Build product and customer lookup maps ────────────────────────────────
        // Matching ignores Magento/Bagisto product status and inventory — disabled or
        // out-of-stock products still receive reviews when the SKU (or Asteria id) exists.
        $this->info('Building product map (status/stock ignored)…');
        $productIdMap = $this->buildProductIdMap($connection);
        $this->line('  '.count($productIdMap).' Magento products matched to Bagisto products (SKU / migrated_from_asteria_id).');

        if (empty($productIdMap)) {
            $this->warn('No matching products found. Import products first (including disabled: ASTERIA_PRODUCTS_ONLY_ENABLED=false).');
        }

        $this->info('Building customer e-mail map…');
        $customerEmailMap = $this->buildCustomerEmailMap($connection);
        $this->line('  '.count($customerEmailMap).' Magento customers matched to Bagisto customers by e-mail.');

        // ── Determine starting review_id ──────────────────────────────────────────
        $lastId = $resetProgress ? 0 : (int) Cache::get(self::PROGRESS_KEY, 0);

        if ($lastId > 0) {
            $this->line("Resuming from Asteria review_id > {$lastId} (use --reset-progress to restart).");
        }

        // ── Dispatch batches ──────────────────────────────────────────────────────
        $totalDispatched = 0;
        $totalSkipped    = 0;
        $batchNumber     = 0;

        $this->info($dryRun ? '[DRY RUN] Scanning reviews…' : 'Dispatching migration jobs…');

        do {
            $rows = $this->fetchBatch($connection, $lastId, $batchSize, $statusId);

            if ($rows->isEmpty()) {
                break;
            }

            $batchNumber++;
            $batchRows = $rows->toArray();

            // Collect unique Magento product IDs in this batch, filter unknown ones.
            $unknownProducts = array_filter(
                array_unique(array_column($batchRows, 'magento_product_id')),
                fn ($id) => ! isset($productIdMap[$id])
            );

            if (! empty($unknownProducts)) {
                $this->warn("  Batch #{$batchNumber}: ".count($unknownProducts).
                    ' Magento product(s) not found in Bagisto: '.implode(', ', array_slice($unknownProducts, 0, 10)));
                $totalSkipped += count(array_filter($batchRows, fn ($r) => ! isset($productIdMap[$r->magento_product_id])));
            }

            $lastId = max(array_column($batchRows, 'review_id'));

            if (! $dryRun) {
                if ($sync) {
                    (new MigrateReviewJob(
                        array_map(fn ($r) => (array) $r, $batchRows),
                        $productIdMap,
                        $customerEmailMap,
                    ))->handle();
                } else {
                    MigrateReviewJob::dispatch(
                        array_map(fn ($r) => (array) $r, $batchRows),
                        $productIdMap,
                        $customerEmailMap,
                    )->onQueue(self::QUEUE_NAME);
                }

                Cache::put(self::PROGRESS_KEY, $lastId, now()->addDays(30));
            }

            $totalDispatched += count($batchRows);

            $this->line(sprintf(
                '  Batch #%d: %d reviews (last review_id=%d)%s',
                $batchNumber,
                count($batchRows),
                $lastId,
                $dryRun ? ' [skipped – dry-run]' : ($sync ? ' [processed]' : ' [enqueued]'),
            ));

            Log::info('MigrateAsteriaReviews: batch '.$batchNumber.', '.count($batchRows).' rows, last_id='.$lastId);

        } while ($rows->count() === $batchSize);

        // ── Summary ───────────────────────────────────────────────────────────────
        $this->newLine();
        $this->info("Done. Batches: {$batchNumber}, reviews processed/enqueued: {$totalDispatched}, unmatched: {$totalSkipped}.");

        if (! $dryRun && ! $sync) {
            $this->line('Run  <comment>php artisan queue:work --queue=review-migration</comment>  to process the queue.');
        }

        return self::SUCCESS;
    }

    // ─────────────────────────────────────────────────────────────────────────────
    // Private helpers
    // ─────────────────────────────────────────────────────────────────────────────

    /**
     * Fetch one page of Magento reviews, enriched with rating and customer e-mail.
     *
     * Magento table names use the default prefix (none).  If your Magento
     * installation uses a prefix (e.g. `mag_`), set ASTERIA_DB_PREFIX in .env and
     * pass it via the `prefix` key in the database connection config.
     *
     * @return \Illuminate\Support\Collection<int, object>
     */
    private function fetchBatch(
        string $connection,
        int $lastId,
        int $limit,
        ?int $statusId
    ): \Illuminate\Support\Collection {
        $query = DB::connection($connection)
            ->table('review AS r')
            ->join('review_detail AS rd', 'rd.review_id', '=', 'r.review_id')
            ->leftJoin('customer_entity AS ce', 'ce.entity_id', '=', 'rd.customer_id')
            ->leftJoinSub(
                // Average all rating dimensions for each review (Quality, Value, Price…)
                DB::connection($connection)
                    ->table('rating_option_vote')
                    ->select('review_id', DB::raw('AVG(`value`) AS avg_rating'))
                    ->groupBy('review_id'),
                'rv',
                'rv.review_id',
                '=',
                'r.review_id'
            )
            ->select([
                'r.review_id',
                'r.created_at',
                'r.entity_pk_value AS magento_product_id',
                'r.status_id',
                'rd.title',
                'rd.detail AS comment',
                'rd.nickname AS name',
                'rd.customer_id AS magento_customer_id',
                'ce.email AS customer_email',
                'rv.avg_rating',
            ])
            ->where('r.entity_id', 1)   // entity_id=1 means "product" in Magento
            ->where('r.review_id', '>', $lastId)
            ->orderBy('r.review_id')
            ->limit($limit);

        if ($statusId !== null) {
            $query->where('r.status_id', $statusId);
        }

        $reviews = $query->get();

        if ($reviews->isEmpty()) {
            return $reviews;
        }

        // Attach image URLs from Senje_Review's review_media_image table (status_id=1 = approved).
        $reviewIds = $reviews->pluck('review_id')->toArray();

        $imageRows = DB::connection($connection)
            ->table('review_media_image')
            ->select('review_id', 'image_id', 'url')
            ->whereIn('review_id', $reviewIds)
            ->where('status_id', 1)
            ->orderBy('review_id')
            ->orderBy('image_id')
            ->get()
            ->groupBy('review_id');

        return $reviews->map(function ($row) use ($imageRows) {
            $row->images = $imageRows->get($row->review_id, collect())
                ->pluck('url')
                ->toArray();

            return $row;
        });
    }

    /**
     * Build magento_product_id → bagisto_product_id.
     *
     * Prefer products.migrated_from_asteria_id, then fall back to SKU.
     * Does not filter Magento or Bagisto product status / inventory.
     *
     * @return array<int, int>
     */
    private function buildProductIdMap(string $connection): array
    {
        $magentoRows = DB::connection($connection)
            ->table('catalog_product_entity')
            ->select('entity_id', 'sku')
            ->get();

        if ($magentoRows->isEmpty()) {
            return [];
        }

        $map = [];

        // 1) Match by migrated_from_asteria_id (covers disabled / out-of-stock Bagisto rows).
        if (Schema::hasColumn('products', 'migrated_from_asteria_id')) {
            $asteriaIds = $magentoRows->pluck('entity_id')->map(fn ($id) => (int) $id)->all();

            foreach (array_chunk($asteriaIds, 500) as $chunk) {
                $rows = DB::table('products')
                    ->select('id', 'migrated_from_asteria_id')
                    ->whereIn('migrated_from_asteria_id', $chunk)
                    ->get();

                foreach ($rows as $row) {
                    $map[(int) $row->migrated_from_asteria_id] = (int) $row->id;
                }
            }
        }

        // 2) Fall back to SKU for Magento products not yet linked by Asteria id.
        $magentoPairs = [];

        foreach ($magentoRows as $row) {
            $magentoId = (int) $row->entity_id;
            $sku = trim((string) $row->sku);

            if ($sku === '' || isset($map[$magentoId])) {
                continue;
            }

            $magentoPairs[$sku] = $magentoId;
        }

        if ($magentoPairs === []) {
            return $map;
        }

        foreach (array_chunk(array_keys($magentoPairs), 500) as $chunk) {
            $rows = DB::table('products')
                ->select('id', 'sku')
                ->whereIn('sku', $chunk)
                ->get();

            foreach ($rows as $row) {
                $magentoId = $magentoPairs[$row->sku] ?? null;

                if ($magentoId !== null && ! isset($map[$magentoId])) {
                    $map[$magentoId] = (int) $row->id;
                }
            }
        }

        return $map;
    }

    /**
     * Build a map of  lowercase_email → bagisto_customer_id.
     *
     * @return array<string, int>
     */
    private function buildCustomerEmailMap(string $connection): array
    {
        // Fetch Magento customer e-mails.
        $magentoEmails = DB::connection($connection)
            ->table('customer_entity')
            ->select('email')
            ->get()
            ->pluck('email')
            ->map(fn ($e) => strtolower((string) $e))
            ->unique()
            ->toArray();

        if (empty($magentoEmails)) {
            return [];
        }

        $map = [];

        foreach (array_chunk($magentoEmails, 500) as $chunk) {
            $rows = DB::table('customers')
                ->select('id', 'email')
                ->whereIn(DB::raw('LOWER(email)'), $chunk)
                ->get();

            foreach ($rows as $row) {
                $map[strtolower($row->email)] = (int) $row->id;
            }
        }

        return $map;
    }

    /**
     * Ensure product_reviews has the deduplication column.
     * Prints instructions and returns false if the column is absent and
     * cannot be auto-created (requires ALTER TABLE permission).
     */
    private function ensureMigrationColumnExists(): bool
    {
        $hasColumn = Schema::hasColumn('product_reviews', 'migrated_from_asteria_id');

        if ($hasColumn) {
            return true;
        }

        $this->warn('Column product_reviews.migrated_from_asteria_id is missing.');
        $this->line('Run the following migration or SQL:');
        $this->line('');
        $this->line('  <comment>php artisan make:migration add_migrated_from_asteria_id_to_product_reviews</comment>');
        $this->line('  Then add to the up() method:');
        $this->line('  <comment>$table->bigInteger(\'migrated_from_asteria_id\')->unsigned()->nullable()->unique();</comment>');
        $this->line('');
        $this->line('  Or directly in MySQL:');
        $this->line('  <comment>ALTER TABLE product_reviews');
        $this->line('    ADD COLUMN migrated_from_asteria_id BIGINT UNSIGNED NULL DEFAULT NULL,');
        $this->line('    ADD UNIQUE INDEX uq_migrated_asteria_id (migrated_from_asteria_id);</comment>');
        $this->line('');

        if (! $this->confirm('Attempt to add the column automatically now?', false)) {
            return false;
        }

        try {
            DB::statement('
                ALTER TABLE product_reviews
                    ADD COLUMN migrated_from_asteria_id BIGINT UNSIGNED NULL DEFAULT NULL,
                    ADD UNIQUE INDEX uq_migrated_asteria_id (migrated_from_asteria_id)
            ');
            $this->info('Column added successfully.');

            return true;
        } catch (\Throwable $e) {
            $this->error('Could not add column automatically: '.$e->getMessage());

            return false;
        }
    }
}