Дмитрий
ae286b9a0d
92 файла одной пачкой. Исключены чужие зоны: CLAUDE.md, .claude/settings.json, docs/observer/.pii-counters.json. gitleaks staged: no leaks found. Не верифицировано тестами - сохранение труда в историю. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
667 lines
34 KiB
PHP
667 lines
34 KiB
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\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(),
|
||
]);
|
||
}
|
||
}
|