<?php

declare(strict_types=1);

namespace App\Jobs;

use App\Exceptions\Billing\InsufficientBalanceException;
use App\Models\ActivityLog;
use App\Models\Deal;
use App\Models\Project;
use App\Models\SupplierLead;
use App\Models\Tenant;
use App\Services\Billing\LedgerService;
use App\Services\Dto\RegionResolution;
use App\Services\LeadDistributor;
use App\Services\LeadRegionResolver;
use App\Services\LeadRouter;
use App\Services\NotificationService;
use App\Services\Pd\PdAuditLogger;
use App\Services\RegionTagResolver;
use App\Services\SupplierProjects\SupplierProjectResolver;
use App\Support\RussianRegions;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable as FoundationQueueable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Carbon;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use RuntimeException;
use Throwable;

/**
 * Routing входящего supplier_lead к eligible Лидерра-проектам (sharing-model).
 *
 * Spec: docs/superpowers/specs/2026-05-10-supplier-integration-design.md §5–§6.
 *       docs/superpowers/specs/2026-05-11-plan4-billing-csv-admin-design.md §3 (Plan 4 Task 4).
 *
 * Алгоритм:
 *   1. Загрузить SupplierLead.
 *   2. Распарсить raw_payload['project'] → (platform, signal_type, identifier).
 *   3. Резолв SupplierProject (resolveOrStub) — persist в supplier_lead.supplier_project_id.
 *   4. LeadRouter::matchEligibleProjects → Collection<Project>.
 *   5. Для каждого Project — DB::transaction с SET LOCAL app.current_tenant_id:
 *      - lockForUpdate Tenant.
 *      - Создать Deal (source_crm_id=vid).
 *      - LedgerService::chargeForDelivery(tenant, deal, lead) — dual-balance
 *        списание (prepaid balance_leads-- ИЛИ rub balance_rub-=tier_price), INSERT
 *        lead_charges + balance_transactions + supplier_lead_costs внутри той же
 *        транзакции. На InsufficientBalanceException — Log::warning + rethrow
 *        (auto-pause flow приходит в Task 6). delivered_today/month++ на проекте,
 *        ActivityLog (EVENT_DEAL_CREATED), NotificationService::notifyNewLead.
 *   6. Обновить SupplierLead.processed_at=now() + deals_created_count.
 */
class RouteSupplierLeadJob implements ShouldQueue
{
    use FoundationQueueable;
    use InteractsWithQueue;
    use Queueable;
    use SerializesModels;

    public int $tries = 3;

    public int $backoff = 60;

    public int $timeout = 60;

    /**
     * Plan 3 Task 3: имя DB-connection (BYPASSRLS-роль crm_supplier_worker), через который
     * supplier-flow обходит RLS для sharing-операций (failed_webhook_jobs с tenant_id=NULL).
     *
     * NB: это НЕ $this->connection из Bus\Queueable — то управляет очередью, не БД.
     * Job's queue connection остаётся default (sync/database), а DB-операции в failed()
     * явно идут через DB::connection(self::DB_CONNECTION). Tenant-scoped транзакции в
     * handle() (createDealCopyForProject) продолжают использовать default `pgsql`
     * с SET LOCAL app.current_tenant_id — там RLS нужна.
     *
     * См. docs/superpowers/specs/2026-05-11-plan3-supplier-sync-design.md §1.
     */
    public const DB_CONNECTION = 'pgsql_supplier';

    public function __construct(public int $supplierLeadId) {}

