<?php

declare(strict_types=1);

namespace App\Console\Commands;

use App\Support\RussianRegions;
use Illuminate\Console\Command;
use Illuminate\Database\Connection;
use Illuminate\Support\Facades\DB;
use OpenSpout\Reader\XLSX\Reader as XlsxReader;

/**
 * Импорт реестра нумерации Россвязи в `phone_ranges` (spec §6).
 *
 *   php artisan phone-ranges:import --file=<csv|xlsx> [--force] [--dry-run]
 *   php artisan phone-ranges:import --dir=<dir с пакетом файлов> [...]
 *
 * Алгоритм:
 *   1. Резолв входных файлов (--file | --dir; --url отложен — оператор качает пакет вручную).
 *   2. Checksum-идемпотентность: совпал с предыдущим `completed` → status='rolled_back', выход.
 *   3. Парсинг (CSV через str_getcsv ';', XLSX через openspout) → нормализованные строки.
 *   4. Маппинг region → subject_code через RussianRegions::nameToCode(). Несматчившиеся → лог в error.
 *   5. Сборка `phone_ranges_staging` (LIKE phone_ranges) + bulk INSERT.
 *   6. --dry-run → staging остаётся для инспекции, swap НЕ делается, status='rolled_back'.
 *      Иначе → atomic RENAME swap + status='completed'.
 *
 * Запись идёт через `pgsql_supplier` (на проде crm_supplier_worker — член crm_migrator,
 * INHERIT даёт CREATE; SET ROLE crm_migrator выравнивает ownership. На dev/test — postgres superuser).
 *
 * NB (swap — operator-validated): committing-swap (шаг 6 else) НЕ покрыт автотестом —
 * RENAME коммитит и сломал бы общую тестовую БД. Свап проверяется первым реальным
 * импортом оператора по runbook (Session 6). Тесты покрывают parse/map/dry-run/idempotency.
 */
class PhoneRangesImportCommand extends Command
{
    /** @var string */
    protected $signature = 'phone-ranges:import
        {--file= : Путь к одному CSV/XLSX файлу реестра}
        {--dir= : Каталог с пакетом файлов реестра (*.csv, *.xlsx)}
        {--url= : (отложено) URL пакета — скачать вручную и использовать --dir}
        {--force : Игнорировать checksum-идемпотентность}
        {--dry-run : Распарсить и собрать staging, но не делать atomic swap}';

    /** @var string */
    protected $description = 'Импорт реестра нумерации Россвязи в phone_ranges (idempotent, atomic swap)';

    /** Connection для DDL/записи (на проде crm_migrator-capable, на dev/test — superuser fallback). */
    private const DDL_CONNECTION = 'pgsql_supplier';

    /** Размер пачки для bulk INSERT в staging. */
    private const INSERT_CHUNK = 1000;

