portal/app/app/Jobs/ProcessWebhookJob.php

<?php

declare(strict_types=1);

namespace App\Jobs;

use App\Models\ActivityLog;
use App\Models\BalanceTransaction;
use App\Models\Deal;
use App\Models\FailedWebhookJob;
use App\Models\Project;
use App\Models\RejectedDealsLog;
use App\Models\SupplierLeadCost;
use App\Models\SystemSetting;
use App\Models\Tenant;
use App\Services\DuplicateDetector;
use App\Services\NotificationService;
use App\Services\SupplierResolver;
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\DB;
use Illuminate\Support\Facades\Log;
use RuntimeException;
use Throwable;

/**
 * Асинхронная обработка webhook'а от crm.bp-gr.ru (narrative §5.5 v8.7).
 *
 * Архитектура:
 *   1. RLS: SET LOCAL app.current_tenant_id внутри транзакции (PgBouncer-safe).
 *   2. Lock на tenant + балансовая проверка → RejectedDealsLog при balance=0.
 *   3. findOrCreate проекта (префикс B[123]_ обрезан).
 *   4. Идемпотентный upsert через pg_advisory_xact_lock (см. upsertDeal()).
 *   5. Для НОВОЙ сделки: списание баланса + BalanceTransaction +
 *      SupplierLeadCost (Ю-2) + ActivityLog(deal.created).
 *
 * Антифрод-дедуп Биз-19 (§10.8.1): при создании НОВОЙ сделки `DuplicateDetector`
 * ищет master по `(tenant_id, phone)` в окне 24 ч. Если master найден — новой
 * сделке проставляется `duplicate_of_id`, баланс НЕ списывается, SupplierLeadCost
 * НЕ создаётся. ActivityLog пишется с context.duplicate_of=master.id.
 *
 * Уведомления (ТЗ §18.5, событие new_lead): после успешного chargeNewLead
 * вызывается NotificationService::notifyNewLead, который рассылает email
 * всем активным user'ам тенанта с включённым каналом email для new_lead.
 *
 * Не входит в текущий PoC (отдельные ветви фазы 1):
 *   - Sentry::captureException в failed() (нет Sentry-DSN на dev-стеке)
 *   - SystemSetting fallback для supplier_id (сейчас лукап через project_suppliers)
 */
class ProcessWebhookJob implements ShouldQueue
{
    use FoundationQueueable;
    use InteractsWithQueue;
    use Queueable;
    use SerializesModels;

    public int $tries = 3;

    public int $backoff = 60;

    public int $timeout = 30;

    /**
     * @param  array<string, mixed>  $data  Webhook payload: vid, project, tag, phone, phones, time
     */
    public function __construct(
        public int $tenantId,
        public array $data,
        public ?int $webhookLogId = null,
    ) {}

    public function handle(): void
    {
        $duplicateDetector = app(DuplicateDetector::class);

        DB::transaction(function () use ($duplicateDetector): void {
            DB::statement('SET LOCAL app.current_tenant_id = '.$this->tenantId);

            $tenant = Tenant::query()
                ->whereKey($this->tenantId)
                ->lockForUpdate()
                ->first();

            if ($tenant === null) {
                throw new RuntimeException("Tenant {$this->tenantId} not found");
            }

            if ((int) $tenant->balance_leads <= 0) {
                $this->logRejection($tenant, RejectedDealsLog::REASON_ZERO_BALANCE);

                return;
            }

            $cleanProjectName = preg_replace('/^B[123]_/', '', (string) $this->data['project']);
            $project = Project::firstOrCreate(
                ['tenant_id' => $tenant->id, 'name' => $cleanProjectName],
                ['type' => 'webhook'],
            );

            $receivedAt = Carbon::createFromTimestamp((int) $this->data['time']);
            $sourceCrmId = (int) $this->data['vid'];

            $deal = $this->upsertDeal(
                tenant: $tenant,
                project: $project,
                sourceCrmId: $sourceCrmId,
                receivedAt: $receivedAt,
            );

            if (! $deal->wasRecentlyCreated) {
                return;
            }

            // Биз-19: master-сделка по phone в окне 24 ч → дубль, без charge.
            $master = $duplicateDetector->findMaster(
                tenantId: $tenant->id,
                phone: (string) $this->data['phone'],
                now: $receivedAt,
            );

            // Сам только что созданный $deal попадает в выборку DuplicateDetector
            // (он уже в БД к моменту lookup'а), поэтому master может ===$deal.
            // Дубль — только если master найден И это НЕ сам deal.
            if ($master !== null && $master->id !== $deal->id) {
                $this->markAsDuplicate($tenant, $deal, $master);

                return;
            }

            $this->chargeNewLead($tenant, $project, $deal);
        });
    }