    public function handle(
        LeadRouter $router,
        SupplierProjectResolver $resolver,
        NotificationService $notifier,
        LedgerService $ledger,
        LeadDistributor $distributor,
        RegionTagResolver $tagResolver,
    ): void {
        $lead = SupplierLead::find($this->supplierLeadId);

        // Терминальный случай: лид удалён/не существует — это НЕ транзиентная ошибка,
        // повтор бессмыслен. НЕ бросаем ModelNotFoundException: иначе queue->failed()
        // пишет строку в failed_webhook_jobs, а RetryFailedSupplierJobsCommand
        // бесконечно перезапускает job (retry-шторм, инцидент 21-22.05.2026 —
        // 25k+ записей по удалённому лиду №1).
        if ($lead === null) {
            Log::warning('supplier_lead.not_found_terminal', [
                'supplier_lead_id' => $this->supplierLeadId,
            ]);

            return;
        }

        // Idempotency guard для retry-сценария ($tries = 3).
        // Если лид уже обработан — выходим, не создаём ghost duplicate'ы deal'ов.
        // CV.11 audit BLOCKER #3 (Plan 2.5 fix).
        if ($lead->processed_at !== null) {
            Log::info('supplier_lead.skipped_already_processed', [
                'supplier_lead_id' => $lead->id,
                'processed_at' => $lead->processed_at->toIso8601String(),
                'deals_created_count' => $lead->deals_created_count,
            ]);

            return;
        }

        // Fast-fail: лид уже был помечен terminal error и не имеет processed_at.
        // Закрывает класс failed_webhook_jobs storm (Finding 2, 2026-05-29).
        // Plan 2026-05-29-supplier-webhook-fast-fail-and-stuck-cleanup.md, Task 2.
        $isTerminalError = $lead->error !== null && (
            str_contains($lead->error, 'does not support')
            || str_contains($lead->error, 'platform mismatch')
            || str_contains($lead->error, 'no matching supplier_project')
        );
        if ($isTerminalError) {
            // Capture original error BEFORE update — $lead->update() mutates
            // the in-memory model, so $lead->error after update() returns the
            // suffixed value, breaking debug logs (review fix).
            $originalError = $lead->error;
            $lead->update([
                'processed_at' => now(),
                'error' => $originalError.' [fast-failed by RouteSupplierLeadJob]',
            ]);
            Log::info('supplier_lead.fast_failed_terminal_error', [
                'supplier_lead_id' => $lead->id,
                'original_error' => $originalError,
            ]);

            return;
        }

        $projectField = (string) ($lead->raw_payload['project'] ?? '');
        [$platform, $signalType, $identifier] = $this->parseProjectField($projectField);

        $supplier = $resolver->resolveOrStub($platform, $signalType, $identifier);
        $lead->update(['supplier_project_id' => $supplier->id]);

        // Lead region resolution (§3.11): резолв региона ДО routing-цикла, чтобы HTTP-вызов
        // DaData (~150мс) не висел внутри tenant-транзакции. Резолвер — из контейнера (не 7-й
        // параметр handle(), чтобы не ломать сигнатуру и существующие вызовы тестов).
        // RegionTagResolver остаётся в DI-цепочке резолвера (fallback-слой).
        $resolution = app(LeadRegionResolver::class)->resolve($lead);
        $lead->update([
            'resolved_subject_code' => $resolution->subjectCode,
            'region_source' => $resolution->source,
            'dadata_qc' => $resolution->qc,
            'phone_operator' => $resolution->phoneOperator,
        ]);

        // Каскад по региону (§3.9): exact → all-RF → fallback. NULL subject_code → шаг 1 пропуск.
        $matched = $router->matchEligibleProjects($supplier, $resolution->subjectCode);
        $selected = $distributor->selectRecipients($matched);

        $createdCount = 0;
        $failures = [];
        foreach ($selected as $project) {
            try {
                if ($this->createDealCopyForProject($lead, $project, $notifier, $ledger, $resolution)) {
                    $createdCount++;
                }
            } catch (Throwable $e) {
                $failures[] = ['project_id' => $project->id, 'tenant_id' => $project->tenant_id, 'error' => $e->getMessage()];
                Log::warning('supplier_lead.per_project_routing_failed', [
                    'supplier_lead_id' => $lead->id,
                    'project_id' => $project->id,
                    'tenant_id' => $project->tenant_id,
                    'exception' => $e->getMessage(),
                ]);
            }
        }

        if ($selected->isNotEmpty() && $createdCount === 0 && count($failures) === $selected->count()) {
            throw new RuntimeException(
                'All eligible projects failed routing for supplier_lead='.$lead->id.
                '; last error: '.($failures[array_key_last($failures)]['error'] ?? 'unknown')
            );
        }

        // Аудит резолва региона — одна строка на лид (§3.10/§7.1). Fail-safe: сбой записи
        // аудит-лога НЕ должен ронять доставку лида (revenue-critical, 30k/сутки).
        $this->logRegionResolution($lead, $resolution, $selected);

        $lead->update([
            'processed_at' => now(),
            'deals_created_count' => $createdCount,
        ]);
    }