    public function handle(): int
    {
        $files = $this->resolveFiles();
        if ($files === null) {
            return self::FAILURE;
        }

        $checksum = $this->computeChecksum($files);
        $dryRun = (bool) $this->option('dry-run');
        $force = (bool) $this->option('force');

        // 2. Идемпотентность по checksum (если не --force).
        if (! $force) {
            $prev = DB::table('phone_ranges_imports')
                ->where('checksum_sha256', $checksum)
                ->where('status', 'completed')
                ->orderByDesc('id')
                ->first();

            if ($prev !== null) {
                DB::table('phone_ranges_imports')->insert([
                    'source_url' => $this->sourceLabel($files),
                    'checksum_sha256' => $checksum,
                    'status' => 'rolled_back',
                    'rows_inserted' => 0,
                    'rows_updated' => 0,
                    'error' => "Идентично импорту #{$prev->id} (checksum совпал) — пропуск.",
                    'imported_at' => now(),
                    'completed_at' => now(),
                ]);
                $this->info("Реестр идентичен импорту #{$prev->id} — пропуск (используйте --force для принудительного импорта).");

                return self::SUCCESS;
            }
        }

        // 3. Журнал импорта (in_progress).
        $importId = (int) DB::table('phone_ranges_imports')->insertGetId([
            'source_url' => $this->sourceLabel($files),
            'checksum_sha256' => $checksum,
            'status' => 'in_progress',
            'imported_at' => now(),
        ]);

        try {
            // 4. Парсинг + маппинг.
            $unmatched = [];
            $rows = [];
            foreach ($files as $file) {
                foreach ($this->parseFile($file) as $rec) {
                    $regionNormalized = RussianRegions::canonicalRegionName($rec['region']);
                    $subjectCode = $regionNormalized === null
                        ? null
                        : (RussianRegions::nameToCode()[$regionNormalized] ?? null);
                    if ($subjectCode === null && trim($rec['region']) !== '') {
                        $unmatched[trim($rec['region'])] = true;
                    }
                    $rows[] = [
                        'def_code' => $rec['def_code'],
                        'from_num' => $rec['from_num'],
                        'to_num' => $rec['to_num'],
                        'operator' => $rec['operator'],
                        'region' => $rec['region'],
                        'region_normalized' => $regionNormalized,
                        'subject_code' => $subjectCode,
                        'imported_at' => now(),
                        'import_id' => $importId,
                    ];
                }
            }

            // 5. Сборка staging.
            $this->buildStaging($rows, $importId);

            $unmatchedNote = $unmatched === []
                ? ''
                : 'Не сопоставлены регионы: '.implode(', ', array_keys($unmatched)).'.';

            if ($dryRun) {
                DB::table('phone_ranges_imports')->where('id', $importId)->update([
                    'status' => 'rolled_back',
                    'rows_inserted' => count($rows),
                    'error' => trim('dry-run (swap не выполнен). '.$unmatchedNote),
                    'completed_at' => now(),
                ]);
                $this->info('dry-run: '.count($rows).' строк в phone_ranges_staging, swap не выполнен.');
                if ($unmatchedNote !== '') {
                    $this->warn($unmatchedNote);
                }

                return self::SUCCESS;
            }

            // 6. Atomic swap (operator-validated — см. docblock).
            $this->atomicSwap();

            DB::table('phone_ranges_imports')->where('id', $importId)->update([
                'status' => 'completed',
                'rows_inserted' => count($rows),
                'error' => $unmatchedNote !== '' ? $unmatchedNote : null,
                'completed_at' => now(),
            ]);
            $this->info('Импортировано '.count($rows).' строк в phone_ranges (atomic swap выполнен).');
            if ($unmatchedNote !== '') {
                $this->warn($unmatchedNote);
            }

            return self::SUCCESS;
        } catch (\Throwable $e) {
            DB::table('phone_ranges_imports')->where('id', $importId)->update([
                'status' => 'failed',
                'error' => mb_substr($e->getMessage(), 0, 2000),
                'completed_at' => now(),
            ]);
            $this->error('Импорт упал: '.$e->getMessage());

            return self::FAILURE;
        }
    }

    /**
     * @return list<string>|null Список файлов или null при ошибке валидации опций.
     */
    private function resolveFiles(): ?array
    {
        $file = $this->option('file');
        $dir = $this->option('dir');
        $url = $this->option('url');

        if ($url !== null) {
            $this->error('--url отложен (пакет ~500-600 файлов). Скачайте вручную и используйте --dir.');

            return null;
        }

        if ($file !== null) {
            if (! is_file($file)) {
                $this->error("Файл не найден: {$file}");

                return null;
            }

            return [$file];
        }

        if ($dir !== null) {
            if (! is_dir($dir)) {
                $this->error("Каталог не найден: {$dir}");

                return null;
            }
            $found = glob(rtrim($dir, '/\\').DIRECTORY_SEPARATOR.'*.{csv,xlsx}', GLOB_BRACE) ?: [];
            if ($found === []) {
                $this->error("В каталоге нет *.csv / *.xlsx: {$dir}");

                return null;
            }
            sort($found);

            return array_values($found);
        }

        $this->error('Укажите --file=<путь> или --dir=<каталог>.');

        return null;
    }

    /**
     * @param  list<string>  $files
     */
    private function computeChecksum(array $files): string
    {
        if (count($files) === 1) {
            return (string) hash_file('sha256', $files[0]);
        }

        $hashes = array_map(static fn (string $f): string => (string) hash_file('sha256', $f), $files);
        sort($hashes);

        return hash('sha256', implode('|', $hashes));
    }