    /**
     * Биз-19: помечаем сделку как дубль master'а. БЕЗ списания баланса
     * и БЕЗ SupplierLeadCost (не наша закупка). ActivityLog пишется с
     * `context.duplicate_of=master.id` для аудита.
     */
    private function markAsDuplicate(Tenant $tenant, Deal $deal, Deal $master): void
    {
        $deal->update(['duplicate_of_id' => $master->id]);

        ActivityLog::create([
            'tenant_id' => $tenant->id,
            'user_id' => null,
            'deal_id' => $deal->id,
            'event' => ActivityLog::EVENT_DEAL_CREATED,
            'context' => [
                'source' => 'webhook',
                'duplicate_of' => $master->id,
            ],
            'created_at' => now(),
        ]);
    }

    private function logRejection(Tenant $tenant, string $reason): void
    {
        $rejected = RejectedDealsLog::create([
            'tenant_id' => $tenant->id,
            'webhook_log_id' => $this->webhookLogId,
            'reason' => $reason,
            'payload' => $this->data,
            'created_at' => now(),
        ]);

        Log::info("webhook.rejected.{$reason}", [
            'tenant_id' => $tenant->id,
            'vid' => $this->data['vid'] ?? null,
        ]);

        // ТЗ §18.5: zero_balance — уведомить тенант. Anti-spam: не более
        // 1 email/час на тенант. Исключаем только что вставленную запись
        // через id (timestamp-сравнение ненадёжно из-за microsecond precision).
        if ($reason === RejectedDealsLog::REASON_ZERO_BALANCE) {
            $previousCount = RejectedDealsLog::query()
                ->where('tenant_id', $tenant->id)
                ->where('reason', $reason)
                ->where('created_at', '>=', now()->subHour())
                ->where('id', '!=', $rejected->id)
                ->count();

            if ($previousCount === 0) {
                app(NotificationService::class)->notifyZeroBalance($tenant);
            }
        }
    }

    /**
     * Списание баланса при создании НОВОЙ сделки + аудит-записи.
     *
     * Все INSERT'ы в одной транзакции — целостность гарантирована (Ю-2):
     * deal + supplier_lead_cost + balance_transaction появляются атомарно.
     */
    private function chargeNewLead(Tenant $tenant, Project $project, Deal $deal): void
    {
        $tenant->decrement('balance_leads');
        $tenant->refresh();

        BalanceTransaction::create([
            'tenant_id' => $tenant->id,
            'type' => BalanceTransaction::TYPE_LEAD_CHARGE,
            'amount_leads' => -1,
            'balance_leads_after' => (int) $tenant->balance_leads,
            'related_type' => Deal::class,
            'related_id' => $deal->id,
            'created_at' => now(),
        ]);

        $resolver = app(SupplierResolver::class);
        $supplierId = $resolver->resolveForProject($project);
        if ($supplierId !== null) {
            SupplierLeadCost::create([
                'deal_id' => $deal->id,
                'received_at' => $deal->received_at,
                'supplier_id' => $supplierId,
                'cost_rub' => $resolver->costRubSnapshot($supplierId),
                'supplier_lead_id' => (int) $this->data['vid'],
                'created_at' => now(),
            ]);
        } else {
            Log::warning('webhook.no_active_supplier', [
                'tenant_id' => $tenant->id,
                'project_id' => $project->id,
                'deal_id' => $deal->id,
            ]);
        }

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

        // Уведомление о новом лиде (ТЗ §18.5). Отправляется ПОСЛЕ всех записей
        // в БД, чтобы при ошибке отправки транзакция уже была зафиксирована.
        // NotificationService сам ловит Throwable от Mail::send и логирует —
        // отказ канала не должен валить webhook.
        $deal->setRelation('project', $project);
        $service = app(NotificationService::class);
        $service->notifyNewLead($tenant, $deal);

        // ТЗ §18.5: low_balance — после lead_charge проверяем порог. Триггерим
        // ТОЛЬКО когда баланс пересекает порог сверху-вниз: balance_after <=
        // threshold AND (balance_after + 1) > threshold. Иначе шлёт спам после
        // каждого lead_charge при balance < threshold.
        $threshold = $this->lowBalanceThreshold();
        $balanceAfter = (int) $tenant->balance_leads;
        if ($balanceAfter <= $threshold && ($balanceAfter + 1) > $threshold) {
            $service->notifyLowBalance($tenant, $threshold);
        }
    }

