nshop/app/Console/Commands/MigrateAsteriaCustomers.php

<?php

namespace App\Console\Commands;

use App\Services\Asteria\Magento1CustomerReader;
use App\Services\Asteria\MagentoPrimaryKey;
use Illuminate\Console\Command;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Hash;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Str;
use Webkul\Customer\Models\CustomerAddress;

/**
 * Migrates storefront customers, addresses, and newsletter subscriptions from Asteria (Magento 1.x).
 *
 * Usage
 * ─────
 * php artisan customers:migrate-asteria
 * php artisan customers:migrate-asteria --batch-size=500
 * php artisan customers:migrate-asteria --reset-progress
 * php artisan customers:migrate-asteria --dry-run
 */
class MigrateAsteriaCustomers extends Command
{
    protected $signature = 'customers:migrate-asteria
        {--batch-size=500     : Number of Magento customers per batch}
        {--reset-progress     : Ignore saved progress and start from entity_id=0}
        {--dry-run            : Count records without writing}
        {--connection=asteria : Laravel DB connection for the Asteria database}';

    protected $description = 'Migrate Asteria (Magento 1.x) customers, addresses, and newsletter subscriptions into Bagisto';

    private const MAGENTO_NEWSLETTER_SUBSCRIBED = 1;

    private const PROGRESS_KEY = 'migrate_asteria_customers_last_id';

    private const INSERT_CHUNK = 200;

    private int $groupId;

    private mixed $channelId;

    private string $placeholderPassword;

    /** @var array<string, int> */
    private array $usedPhones = [];

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

        DB::disableQueryLog();

        try {
            DB::connection($connection)->getPdo();
        } catch (\Throwable $e) {
            $this->error("Cannot connect to Asteria DB (connection='{$connection}'): ".$e->getMessage());

            return self::FAILURE;
        }

        foreach (['customer_entity', 'eav_attribute'] as $table) {
            if (! Schema::connection($connection)->hasTable($table)) {
                $this->error("Asteria table '{$table}' is missing on connection '{$connection}'.");

                return self::FAILURE;
            }
        }

        if (! Schema::hasColumn('customers', 'migrated_from_asteria_id')
            || ! Schema::hasColumn('customers', 'legacy_password')
            || ! Schema::hasColumn('customers', 'customer_source')) {
            $this->error('customers.migrated_from_asteria_id / legacy_password / customer_source are missing. Run php artisan migrate.');

            return self::FAILURE;
        }

        $groupId = DB::table('customer_groups')->where('code', 'general')->value('id');

        if (! $groupId) {
            $this->error("Bagisto customer group 'general' was not found.");

            return self::FAILURE;
        }

        $this->groupId = (int) $groupId;
        $this->channelId = core()->getDefaultChannel()?->id ?? core()->getCurrentChannel()?->id;
        $this->placeholderPassword = Hash::make(Str::random(32));

        $reader = new Magento1CustomerReader($connection);
        $lastId = $resetProgress ? 0 : (int) Cache::get(self::PROGRESS_KEY, 0);

        if ($resetProgress) {
            Cache::forget(self::PROGRESS_KEY);
        }

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

        $this->usedPhones = DB::table('customers')
            ->whereNotNull('phone')
            ->where('phone', '!=', '')
            ->pluck('phone')
            ->map(fn ($phone) => mb_strtolower((string) $phone))
            ->flip()
            ->all();

        $created = 0;
        $linked = 0;
        $skipped = 0;
        $addressesImported = 0;
        $batchNumber = 0;

        $this->info($dryRun ? '[DRY RUN] Scanning Magento customers…' : 'Migrating Magento customers…');