    /**
     * @param  list<string>  $files
     */
    private function sourceLabel(array $files): string
    {
        return $this->option('url')
            ?? $this->option('dir')
            ?? ($files[0] ?? 'unknown');
    }

    /**
     * Парсит один файл реестра в нормализованные строки.
     *
     * @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
     */
    private function parseFile(string $path): array
    {
        $ext = strtolower(pathinfo($path, PATHINFO_EXTENSION));

        return $ext === 'xlsx'
            ? $this->parseXlsx($path)
            : $this->parseCsv($path);
    }

    /**
     * @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
     */
    private function parseCsv(string $path): array
    {
        $content = (string) file_get_contents($path);
        // BOM strip + split строк (CRLF/CR/LF).
        $content = preg_replace('/^\xEF\xBB\xBF/', '', $content) ?? $content;
        $lines = preg_split('/\r\n|\r|\n/', rtrim($content)) ?: [];
        if ($lines === []) {
            return [];
        }

        $header = str_getcsv((string) array_shift($lines), ';');
        $cols = $this->resolveColumns($header);

        $out = [];
        foreach ($lines as $line) {
            if (trim($line) === '') {
                continue;
            }
            $cells = str_getcsv($line, ';');
            $rec = $this->mapCells($cells, $cols);
            if ($rec !== null) {
                $out[] = $rec;
            }
        }

        return $out;
    }

    /**
     * Парсинг XLSX через openspout (operator-real-files; CSV-ветка покрыта тестом).
     *
     * @return list<array{def_code:int, from_num:int, to_num:int, operator:string, region:string}>
     */
    private function parseXlsx(string $path): array
    {
        $reader = new XlsxReader;
        $reader->open($path);

        $out = [];
        $cols = null;
        foreach ($reader->getSheetIterator() as $sheet) {
            foreach ($sheet->getRowIterator() as $row) {
                $cells = array_map(static fn ($c): string => (string) $c, $row->toArray());
                if ($cols === null) {
                    $cols = $this->resolveColumns($cells);

                    continue;
                }
                $rec = $this->mapCells($cells, $cols);
                if ($rec !== null) {
                    $out[] = $rec;
                }
            }
            break; // только первый лист
        }
        $reader->close();

        return $out;
    }

    /**
     * Сопоставляет индексы колонок по заголовку (русские имена Россвязи) с позиционным fallback.
     *
     * @param  list<string>  $header
     * @return array{def:int, from:int, to:int, operator:int, region:int}
     */
    private function resolveColumns(array $header): array
    {
        $cols = ['def' => 0, 'from' => 1, 'to' => 2, 'operator' => 4, 'region' => 5];

        foreach ($header as $i => $cell) {
            $n = preg_replace('/[\s\/]+/u', '', mb_strtolower(trim((string) $cell))) ?? '';
            if (str_contains($n, 'def') || str_contains($n, 'авс')) {
                $cols['def'] = $i;
            } elseif ($n === 'от') {
                $cols['from'] = $i;
            } elseif ($n === 'до') {
                $cols['to'] = $i;
            } elseif (str_contains($n, 'оператор')) {
                $cols['operator'] = $i;
            } elseif (str_contains($n, 'регион')) {
                $cols['region'] = $i;
            }
        }

        return $cols;
    }

    /**
     * @param  list<string>  $cells
     * @param  array{def:int, from:int, to:int, operator:int, region:int}  $cols
     * @return array{def_code:int, from_num:int, to_num:int, operator:string, region:string}|null
     */
    private function mapCells(array $cells, array $cols): ?array
    {
        $def = (int) preg_replace('/\D+/', '', $cells[$cols['def']] ?? '');
        if ($def === 0) {
            return null; // пустая/битая строка
        }

        return [
            'def_code' => $def,
            'from_num' => (int) preg_replace('/\D+/', '', $cells[$cols['from']] ?? '0'),
            'to_num' => (int) preg_replace('/\D+/', '', $cells[$cols['to']] ?? '0'),
            'operator' => trim((string) ($cells[$cols['operator']] ?? '')),
            'region' => trim((string) ($cells[$cols['region']] ?? '')),
        ];
    }