    /**
     * Читает порог из system_settings.low_balance_threshold_leads.
     * Default 10 (см. schema.sql:2239 seed).
     */
    private function lowBalanceThreshold(): int
    {
        $setting = SystemSetting::query()->where('key', 'low_balance_threshold_leads')->first();
        if ($setting === null) {
            return 10;
        }

        return (int) $setting->value;
    }

    /**
     * Идемпотентная upsert-логика через advisory lock (§5.5 v8.7).
     *
     * Стратегия:
     *   1. pg_advisory_xact_lock(tenant_id, vid) — сериализует все операции
     *      с (tenant_id, source_crm_id) на время транзакции.
     *   2. SELECT в webhook_dedup_keys — атомарно из-за lock.
     *   3a. Если найдено — UPDATE deal по composite-ключу (id, received_at).
     *   3b. Иначе — INSERT deal первым (FK immediate OK), затем INSERT dedup_key.
     *
     * См. db/CHANGELOG_schema.md §W для архитектурного обоснования
     * (PG savepoint+DEFERRED quirk, отказ от двустадийного INSERT-в-dedup-keys-первым).
     */
    private function upsertDeal(
        Tenant $tenant,
        Project $project,
        int $sourceCrmId,
        Carbon $receivedAt,
    ): Deal {
        // pg_advisory_xact_lock(bigint): комбинируем (tenant_id, source_crm_id)
        // в один bigint — верхние 32 бита tenant_id, нижние 32 — source_crm_id.
        $lockKey = (($tenant->id & 0xFFFFFFFF) << 32) | ($sourceCrmId & 0xFFFFFFFF);
        DB::statement('SELECT pg_advisory_xact_lock(?)', [$lockKey]);

        $existing = DB::selectOne(
            'SELECT deal_id, deal_received_at FROM webhook_dedup_keys WHERE tenant_id = ? AND source_crm_id = ?',
            [$tenant->id, $sourceCrmId],
        );

        if ($existing !== null) {
            $deal = Deal::query()
                ->where('id', $existing->deal_id)
                ->where('received_at', $existing->deal_received_at)
                ->firstOrFail();

            $deal->update([
                'phone' => (string) $this->data['phone'],
                'phones' => $this->data['phones'] ?? [(string) $this->data['phone']],
                // status НЕ перезаписываем — менеджер мог изменить.
            ]);

            DB::table('webhook_dedup_keys')
                ->where('tenant_id', $tenant->id)
                ->where('source_crm_id', $sourceCrmId)
                ->update(['updated_at' => now()]);

            $deal->wasRecentlyCreated = false;

            return $deal;
        }

        $deal = Deal::create([
            'tenant_id' => $tenant->id,
            'source_crm_id' => $sourceCrmId,
            'project_id' => $project->id,
            'phone' => (string) $this->data['phone'],
            'phones' => $this->data['phones'] ?? [(string) $this->data['phone']],
            'status' => 'new',
            'received_at' => $receivedAt,
        ]);

        DB::table('webhook_dedup_keys')->insert([
            'tenant_id' => $tenant->id,
            'source_crm_id' => $sourceCrmId,
            'deal_id' => $deal->id,
            'deal_received_at' => $deal->received_at,
            'created_at' => now(),
        ]);

        return $deal;
    }

    /**
     * Финальный callback после исчерпания всех ретраев ($tries=3).
     *
     * Сохраняет упавший job в `failed_webhook_jobs` для ручного разбора и
     * возможного повторного запуска через админку SaaS. RLS не задаём —
     * tenant_id из job-state передаётся как есть (failed-callback запускается
     * вне транзакции воркера). На production добавляется Sentry::captureException.
     *
     * NB: записывается через DB::table (не через FailedWebhookJob::create),
     * чтобы избежать RLS-фильтрации при отсутствии app.current_tenant_id —
     * запись должна попасть в БД даже в катастрофическом сценарии.
     */
    public function failed(Throwable $e): void
    {
        DB::table('failed_webhook_jobs')->insert([
            'tenant_id' => $this->tenantId,
            'webhook_log_id' => $this->webhookLogId,
            'raw_payload' => json_encode($this->data, JSON_UNESCAPED_UNICODE),
            'exception' => $e->getMessage(),
            'retry_count' => $this->tries,
            'failed_at' => now(),
        ]);

        Log::error('webhook.job_failed_permanently', [
            'tenant_id' => $this->tenantId,
            'vid' => $this->data['vid'] ?? null,
            'exception' => $e->getMessage(),
        ]);

        // TODO(production): Sentry::captureException($e);
    }
}