    /**
     * Парсит поле raw_payload['project'] (формат `B[123]_<rest>`):
     *   - rest вида `7\d{10}` → call (телефон-номер для звонка-сигнала);
     *   - rest вида `^[a-z0-9-]+(\.[a-z0-9-]+)+$` → site (rest целиком — домен);
     *   - rest со встроенным доменом в свободном тексте → site (identifier =
     *     извлечённый домен; поставщик иногда шлёт имя вида `заявка carmoney.ru/`
     *     или `Платежи cabinet.caranga.ru/login` — регрессия 18.05.2026, 21 лид);
     *   - иначе → sms (короткое имя отправителя SMS-шлюза).
     *
     * @return array{0: string, 1: string, 2: string} [platform, signal_type, identifier]
     */
    private function parseProjectField(string $project): array
    {
        if (preg_match('/^(B[123])_(.+)$/', $project, $m) === 1) {
            $platform = $m[1];
            $rest = $m[2];
        } else {
            // Phase 3: проекты без B-префикса попадают в DIRECT.
            // Весь project считается identifier-частью; signal_type определяется
            // тем же regex'ом, что для $rest у B-префиксных.
            $platform = 'DIRECT';
            $rest = $project;
        }

        // Домен с латинским TLD ≥2 букв (последний сегмент — только буквы), допускается
        // в любой позиции строки. Соответствует чистому rest и встроенному в текст домену.
        $domainRe = '/(?<![a-z0-9.\-])([a-z0-9][a-z0-9\-]*(?:\.[a-z0-9][a-z0-9\-]*)*\.[a-z]{2,})/i';

        if (preg_match('/^7\d{10}$/', $rest) === 1) {
            $signalType = 'call';
            $identifier = $rest;
        } elseif (preg_match('/^[a-z0-9-]+(\.[a-z0-9-]+)+$/i', $rest) === 1) {
            $signalType = 'site';
            $identifier = $rest;
        } elseif (preg_match($domainRe, $rest, $dm) === 1) {
            // Домен извлечён из свободного текста — это сайт-сигнал.
            $signalType = 'site';
            $identifier = mb_strtolower($dm[1]);
        } else {
            $signalType = 'sms';
            $identifier = $rest;
        }

        return [$platform, $signalType, $identifier];
    }

