portal/app/app/Jobs/RouteSupplierLeadJob.php

<?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\LeadDistributor;
use App\Services\LeadRouter;
use App\Services\NotificationService;
use App\Services\Pd\PdAuditLogger;
use App\Services\RegionTagResolver;
use App\Services\SupplierProjects\SupplierProjectResolver;
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\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;
        }

        $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]);

        $matched = $router->matchEligibleProjects($supplier);
        $selected = $distributor->selectRecipients($matched); // cap=3 случайных

        $subjectCode = $tagResolver->resolve((string) ($lead->raw_payload['tag'] ?? ''));

        $createdCount = 0;
        $failures = [];
        foreach ($selected as $project) {
            try {
                if ($this->createDealCopyForProject($lead, $project, $notifier, $ledger, $subjectCode)) {
                    $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')
            );
        }

        $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,
        ?int $subjectCode,
    ): bool {
        try {
            return DB::transaction(function () use ($lead, $project, $notifier, $ledger, $subjectCode): 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();
                $effectiveLimit = $lockedProject->effective_daily_limit_today ?? $lockedProject->daily_limit_target;
                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 и опционально received_at через
                    // DB::table (надёжнее Eloquent save() на партиционированной таблице).
                    $newReceivedAt = ($lead->received_at !== null && $lead->received_at->gt($existingMergeable->received_at))
                        ? $lead->received_at
                        : null;
                    $updateData = ['source_crm_id' => $lead->vid, 'updated_at' => now()];
                    if ($newReceivedAt !== null) {
                        $updateData['received_at'] = $newReceivedAt;
                    }
                    DB::table('deals')
                        ->where('id', $existingMergeable->id)
                        ->where('received_at', $existingMergeable->received_at)
                        ->update($updateData);

                    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];

                $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' => $subjectCode,
                ]);

                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');

                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,
        ]);
    }

    /**
     * Финальный 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(),
        ]);
    }
}