        do {
            $started = microtime(true);
            $customers = $reader->fetchCustomers($lastId, $batchSize);

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

            $batchNumber++;
            $lastId = (int) $customers->max('entity_id');
            $asteriaIds = $customers->pluck('entity_id')->map(fn ($id) => (int) $id)->all();
            $emails = $customers
                ->pluck('email')
                ->map(fn ($email) => strtolower(trim((string) $email)))
                ->filter()
                ->all();
            $addresses = $reader->fetchAddresses($asteriaIds)->groupBy(fn (array $row) => (int) $row['parent_id']);
            $subscribers = $reader->fetchSubscribers($asteriaIds, $emails);

            if ($dryRun) {
                $created += $customers->count();
                $addressesImported += $addresses->flatten(1)->count();
                $this->line(sprintf(
                    '  Batch #%d: %d customers, %d addresses (last entity_id=%d) [skipped – dry-run] (%.1fs)',
                    $batchNumber,
                    $customers->count(),
                    $addresses->flatten(1)->count(),
                    $lastId,
                    microtime(true) - $started
                ));

                continue;
            }

            $result = $this->persistBatch($customers, $addresses, $subscribers);

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

            $created += $result['created'];
            $linked += $result['linked'];
            $skipped += $result['skipped'];
            $addressesImported += $result['addresses'];

            $this->line(sprintf(
                '  Batch #%d: created=%d linked=%d skipped=%d addresses=%d (last entity_id=%d) (%.1fs)',
                $batchNumber,
                $result['created'],
                $result['linked'],
                $result['skipped'],
                $result['addresses'],
                $lastId,
                microtime(true) - $started
            ));

            Log::info('MigrateAsteriaCustomers: batch '.$batchNumber.', last_id='.$lastId);
        } while ($customers->count() === $batchSize);

        $subscriptions = $this->importAllNewsletterSubscribers($reader, $batchSize, $dryRun);

        $this->newLine();
        $this->info("Done. Batches: {$batchNumber}, created: {$created}, linked: {$linked}, skipped: {$skipped}, addresses: {$addressesImported}, subscriptions: {$subscriptions}.");

        if (! $dryRun) {
            MagentoPrimaryKey::bumpAutoIncrement('customers');
        }