    /**
     * Создаёт deal-копию в одной транзакции для конкретного Project.
     * Возвращает true — если deal создан и баланс списан, счётчики выросли.
     * false — если лимит исчерпан под блокировкой (deal не создаётся).
     */
    private function createDealCopyForProject(
        SupplierLead $lead,
        Project $project,
        NotificationService $notifier,
        LedgerService $ledger,
        RegionResolution $resolution,
    ): bool {
        // routing_step проставлен LeadRouter'ом на matched-проекте; захватываем ДО
        // переназначения $project = $lockedProject (fresh query без этого атрибута).
        $routingStep = (int) ($project->routing_step ?? 1);

        try {
            return DB::transaction(function () use ($lead, $project, $notifier, $ledger, $resolution, $routingStep): bool {
                DB::statement("SET LOCAL app.current_tenant_id = '{$project->tenant_id}'");

                /** @var Tenant $tenant */
                $tenant = Tenant::query()
                    ->whereKey($project->tenant_id)
                    ->lockForUpdate()
                    ->firstOrFail();

                // Concurrency recheck: lockForUpdate(Project) + recheck delivered_today
                // против лимита под блокировкой. Closes CV.11 audit BLOCKER #2 (Plan 2.5).
                // matchEligibleProjects делал SELECT без lock'а — между snapshot'ом и
                // этой транзакцией concurrent webhook мог инкрементить счётчик до limit.
                // Если лимит уже исчерпан — return false (deal не создаём, баланс не списываем).
                /** @var Project $lockedProject */
                $lockedProject = Project::query()
                    ->whereKey($project->id)
                    ->lockForUpdate()
                    ->firstOrFail();

                // R-09 (Task 2.6, spec §4.2.4): recheck is_active под lock'ом.
                // matchEligibleProjects читает snapshot за активную дату (фиксированный
                // на 18:00 МСК); клиент мог нажать «пауза» в окне между matchEligible и
                // этой транзакцией. Snapshot всё ещё говорит "доставлять", но live state
                // — не доставляем (контракт «paused under lock = stop»).
                if (! $lockedProject->is_active) {
                    Log::info('supplier_lead.project_paused_under_lock', [
                        'supplier_lead_id' => $lead->id,
                        'project_id' => $lockedProject->id,
                        'tenant_id' => $tenant->id,
                    ]);

                    return false;
                }

                // R-04 + R-06 (Task 2.6, spec §4.2.4): лимит из snapshot, не live.
                // Slepok-инвариант — лимит зафиксирован на 18:00 МСК; live daily_limit_target
                // (или effective_daily_limit_today) мог быть уменьшен после слепка, но это
                // не должно прерывать поток уже зафиксированного слепка поставщика.
                $msk = Carbon::now('Europe/Moscow');
                $activeDate = $msk->hour >= 21
                    ? $msk->copy()->addDay()->toDateString()
                    : $msk->toDateString();
                $snapshot = DB::connection('pgsql_supplier')
                    ->table('project_routing_snapshots')
                    ->where('snapshot_date', $activeDate)
                    ->where('project_id', $lockedProject->id)
                    ->lockForUpdate()
                    ->first();
                if ($snapshot === null) {
                    Log::info('supplier_lead.no_snapshot_skipped', [
                        'supplier_lead_id' => $lead->id,
                        'project_id' => $lockedProject->id,
                        'tenant_id' => $tenant->id,
                        'active_date' => $activeDate,
                    ]);

                    return false;
                }
                $effectiveLimit = (int) $snapshot->daily_limit;

                if ($lockedProject->delivered_today >= $effectiveLimit) {
                    Log::info('supplier_lead.project_at_limit_skipped', [
                        'supplier_lead_id' => $lead->id,
                        'project_id' => $lockedProject->id,
                        'tenant_id' => $tenant->id,
                        'delivered_today' => $lockedProject->delivered_today,
                        'effective_limit' => $effectiveLimit,
                    ]);

                    return false;
                }
                $project = $lockedProject;

                // Phase 2 fix: merge с CSV-recovered deal если webhook догоняет.
                // Идемпотентность race condition между CsvReconcileJob (vid=NULL, recovered
                // from CSV) и webhook (vid=int, реальный supplier-id). До этой проверки они
                // создавали 2 deal'a (DD снят Spec B Phase 1). Merge выполняется только если:
                //   - webhook ЕСТЬ настоящий vid (lead.vid !== null) — без vid merge'ить нечего;
                //   - csv-recovered deal существует за последние 24h, тот же phone+project+tenant;
                //   - csv-recovered deal БЕЗ source_crm_id (т.е. он именно CSV-recovered, не другой webhook).
                // При merge: UPDATE existing.source_crm_id, INSERT supplier_lead_deliveries,
                // БЕЗ chargeForDelivery (LeadCharge уже есть с момента CSV recovery).
                $existingMergeable = null;
                if ($lead->vid !== null) {
                    $existingMergeable = Deal::query()
                        ->where('tenant_id', $tenant->id)
                        ->where('phone', (string) $lead->phone)
                        ->where('project_id', $project->id)
                        ->whereNull('source_crm_id')
                        ->where('received_at', '>=', now()->subDay())
                        ->lockForUpdate()
                        ->first();
                }
                if ($existingMergeable !== null) {
                    // Заполняем supplier_lead.id у обоих SupplierLead → одному Deal
                    DB::table('supplier_lead_deliveries')->insert([
                        'supplier_lead_id' => $lead->id,
                        'tenant_id' => $tenant->id,
                        'deal_id' => $existingMergeable->id,
                        'created_at' => now(),
                    ]);
                    // Обновляем только source_crm_id + updated_at через DB::table.
                    // NB (регрессия 26.05.2026 04:12-05:03 UTC, 9 failed_jobs):
                    // received_at — partition key, и lead_charges имеет FK
                    // (deal_id, deal_received_at) с ON DELETE CASCADE, но
                    // ON UPDATE NO ACTION (default). Любое изменение received_at
                    // ломает FK даже в той же месячной партиции (даже DEFERRABLE
                    // INITIALLY DEFERRED не помогает — проверка падает на COMMIT).
                    // CSV-recovered received_at сохраняем как есть — отличие на минуты
                    // несущественно, чем риск каскадного DELETE lead_charges.
                    // §3.12: при merge обновляем регион/оператора, если webhook-резолв из
                    // источника выше рангом (dadata/rossvyaz), чем tag CSV-восстановления.
                    // deals не хранит region_source (он на supplier_leads + в журнале), поэтому
                    // ранг определяем по факту источника: dadata/rossvyaz всегда достовернее
                    // tag'а, на котором строилась CSV-recovery (RegionResolution::SOURCE_RANK).
                    $mergeUpdate = ['source_crm_id' => $lead->vid, 'updated_at' => now()];
                    if (in_array($resolution->source, ['dadata', 'rossvyaz'], true) && $resolution->subjectCode !== null) {
                        $mergeUpdate['subject_code'] = $resolution->subjectCode;
                        $mergeUpdate['phone_operator'] = $resolution->phoneOperator;
                        $mergeUpdate['city'] = RussianRegions::CODE_TO_NAME[$resolution->subjectCode] ?? null;
                    }
                    DB::table('deals')
                        ->where('id', $existingMergeable->id)
                        ->where('received_at', $existingMergeable->received_at)
                        ->update($mergeUpdate);

                    Log::info('supplier_lead.merged_into_csv_recovered', [
                        'supplier_lead_id' => $lead->id,
                        'merged_into_deal_id' => $existingMergeable->id,
                        'tenant_id' => $tenant->id,
                    ]);

                    return true; // считаем «доставленным», но без второго списания
                }

                // Spec B: per-(supplier_lead, tenant) lock — одна поставка одному клиенту = один раз.
                // insertOrIgnore вернёт 0, если строка уже существует (повтор/гонка/CSV-recovery).
                $locked = DB::table('supplier_lead_deliveries')->insertOrIgnore([
                    'supplier_lead_id' => $lead->id,
                    'tenant_id' => $tenant->id,
                    'created_at' => now(),
                ]);
                if ($locked === 0) {
                    Log::info('supplier_lead.delivery_already_locked', [
                        'supplier_lead_id' => $lead->id,
                        'tenant_id' => $tenant->id,
                    ]);

                    return false;
                }

                $payload = $lead->raw_payload ?? [];
                $receivedAt = isset($payload['time'])
                    ? Carbon::createFromTimestamp((int) $payload['time'])
                    : ($lead->received_at ?? Carbon::now());

                /** @var array<int, string> $phones */
                $phones = isset($payload['phones']) && is_array($payload['phones'])
                    ? array_values(array_map('strval', $payload['phones']))
                    : [(string) $lead->phone];

                // §3.10: на шаге 3 (запасной канал) регион сделки подменяется на регион
                // клиента (первый подписанный субъект из snapshot); настоящий регион —
                // в lead_region_resolution_log.actual_subject_code. region_substituted флажит подмену.
                $dealSubjectCode = $routingStep < 3
                    ? $resolution->subjectCode
                    : ($this->pickSubstituteRegion((string) ($snapshot->regions ?? '{}')) ?? $resolution->subjectCode);

                $deal = Deal::create([
                    'tenant_id' => $tenant->id,
                    'source_crm_id' => $lead->vid,
                    'project_id' => $project->id,
                    'phone' => (string) $lead->phone,
                    'phones' => $phones,
                    'status' => 'new',
                    'received_at' => $receivedAt,
                    'subject_code' => $dealSubjectCode,
                    // «Город» (UI deals.city) — человекочитаемое имя НАСТОЯЩЕГО региона лида
                    // по резолву (даже если subject_code подменён на шаге 3). NULL → колонка пустая.
                    'city' => $resolution->subjectCode !== null
                        ? (RussianRegions::CODE_TO_NAME[$resolution->subjectCode] ?? null)
                        : null,
                    'phone_operator' => $resolution->phoneOperator,
                    'region_substituted' => $routingStep === 3,
                ]);

                DB::table('supplier_lead_deliveries')
                    ->where('supplier_lead_id', $lead->id)
                    ->where('tenant_id', $tenant->id)
                    ->update(['deal_id' => $deal->id]);

                // Task 6: $ledger->chargeForDelivery бросит InsufficientBalanceException —
                // транзакция откатится, и outer catch ниже отловит для auto-pause flow.
                $ledger->chargeForDelivery($tenant, $deal, $lead);

                $project->increment('delivered_today');
                $project->increment('delivered_in_month');

                // Task 2.6: атомарный инкремент snapshot.delivered_count
                // (для CSV business-drift reconcile — Task 2.5 closure cont'd).
                DB::connection('pgsql_supplier')
                    ->table('project_routing_snapshots')
                    ->where('snapshot_date', $activeDate)
                    ->where('project_id', $project->id)
                    ->increment('delivered_count');

                ActivityLog::create([
                    'tenant_id' => $tenant->id,
                    'user_id' => null,
                    'deal_id' => $deal->id,
                    'event' => ActivityLog::EVENT_DEAL_CREATED,
                    'context' => [
                        'source' => 'supplier_webhook',
                        'supplier_lead_id' => $lead->id,
                    ],
                    'created_at' => now(),
                ]);

                app(PdAuditLogger::class)->record(
                    action: 'created', subjectType: 'lead', subjectId: $deal->id,
                    purpose: 'lead_create_supplier', tenantId: (int) $deal->tenant_id,
                    actorTenantUserId: null, actorAdminUserId: null, ip: null,
                );

                // setRelation чтобы NotificationService мог подтянуть
                // deal->project без N+1 lookup'а под RLS.
                $deal->setRelation('project', $project);
                $notifier->notifyNewLead($tenant, $deal);

                return true;
            });
        } catch (InsufficientBalanceException $e) {
            // Транзакция уже rolled back — Deal не создан, balance не тронут.
            // Запускаем auto-pause flow (Plan 4 Task 6 §4.4) и возвращаем false,
            // чтобы handle()-loop продолжил routing к остальным tenant'ам без rethrow.
            $this->handleInsufficientBalance($lead, $project, $e);

            return false;
        }
    }