    /**
     * Собирает phone_ranges_staging (LIKE phone_ranges) и заливает строки.
     *
     * id: НЕ копируем серийный default через INCLUDING DEFAULTS — он ссылается на
     * исходную последовательность phone_ranges, которую atomic-swap уничтожает
     * (DROP phone_ranges_old CASCADE) после первого импорта, оставляя staging.id
     * без default (NOT NULL violation на повторном импорте). Вместо этого даём
     * staging собственную последовательность с уникальным по import_id именем,
     * OWNED BY колонкой id → она переезжает при RENAME и дропается вместе со
     * старой таблицей (без коллизий имён и без утечки последовательностей).
     *
     * @param  list<array<string, mixed>>  $rows
     */
    private function buildStaging(array $rows, int $importId): void
    {
        $c = DB::connection(self::DDL_CONNECTION);
        $this->elevate($c);

        $seq = "phone_ranges_stg_seq_{$importId}";
        $c->statement('DROP TABLE IF EXISTS phone_ranges_staging CASCADE');
        $c->statement('CREATE TABLE phone_ranges_staging (LIKE phone_ranges INCLUDING CONSTRAINTS)');
        $c->statement("CREATE SEQUENCE {$seq}");
        $c->statement("ALTER TABLE phone_ranges_staging ALTER COLUMN id SET DEFAULT nextval('{$seq}')");
        $c->statement("ALTER SEQUENCE {$seq} OWNED BY phone_ranges_staging.id");
        $c->statement('CREATE INDEX IF NOT EXISTS idx_phone_ranges_staging_lookup ON phone_ranges_staging (def_code, from_num, to_num)');

        foreach (array_chunk($rows, self::INSERT_CHUNK) as $chunk) {
            $c->table('phone_ranges_staging')->insert($chunk);
        }
    }

    /**
     * Atomic swap живого phone_ranges на staging (spec §6.2 шаг 6).
     *
     * NB: НЕ покрыт автотестом (committing RENAME сломал бы общую тестовую БД).
     * Проверяется первым реальным импортом оператора (Session 6 runbook).
     * Сохраняет одну предыдущую версию (phone_ranges_old) для `phone-ranges:rollback`.
     * GRANT'ы переустанавливаются (RENAME их не переносит); lookup-индекс на новой
     * таблице носит имя idx_phone_ranges_staging_lookup (косметика — имя занято _old).
     */
    private function atomicSwap(): void
    {
        $c = DB::connection(self::DDL_CONNECTION);
        $this->elevate($c);

        // Транзакция вокруг свапа (spec §6.2): PostgreSQL поддерживает транзакционный
        // DDL, поэтому DROP+RENAME+RENAME+GRANT атомарны. Обрыв процесса между
        // переименованиями не оставит phone_ranges несуществующей — откат вернёт
        // живую таблицу (раньше 4 авто-коммит-statement'а оставляли окно, в котором
        // Россвязь-lookup падал бы до ручного восстановления).
        $c->transaction(function () use ($c) {
            $c->statement('DROP TABLE IF EXISTS phone_ranges_old CASCADE');
            $c->statement('ALTER TABLE phone_ranges RENAME TO phone_ranges_old');
            $c->statement('ALTER TABLE phone_ranges_staging RENAME TO phone_ranges');
            $c->statement('GRANT SELECT ON phone_ranges TO crm_app_user, crm_supplier_worker');
        });
    }

    /**
     * SET ROLE crm_migrator для корректного ownership на проде; на dev/test роль
     * отсутствует → RESET и работаем как superuser (зеркало миграционного паттерна).
     */
    private function elevate(Connection $c): void
    {
        try {
            $c->statement('SET ROLE crm_migrator');
            $canCreate = $c->selectOne("SELECT has_schema_privilege('crm_migrator', 'public', 'CREATE') AS ok");
            if (! $canCreate || ! $canCreate->ok) {
                $c->statement('RESET ROLE');
            }
        } catch (\Throwable) {
            // окружение без роли — продолжаем как superuser
        }
    }
}