        return self::SUCCESS;
    }

    /**
     * @param  Collection<int, array<string, mixed>>  $customers
     * @param  Collection<int, Collection<int, array<string, mixed>>>  $addresses
     * @param  Collection<int, array<string, mixed>>  $subscribers
     * @return array{created: int, linked: int, skipped: int, addresses: int}
     */
    private function persistBatch(Collection $customers, Collection $addresses, Collection $subscribers): array
    {
        $now = now()->format('Y-m-d H:i:s');
        $created = 0;
        $linked = 0;
        $skipped = 0;
        $importedAddresses = 0;
        [$subscribersByCustomerId, $subscribersByEmail] = $this->indexLatestSubscribers($subscribers);

        $asteriaIds = $customers->pluck('entity_id')->map(fn ($id) => (int) $id)->unique()->values()->all();
        $emails = $customers
            ->pluck('email')
            ->map(fn ($email) => strtolower(trim((string) $email)))
            ->filter(fn ($email) => $email !== '' && filter_var($email, FILTER_VALIDATE_EMAIL))
            ->unique()
            ->values()
            ->all();

        [$byAsteriaId, $byEmail] = $this->loadExistingCustomers($asteriaIds, $emails);
        $occupiedIds = MagentoPrimaryKey::occupiedIds('customers', $asteriaIds);

        $insertCustomers = [];
        $linkUpdates = [];
        $sourceUpdates = [];
        $addressJobs = [];
        $seenEmails = [];

        foreach ($customers as $row) {
            $email = strtolower(trim((string) ($row['email'] ?? '')));
            $asteriaId = (int) $row['entity_id'];

            if ($email === '' || ! filter_var($email, FILTER_VALIDATE_EMAIL)) {
                $skipped++;

                continue;
            }

            $existing = $byAsteriaId[$asteriaId] ?? $byEmail[$email] ?? null;

            if ($existing || isset($seenEmails[$email])) {
                if ($existing && empty($existing->migrated_from_asteria_id)) {
                    $linkUpdates[(int) $existing->id] = $asteriaId;
                    $existing->migrated_from_asteria_id = $asteriaId;
                    $byAsteriaId[$asteriaId] = $existing;
                }

                if ($existing) {
                    $addressJobs[] = ['customer' => $existing, 'row' => $row];
                    $linked++;
                    $source = $this->mapCustomerSource($row['source'] ?? $row['customer_source'] ?? null);

                    if ($source !== null) {
                        $sourceUpdates[(int) $existing->id] = $source;
                    }
                } else {
                    $skipped++;
                }

                continue;
            }

            if (isset($occupiedIds[$asteriaId])) {
                $skipped++;
                $this->warn("Customer {$email} skipped: customers.id {$asteriaId} is already occupied.");
                Log::warning("MigrateAsteriaCustomers: customers.id {$asteriaId} already occupied; skipping Magento customer {$asteriaId} ({$email}).");

                continue;
            }

            $seenEmails[$email] = true;
            $phone = $this->uniquePhone((string) ($row['telephone'] ?? ''), $this->usedPhones);
            $subscriber = $subscribersByCustomerId[$asteriaId] ?? $subscribersByEmail[$email] ?? null;

            $insertCustomers[$asteriaId] = [
                'id'                        => $asteriaId,
                'migrated_from_asteria_id'  => $asteriaId,
                'first_name'                => $this->requiredName($row['firstname'] ?? null, $email),
                'last_name'                 => trim((string) ($row['lastname'] ?? '')) ?: '-',
                'gender'                    => $this->mapGender($row['gender'] ?? null),
                'date_of_birth'             => $this->mapDate($row['dob'] ?? null),
                'email'                     => $email,
                'phone'                     => $phone,
                'password'                  => $this->placeholderPassword,
                'legacy_password'           => $this->nullableString($row['password_hash'] ?? null),
                'api_token'                 => Str::random(80),
                'customer_group_id'         => $this->groupId,
                'channel_id'                => $this->channelId,
                'customer_source'           => $this->mapCustomerSource($row['source'] ?? $row['customer_source'] ?? null),
                'subscribed_to_news_letter' => $this->isMagentoSubscribed($subscriber) ? 1 : 0,
                'status'                    => ((int) ($row['is_active'] ?? 1)) === 1 ? 1 : 0,
                'is_verified'               => 1,
                'is_suspended'              => 0,
                'created_at'                => ! empty($row['created_at']) ? $row['created_at'] : $now,
                'updated_at'                => $now,
            ];

            $addressJobs[] = ['asteria_id' => $asteriaId, 'row' => $row, 'email' => $email, 'first_name' => $insertCustomers[$asteriaId]['first_name'], 'last_name' => $insertCustomers[$asteriaId]['last_name'], 'phone' => $phone];
            $created++;
        }

        DB::transaction(function () use (
            $insertCustomers,
            $linkUpdates,
            $sourceUpdates,
            $addressJobs,
            $addresses,
            $subscribers,
            &$importedAddresses
        ) {
            if ($insertCustomers !== []) {
                $this->insertRows('customers', array_values($insertCustomers));

                $inserted = DB::table('customers')
                    ->whereIn('migrated_from_asteria_id', array_keys($insertCustomers))
                    ->get(['id', 'migrated_from_asteria_id', 'email', 'first_name', 'last_name', 'phone']);

                $byInsertedAsteriaId = $inserted->keyBy(fn ($row) => (int) $row->migrated_from_asteria_id);
            } else {
                $byInsertedAsteriaId = collect();
            }

            foreach ($linkUpdates as $customerId => $asteriaId) {
                DB::table('customers')
                    ->where('id', $customerId)
                    ->whereNull('migrated_from_asteria_id')
                    ->update(['migrated_from_asteria_id' => $asteriaId]);
            }

            foreach ($sourceUpdates as $customerId => $source) {
                DB::table('customers')
                    ->where('id', $customerId)
                    ->update(['customer_source' => $source]);
            }

            $customerIdsForAddresses = [];

            foreach ($addressJobs as $job) {
                if (isset($job['customer'])) {
                    $customerIdsForAddresses[] = (int) $job['customer']->id;
                } elseif (isset($job['asteria_id'])) {
                    $inserted = $byInsertedAsteriaId->get((int) $job['asteria_id']);

                    if ($inserted) {
                        $customerIdsForAddresses[] = (int) $inserted->id;
                    }
                }
            }

            $existingAddressIds = $this->loadExistingAddressIds($customerIdsForAddresses);
            $addressInserts = [];

            foreach ($addressJobs as $job) {
                $row = $job['row'];
                $addressRows = $addresses->get((int) $row['entity_id'], collect());

                if ($addressRows->isEmpty()) {
                    continue;
                }

                if (isset($job['customer'])) {
                    $customerId = (int) $job['customer']->id;
                    $email = (string) $job['customer']->email;
                    $firstName = (string) $job['customer']->first_name;
                    $lastName = (string) $job['customer']->last_name;
                    $phone = $job['customer']->phone;
                } else {
                    $inserted = $byInsertedAsteriaId->get((int) $job['asteria_id']);

                    if (! $inserted) {
                        continue;
                    }

                    $customerId = (int) $inserted->id;
                    $email = (string) $inserted->email;
                    $firstName = (string) $inserted->first_name;
                    $lastName = (string) $inserted->last_name;
                    $phone = $inserted->phone;
                }

                $existingIds = $existingAddressIds[$customerId] ?? [];

                foreach ($this->buildAddressRows($customerId, $row, $addressRows, $email, $firstName, $lastName, $phone, $existingIds) as $addressRow) {
                    $addressInserts[] = $addressRow;
                    $importedAddresses++;
                }
            }

            if ($addressInserts !== []) {
                $this->insertRows('addresses', $addressInserts);
            }

            $this->persistSubscriberRows($subscribers);
        });

        return [
            'created'   => $created,
            'linked'    => $linked,
            'skipped'   => $skipped,
            'addresses' => $importedAddresses,
        ];
    }

    private function importAllNewsletterSubscribers(Magento1CustomerReader $reader, int $batchSize, bool $dryRun): int
    {
        if (! Schema::hasTable('subscribers_list')) {
            $this->warn('subscribers_list is missing; skipping newsletter import.');

            return 0;
        }

        $synced = 0;
        $lastId = 0;

        do {
            $rows = $reader->fetchSubscribersAfter($lastId, $batchSize);

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

            $lastId = (int) $rows->max('subscriber_id');
            $synced += $this->persistSubscriberRows($rows, $dryRun);
        } while ($rows->count() === $batchSize);

        return $synced;
    }

    /**
     * @param  Collection<int, array<string, mixed>>  $subscribers
     */
    private function persistSubscriberRows(Collection $subscribers, bool $dryRun = false): int
    {
        if ($subscribers->isEmpty()) {
            return 0;
        }

        [$byCustomerId, $byEmail] = $this->indexLatestSubscribers($subscribers);

        if ($byCustomerId === [] && $byEmail === []) {
            return 0;
        }

        [$customersByAsteriaId, $customersByEmail] = $this->loadCustomersForSubscribers(
            array_keys($byCustomerId),
            array_keys($byEmail)
        );

        $flagOn = [];
        $flagOff = [];

        foreach ($byCustomerId as $asteriaId => $sub) {
            $customer = $customersByAsteriaId[$asteriaId] ?? null;

            if (! $customer) {
                continue;
            }

            if ($this->isMagentoSubscribed($sub)) {
                $flagOn[(int) $customer->id] = true;
            } else {
                $flagOff[(int) $customer->id] = true;
            }
        }

        foreach ($byEmail as $email => $sub) {
            $customer = $customersByEmail[$email] ?? null;

            if (! $customer || isset($flagOn[(int) $customer->id]) || isset($flagOff[(int) $customer->id])) {
                continue;
            }

            if ($this->isMagentoSubscribed($sub)) {
                $flagOn[(int) $customer->id] = true;
            } else {
                $flagOff[(int) $customer->id] = true;
            }
        }

        $listRows = [];

        foreach ($byEmail as $email => $sub) {
            if ($email === '' || ! filter_var($email, FILTER_VALIDATE_EMAIL)) {
                continue;
            }

            $customer = $customersByEmail[$email] ?? null;

            if (! $customer) {
                $asteriaId = (int) ($sub['customer_id'] ?? 0);
                $customer = $asteriaId > 0 ? ($customersByAsteriaId[$asteriaId] ?? null) : null;
            }

            if (! $customer && ! $this->isMagentoSubscribed($sub)) {
                continue;
            }

            $listRows[$email] = [
                'email'         => $email,
                'is_subscribed' => $this->isMagentoSubscribed($sub) ? 1 : 0,
                'token'         => $this->nullableString($sub['subscriber_confirm_code'] ?? null),
                'customer_id'   => $customer ? (int) $customer->id : null,
                'channel_id'    => $customer?->channel_id ?? $this->channelId,
                'changed_at'    => $sub['change_status_at'] ?? null,
            ];
        }

        $synced = count($listRows);

        if ($dryRun) {
            return $synced;
        }

        $flagOffIds = array_keys(array_diff_key($flagOff, $flagOn));
        $flagOnIds = array_keys($flagOn);

        if ($flagOffIds !== []) {
            DB::table('customers')->whereIn('id', $flagOffIds)->update(['subscribed_to_news_letter' => 0]);
        }

        if ($flagOnIds !== []) {
            DB::table('customers')->whereIn('id', $flagOnIds)->update(['subscribed_to_news_letter' => 1]);
        }

        if ($listRows === []) {
            return 0;
        }

        $existing = $this->loadExistingSubscriberLists(array_keys($listRows));
        $now = now()->format('Y-m-d H:i:s');
        $inserts = [];

        foreach ($listRows as $email => $row) {
            $existingRow = $existing[$email] ?? null;

            if ($existingRow) {
                DB::table('subscribers_list')->where('id', $existingRow->id)->update([
                    'is_subscribed' => $row['is_subscribed'],
                    'customer_id'   => $row['customer_id'] ?? $existingRow->customer_id,
                    'updated_at'    => $now,
                ]);

                continue;
            }

            $inserts[] = [
                'email'         => $row['email'],
                'is_subscribed' => $row['is_subscribed'],
                'token'         => $row['token'] ?? uniqid(),
                'customer_id'   => $row['customer_id'],
                'channel_id'    => $row['channel_id'] ?? $this->channelId,
                'created_at'    => $this->mapDateTime($row['changed_at'] ?? null) ?? $now,
                'updated_at'    => $now,
            ];
        }

        if ($inserts !== []) {
            $this->insertRows('subscribers_list', $inserts);
        }

        return $synced;
    }

    /**
     * @param  Collection<int, array<string, mixed>>  $subscribers
     * @return array{0: array<int, array<string, mixed>>, 1: array<string, array<string, mixed>>}
     */
    private function indexLatestSubscribers(Collection $subscribers): array
    {
        $byCustomerId = [];
        $byEmail = [];

        foreach ($subscribers as $sub) {
            $customerId = (int) ($sub['customer_id'] ?? 0);
            $email = strtolower(trim((string) ($sub['subscriber_email'] ?? '')));

            if ($customerId > 0) {
                $byCustomerId[$customerId] = $this->newerSubscriber($byCustomerId[$customerId] ?? null, $sub);
            }

            if ($email !== '') {
                $byEmail[$email] = $this->newerSubscriber($byEmail[$email] ?? null, $sub);
            }
        }

        return [$byCustomerId, $byEmail];
    }

    /**
     * @param  array<string, mixed>|null  $current
     * @param  array<string, mixed>  $candidate
     * @return array<string, mixed>
     */
    private function newerSubscriber(?array $current, array $candidate): array
    {
        if ($current === null) {
            return $candidate;
        }

        $currentAt = (string) ($current['change_status_at'] ?? '');
        $candidateAt = (string) ($candidate['change_status_at'] ?? '');

        if ($candidateAt !== $currentAt) {
            return $candidateAt > $currentAt ? $candidate : $current;
        }

        return ((int) ($candidate['subscriber_id'] ?? 0)) >= ((int) ($current['subscriber_id'] ?? 0))
            ? $candidate
            : $current;
    }

    /**
     * @param  array<string, mixed>|null  $subscriber
     */
    private function isMagentoSubscribed(?array $subscriber): bool
    {
        return $subscriber !== null
            && ((int) ($subscriber['subscriber_status'] ?? 0)) === self::MAGENTO_NEWSLETTER_SUBSCRIBED;
    }

    /**
     * @param  array<int, int>  $asteriaIds
     * @param  array<int, string>  $emails
     * @return array{0: array<int, object>, 1: array<string, object>}
     */
    private function loadCustomersForSubscribers(array $asteriaIds, array $emails): array
    {
        $asteriaIds = array_values(array_unique(array_filter(array_map('intval', $asteriaIds))));
        $emails = array_values(array_unique(array_filter($emails)));

        $byAsteriaId = [];
        $byEmail = [];
        $select = ['id', 'email', 'channel_id', 'migrated_from_asteria_id'];

        if ($asteriaIds !== []) {
            foreach (DB::table('customers')->select($select)->whereIn('migrated_from_asteria_id', $asteriaIds)->get() as $customer) {
                $byAsteriaId[(int) $customer->migrated_from_asteria_id] = $customer;
                $byEmail[strtolower((string) $customer->email)] = $customer;
            }
        }

        if ($emails !== []) {
            foreach (DB::table('customers')->select($select)->whereIn(DB::raw('LOWER(email)'), $emails)->get() as $customer) {
                $byEmail[strtolower((string) $customer->email)] = $byEmail[strtolower((string) $customer->email)] ?? $customer;

                if ($customer->migrated_from_asteria_id) {
                    $byAsteriaId[(int) $customer->migrated_from_asteria_id] = $byAsteriaId[(int) $customer->migrated_from_asteria_id] ?? $customer;
                }
            }
        }

        return [$byAsteriaId, $byEmail];
    }

    /**
     * @param  array<int, string>  $emails
     * @return array<string, object>
     */
    private function loadExistingSubscriberLists(array $emails): array
    {
        $emails = array_values(array_unique(array_filter($emails)));

        if ($emails === []) {
            return [];
        }

        $byEmail = [];

        foreach (DB::table('subscribers_list')->whereIn(DB::raw('LOWER(email)'), $emails)->get(['id', 'email', 'customer_id', 'token']) as $row) {
            $byEmail[strtolower((string) $row->email)] = $row;
        }

        return $byEmail;
    }

    private function mapDateTime(mixed $value): ?string
    {
        $value = trim((string) $value);

        if ($value === '' || str_starts_with($value, '0000-00-00')) {
            return null;
        }

        return $value;
    }

    /**
     * @param  array<int, int>  $asteriaIds
     * @param  array<int, string>  $emails
     * @return array{0: array<int, object>, 1: array<string, object>}
     */
    private function loadExistingCustomers(array $asteriaIds, array $emails): array
    {
        $byAsteriaId = [];
        $byEmail = [];

        $select = ['id', 'email', 'first_name', 'last_name', 'phone', 'migrated_from_asteria_id', 'legacy_password'];

        if ($asteriaIds !== []) {
            foreach (DB::table('customers')->select($select)->whereIn('migrated_from_asteria_id', $asteriaIds)->get() as $customer) {
                $byAsteriaId[(int) $customer->migrated_from_asteria_id] = $customer;
                $byEmail[strtolower((string) $customer->email)] = $customer;
            }
        }

        if ($emails !== []) {
            foreach (DB::table('customers')->select($select)->whereIn(DB::raw('LOWER(email)'), $emails)->get() as $customer) {
                $byEmail[strtolower((string) $customer->email)] = $byEmail[strtolower((string) $customer->email)] ?? $customer;

                if ($customer->migrated_from_asteria_id) {
                    $byAsteriaId[(int) $customer->migrated_from_asteria_id] = $byAsteriaId[(int) $customer->migrated_from_asteria_id] ?? $customer;
                }
            }
        }

        return [$byAsteriaId, $byEmail];
    }

    /**
     * @param  array<int, int>  $customerIds
     * @return array<int, array<int, int>>
     */
    private function loadExistingAddressIds(array $customerIds): array
    {
        $customerIds = array_values(array_unique(array_filter($customerIds)));

        if ($customerIds === []) {
            return [];
        }

        $map = [];

        $rows = DB::table('addresses')
            ->where('address_type', CustomerAddress::ADDRESS_TYPE)
            ->whereIn('customer_id', $customerIds)
            ->get(['customer_id', 'additional']);

        foreach ($rows as $row) {
            $asteriaId = $this->asteriaAddressIdFromAdditional($row->additional);

            if ($asteriaId) {
                $map[(int) $row->customer_id][] = $asteriaId;
            }
        }

        return $map;
    }

    /**
     * @param  Collection<int, array<string, mixed>>  $addressRows
     * @param  array<int, int>  $existingIds
     * @return array<int, array<string, mixed>>
     */
    private function buildAddressRows(
        int $customerId,
        array $customerRow,
        $addressRows,
        string $email,
        string $firstName,
        string $lastName,
        mixed $phone,
        array $existingIds
    ): array {
        $defaultBilling = (int) ($customerRow['default_billing'] ?? 0);
        $defaultShipping = (int) ($customerRow['default_shipping'] ?? 0);
        $now = now()->format('Y-m-d H:i:s');
        $rows = [];

        foreach ($addressRows as $row) {
            $asteriaAddressId = (int) $row['entity_id'];

            if (in_array($asteriaAddressId, $existingIds, true)) {
                continue;
            }

            $rows[] = [
                'customer_id'      => $customerId,
                'address_type'     => CustomerAddress::ADDRESS_TYPE,
                'first_name'       => $this->requiredName($row['firstname'] ?? null, $firstName),
                'last_name'        => trim((string) ($row['lastname'] ?? '')) ?: $lastName,
                'company_name'     => $this->nullableString($row['company'] ?? null),
                'address'          => $this->mapStreet($row['street'] ?? null) ?: '-',
                'city'             => trim((string) ($row['city'] ?? '')) ?: '-',
                'state'            => $this->nullableString($row['region'] ?? null),
                'country'          => $this->nullableString($row['country_id'] ?? null),
                'postcode'         => $this->nullableString($row['postcode'] ?? null),
                'email'            => $email,
                'phone'            => $this->nullableString($row['telephone'] ?? null) ?? $phone,
                'default_address'  => $defaultBilling > 0 && $asteriaAddressId === $defaultBilling ? 1 : 0,
                'use_for_shipping' => $defaultShipping > 0 && $asteriaAddressId === $defaultShipping ? 1 : 0,
                'additional'       => json_encode(['asteria_address_id' => $asteriaAddressId]),
                'created_at'       => $now,
                'updated_at'       => $now,
            ];
        }

        return $rows;
    }

    /**
     * @param  array<int, array<string, mixed>>  $rows
     */
    private function insertRows(string $table, array $rows): void
    {
        foreach (array_chunk($rows, self::INSERT_CHUNK) as $chunk) {
            DB::table($table)->insert($chunk);
        }
    }

    /**
     * @param  array<string, int>  $usedPhones
     */
    private function uniquePhone(string $phone, array &$usedPhones): ?string
    {
        $phone = trim($phone);

        if ($phone === '') {
            return null;
        }

        $key = mb_strtolower($phone);

        if (isset($usedPhones[$key])) {
            return null;
        }

        $usedPhones[$key] = 1;

        return $phone;
    }

    private function mapGender(mixed $value): ?string
    {
        return match ((int) $value) {
            1       => 'Male',
            2       => 'Female',
            default => null,
        };
    }

    /**
     * Magento 1 stores `source` as a free-text varchar attribute (e.g. "popup"),
     * so the raw value is kept as-is rather than coerced to a code.
     */
    private function mapCustomerSource(mixed $value): ?string
    {
        if (! is_scalar($value)) {
            return null;
        }

        $value = trim((string) $value);

        return $value === '' ? null : mb_substr($value, 0, 64);
    }

    private function mapDate(mixed $value): ?string
    {
        $value = trim((string) $value);

        if ($value === '' || str_starts_with($value, '0000-00-00')) {
            return null;
        }

        return substr($value, 0, 10);
    }

    private function mapStreet(mixed $value): string
    {
        $value = trim((string) $value);

        if ($value === '') {
            return '';
        }

        $lines = preg_split("/\r\n|\n|\r/", $value) ?: [];

        return implode(', ', array_filter(array_map('trim', $lines)));
    }

    private function requiredName(mixed $value, string $fallback): string
    {
        $value = trim((string) $value);

        if ($value !== '') {
            return $value;
        }

        $local = strstr($fallback, '@', true);

        return $local !== false && $local !== '' ? $local : 'Customer';
    }

    private function nullableString(mixed $value): ?string
    {
        $value = trim((string) $value);

        return $value === '' ? null : $value;
    }

    private function asteriaAddressIdFromAdditional(mixed $additional): ?int
    {
        if (is_string($additional) && $additional !== '') {
            $additional = json_decode($additional, true);
        }

        if (! is_array($additional)) {
            return null;
        }

        return isset($additional['asteria_address_id'])
            ? (int) $additional['asteria_address_id']
            : null;
    }
}