    /**
     * Auto-pause flow при недостатке баланса (Plan 4 Task 6 §4.4):
     *
     *   1. UPDATE projects.is_active=false через pgsql_supplier (BYPASSRLS), потому
     *      что текущая транзакция уже rolled back — её SET LOCAL app.current_tenant_id
     *      отвалился и обычный pgsql connection не сможет апдейтить чужой tenant'овый
     *      row под политикой tenant_isolation.
     *   2. Email-алерт ZeroBalancePausedMail с rate-limit 1/час/tenant — через
     *      Redis SETNX (Cache::add вернёт true только при первой попытке за час).
     *   3. Log::warning с диагностикой суммы баланса и цены тарифа.
     */
    private function handleInsufficientBalance(
        SupplierLead $lead,
        Project $project,
        InsufficientBalanceException $e,
    ): void {
        // 1) UPDATE projects.is_active=false через pgsql_supplier (BYPASSRLS).
        DB::connection(self::DB_CONNECTION)
            ->update('UPDATE projects SET is_active = false WHERE id = ?', [$project->id]);

        // 2) Email-алерт с rate-limit 1/час/tenant через Redis SETNX (Cache::add).
        $cacheKey = "billing:zero_balance_alert:{$project->tenant_id}";
        if (Cache::store('redis')->add($cacheKey, true, now()->addHour())) {
            $project->loadMissing('tenant');
            app(NotificationService::class)->notifyZeroBalancePaused(
                $project->tenant,
                $project,
                $e->priceKopecks,
            );
        }

        Log::warning('billing.project_paused_insufficient_balance', [
            'tenant_id' => $project->tenant_id,
            'project_id' => $project->id,
            'supplier_lead_id' => $lead->id,
            'price_kopecks' => $e->priceKopecks,
            'balance_rub' => $e->balanceRub,
        ]);
    }

    /**
     * Аудит резолва региона лида — одна строка на лид в lead_region_resolution_log (§7.1).
     * Fail-safe: сбой записи (например, отсутствие партиции received_at) логируется warning'ом,
     * но НЕ прерывает доставку (revenue-critical). INSERT через pgsql_supplier (GRANT INSERT
     * у crm_supplier_worker). Телефон маскируется до INSERT — сырой номер в лог не пишется.
     *
     * @param  Collection<int, Project>  $selected
     */
    private function logRegionResolution(SupplierLead $lead, RegionResolution $resolution, Collection $selected): void
    {
        try {
            $first = $selected->first();
            $routingStep = $first !== null ? (int) ($first->routing_step ?? 1) : null;
            $substituted = ($routingStep === 3 && $first !== null)
                ? ($this->pickSubstituteRegion((string) ($first->snapshot_regions ?? '{}')) ?? $resolution->subjectCode)
                : null;

            $tagCode = app(RegionTagResolver::class)->resolve((string) ($lead->raw_payload['tag'] ?? ''));

            DB::connection(self::DB_CONNECTION)->table('lead_region_resolution_log')->insert([
                'supplier_lead_id' => $lead->id,
                'received_at' => $lead->received_at ?? now(),
                'phone_masked' => $this->maskPhone((string) $lead->phone),
                'subject_code_resolved' => $resolution->subjectCode,
                'subject_code_from_tag' => $tagCode,
                'region_source' => $resolution->source,
                'dadata_qc' => $resolution->qc,
                'dadata_provider' => $resolution->phoneOperator,
                'dadata_type' => null,
                'dadata_response_masked' => $resolution->dadataResponseMasked !== null
                    ? json_encode($resolution->dadataResponseMasked, JSON_UNESCAPED_UNICODE)
                    : null,
                'rossvyaz_matched' => $resolution->rossvyazMatched,
                'actual_subject_code' => $resolution->actualSubjectCode,
                'substituted_subject_code' => $substituted,
                'routing_step' => $routingStep,
                'phone_operator' => $resolution->phoneOperator,
                'cache_hit' => $resolution->cacheHit,
                'duration_ms' => $resolution->durationMs,
            ]);
        } catch (Throwable $e) {
            Log::warning('lead_region_resolution.log_write_failed', [
                'supplier_lead_id' => $lead->id,
                'exception' => $e->getMessage(),
            ]);
        }
    }

    /**
     * Первый код субъекта из PG INT[]-литерала ('{82,83}' → 82; '{}' → null) — регион клиента
     * для подмены на запасном канале (§3.10).
     */
    private function pickSubstituteRegion(string $regionsLiteral): ?int
    {
        return $this->parseSubjectCodes($regionsLiteral)[0] ?? null;
    }

    /**
     * @return list<int> '{82,83}' → [82,83]; '{}'/'' → []
     */
    private function parseSubjectCodes(string $regionsLiteral): array
    {
        $inner = trim($regionsLiteral, '{}');
        if ($inner === '') {
            return [];
        }

        return array_values(array_map('intval', explode(',', $inner)));
    }

    /**
     * Маскирование телефона для лога (§7.1): первые 4 + последние 4 цифры (7916***4567).
     */
    private function maskPhone(string $phone): string
    {
        $digits = preg_replace('/\D+/', '', $phone) ?? '';
        if (strlen($digits) < 8) {
            return '***';
        }

        return substr($digits, 0, 4).'***'.substr($digits, -4);
    }

    /**
     * Финальный callback после исчерпания всех ретраев ($tries=3).
     *
     * Сохраняет упавший job в `failed_webhook_jobs` (tenant_id=null — sharing-flow,
     * tenant ещё не определён на момент routing'а) для ручного разбора.
     * supplier_lead.error апдейтится текстом исключения.
     *
     * INSERT с tenant_id=NULL проходит благодаря DB_CONNECTION='pgsql_supplier'
     * (BYPASSRLS-роль crm_supplier_worker — обходит политику tenant_isolation,
     * которая под обычной ролью отвергла бы NULL). Закрыто Plan 3 Task 3.
     */
    public function failed(Throwable $e): void
    {
        DB::connection(self::DB_CONNECTION)->table('failed_webhook_jobs')->insert([
            'tenant_id' => null,
            'webhook_log_id' => null,
            'raw_payload' => json_encode([
                'supplier_lead_id' => $this->supplierLeadId,
            ], JSON_UNESCAPED_UNICODE),
            'exception' => $e->getMessage(),
            'retry_count' => $this->tries,
            'failed_at' => now(),
        ]);

        SupplierLead::query()
            ->whereKey($this->supplierLeadId)
            ->update(['error' => $e->getMessage()]);

        Log::error('supplier_lead.routing_failed_permanently', [
            'supplier_lead_id' => $this->supplierLeadId,
            'exception' => $e->getMessage(),
        ]);
    }
}
