diff --git a/app/app/Console/Commands/AutopodborStudyTickCommand.php b/app/app/Console/Commands/AutopodborStudyTickCommand.php new file mode 100644 index 00000000..a53dd816 --- /dev/null +++ b/app/app/Console/Commands/AutopodborStudyTickCommand.php @@ -0,0 +1,22 @@ +tick(); + + return self::SUCCESS; + } +} diff --git a/app/app/Jobs/Autopodbor/RunAutopodborStudyJob.php b/app/app/Jobs/Autopodbor/RunAutopodborStudyJob.php index 768eea04..883a8a9a 100644 --- a/app/app/Jobs/Autopodbor/RunAutopodborStudyJob.php +++ b/app/app/Jobs/Autopodbor/RunAutopodborStudyJob.php @@ -12,6 +12,7 @@ use App\Services\Autopodbor\Agent\Dto\StudyCompetitorRequest; use App\Services\Autopodbor\AutopodborChargeService; use App\Services\Autopodbor\AutopodborDedup; use App\Services\Autopodbor\AutopodborNormalizer; +use App\Services\Autopodbor\AutopodborStudyScheduler; use App\Services\Autopodbor\RunProgressChannel; use App\Support\SystemSettings; use Illuminate\Bus\Queueable; @@ -53,116 +54,126 @@ class RunAutopodborStudyJob implements ShouldQueue AutopodborChargeService $charge, AutopodborNormalizer $norm, RunProgressChannel $progress, + AutopodborStudyScheduler $scheduler, ): void { - // Сначала представляемся базе, ПОТОМ читаем прогон: на боевом роль crm_app_user под FORCE RLS - // не видит строку без tenant-контекста (на dev маскировалось суперюзером postgres → ModelNotFound - // всплыл только на проде 07.07.2026). tenantId приходит из dispatch. Старый путь без tenantId - // (только тесты, где ambient-контекст уже задан) — безопасный откат ниже. Контекст сессионный — - // все дальнейшие запросы (включая вложенные транзакции charge) видят GUC. - if ($this->tenantId !== null) { - DB::statement("SELECT set_config('app.current_tenant_id', ?, false)", [(string) $this->tenantId]); - } - - $run = AutopodborRun::findOrFail($this->runId); - - if ($this->tenantId === null) { - DB::statement("SELECT set_config('app.current_tenant_id', ?, false)", [(string) $run->tenant_id]); - } - - // Идемпотентность: завершённый прогон повторно не обрабатываем (защита от лишнего ретрая/повторного dispatch). - if (in_array($run->status, ['done', 'empty', 'failed'], true)) { - return; - } - - $run->update(['status' => 'running', 'started_at' => now()]); - - // Привязываем канал прогресса к этому прогону — движок отметит этапы изучения именно сюда. - $progress->bind($run->id); - + // Внешний try/finally: распорядитель должен взять следующего кандидата ПОСЛЕ ЛЮБОГО + // завершения этого прогона — done/empty/failed, включая ранние return'ы и проброшенное + // исключение из внутреннего catch. Дорожка шага 2 одна на весь портал (см. + // AutopodborStudyScheduler) — если не звать tick() здесь, освободившаяся дорожка простаивала + // бы до следующего крон-тика (autopodbor:study-tick, страховка раз в минуту). try { - $comp = AutopodborCompetitor::findOrFail($run->competitor_id); + // Сначала представляемся базе, ПОТОМ читаем прогон: на боевом роль crm_app_user под FORCE RLS + // не видит строку без tenant-контекста (на dev маскировалось суперюзером postgres → ModelNotFound + // всплыл только на проде 07.07.2026). tenantId приходит из dispatch. Старый путь без tenantId + // (только тесты, где ambient-контекст уже задан) — безопасный откат ниже. Контекст сессионный — + // все дальнейшие запросы (включая вложенные транзакции charge) видят GUC. + if ($this->tenantId !== null) { + DB::statement("SELECT set_config('app.current_tenant_id', ?, false)", [(string) $this->tenantId]); + } - $res = $agent->studyCompetitor(new StudyCompetitorRequest( - competitor: [ - 'name' => $comp->name, - 'site_url' => $comp->site_url, - 'directory_urls' => $comp->directory_urls ?? [], - 'elements' => $comp->elements ?? [], - 'is_federal' => (bool) $comp->is_federal, - ], - regionCode: (int) $run->region_code, - )); + $run = AutopodborRun::findOrFail($this->runId); - $unique = $dedup->dedupSources($res->sources); - - if ($unique === []) { - $run->update(['status' => 'empty', 'finished_at' => now()]); + if ($this->tenantId === null) { + DB::statement("SELECT set_config('app.current_tenant_id', ?, false)", [(string) $run->tenant_id]); + } + // Идемпотентность: завершённый прогон повторно не обрабатываем (защита от лишнего ретрая/повторного dispatch). + if (in_array($run->status, ['done', 'empty', 'failed'], true)) { return; } - // Классификация для итога (важно при ПОВТОРНОМ сборе): сколько источников совсем новых, - // а какие — ранее удалённые (box=archived), которые движок нашёл снова. Известные (в работе/ - // предложениях) не трогаем — updateOrCreate не меняет им box (его нет в массиве значений). - $newCount = 0; - $refoundDeletedIds = []; - foreach ($unique as $s) { - $identifier = $s['signal_type'] === 'call' - ? $norm->phone($s['identifier']) - : $norm->domainHead($s['identifier']); + $run->update(['status' => 'running', 'started_at' => now()]); - $existing = AutopodborSource::where('competitor_id', $comp->id) - ->where('dedup_key', $s['dedup_key']) - ->first(); + // Привязываем канал прогресса к этому прогону — движок отметит этапы изучения именно сюда. + $progress->bind($run->id); - AutopodborSource::updateOrCreate( - [ - 'competitor_id' => $comp->id, - 'dedup_key' => $s['dedup_key'], + try { + $comp = AutopodborCompetitor::findOrFail($run->competitor_id); + + $res = $agent->studyCompetitor(new StudyCompetitorRequest( + competitor: [ + 'name' => $comp->name, + 'site_url' => $comp->site_url, + 'directory_urls' => $comp->directory_urls ?? [], + 'elements' => $comp->elements ?? [], + 'is_federal' => (bool) $comp->is_federal, ], - [ - 'tenant_id' => $run->tenant_id, - 'study_run_id' => $run->id, - 'signal_type' => $s['signal_type'], - 'identifier' => $identifier, - 'phone_kind' => $s['phone_kind'] ?? null, - 'phone_type' => $s['phone_type'] ?? null, - 'provenance_url' => $s['provenance_url'] ?? null, - 'provenance_label' => $s['provenance_label'] ?? null, - 'where_found' => $s['where_found'] ?? null, - 'office' => $s['office'] ?? null, - 'confirmations' => $s['confirmations'] ?? 1, - ] - ); + regionCode: (int) $run->region_code, + )); - if ($existing === null) { - $newCount++; // совсем новый → в предложения - } elseif ($existing->box === 'archived') { - $refoundDeletedIds[] = $existing->id; // ранее удалённый, найден снова → «вернуть?» + $unique = $dedup->dedupSources($res->sources); + + if ($unique === []) { + $run->update(['status' => 'empty', 'finished_at' => now()]); + + return; } + + // Классификация для итога (важно при ПОВТОРНОМ сборе): сколько источников совсем новых, + // а какие — ранее удалённые (box=archived), которые движок нашёл снова. Известные (в работе/ + // предложениях) не трогаем — updateOrCreate не меняет им box (его нет в массиве значений). + $newCount = 0; + $refoundDeletedIds = []; + foreach ($unique as $s) { + $identifier = $s['signal_type'] === 'call' + ? $norm->phone($s['identifier']) + : $norm->domainHead($s['identifier']); + + $existing = AutopodborSource::where('competitor_id', $comp->id) + ->where('dedup_key', $s['dedup_key']) + ->first(); + + AutopodborSource::updateOrCreate( + [ + 'competitor_id' => $comp->id, + 'dedup_key' => $s['dedup_key'], + ], + [ + 'tenant_id' => $run->tenant_id, + 'study_run_id' => $run->id, + 'signal_type' => $s['signal_type'], + 'identifier' => $identifier, + 'phone_kind' => $s['phone_kind'] ?? null, + 'phone_type' => $s['phone_type'] ?? null, + 'provenance_url' => $s['provenance_url'] ?? null, + 'provenance_label' => $s['provenance_label'] ?? null, + 'where_found' => $s['where_found'] ?? null, + 'office' => $s['office'] ?? null, + 'confirmations' => $s['confirmations'] ?? 1, + ] + ); + + if ($existing === null) { + $newCount++; // совсем новый → в предложения + } elseif ($existing->box === 'archived') { + $refoundDeletedIds[] = $existing->id; // ранее удалённый, найден снова → «вернуть?» + } + } + + $price = (string) (SystemSettings::get('autopodbor_price_study_rub') ?? '0'); + $charge->chargeForRun($run, $price); + + $comp->update(['studied_at' => now(), 'study_run_id' => $run->id]); + + $run->update([ + 'status' => 'done', + 'finished_at' => now(), + 'result' => [ + 'new_sources' => $newCount, + 'refound_deleted' => count($refoundDeletedIds), + 'refound_deleted_ids' => array_values($refoundDeletedIds), + ], + ]); + } catch (\Throwable $e) { + $run->update([ + 'status' => 'failed', + 'error_code' => substr($e->getMessage(), 0, 64), + 'finished_at' => now(), + ]); + throw $e; } - - $price = (string) (SystemSettings::get('autopodbor_price_study_rub') ?? '0'); - $charge->chargeForRun($run, $price); - - $comp->update(['studied_at' => now(), 'study_run_id' => $run->id]); - - $run->update([ - 'status' => 'done', - 'finished_at' => now(), - 'result' => [ - 'new_sources' => $newCount, - 'refound_deleted' => count($refoundDeletedIds), - 'refound_deleted_ids' => array_values($refoundDeletedIds), - ], - ]); - } catch (\Throwable $e) { - $run->update([ - 'status' => 'failed', - 'error_code' => substr($e->getMessage(), 0, 64), - 'finished_at' => now(), - ]); - throw $e; + } finally { + $scheduler->tick(); } } } diff --git a/app/routes/console.php b/app/routes/console.php index b992d7f4..50c45482 100644 --- a/app/routes/console.php +++ b/app/routes/console.php @@ -204,3 +204,20 @@ Schedule::command('scheduler:check-heartbeats') ->timezone('Europe/Moscow') ->onSuccess(fn () => $hb->recordRunResult('scheduler:check-heartbeats', true, null, null)) ->onFailure(fn () => $hb->recordRunResult('scheduler:check-heartbeats', false, 'Command failed', null)); + +// Задача 6 (пакетный сбор источников): страховка распорядителя шага 2. Джоба +// RunAutopodborStudyJob уже зовёт AutopodborStudyScheduler::tick() в finally после +// КАЖДОГО завершения прогона — это основной канал продвижения очереди. Этот крон — +// подстраховка на случай, если дорожка освободилась не через завершение джобы +// (например воркер шага 2 не поднят/упал молча) — раз в минуту проверяет и, +// если дорожка свободна, а очередь не пуста, стартует следующего. +// +// НЕ используем ->withoutOverlapping(): требует таблицу cache_locks, которой у нас +// нет в schema.sql (см. .env.example CACHE_STORE=database + комментарий выше в этом +// файле у Hole #2). Overlap и так не страшен — pickNextRunId()/tick() уже защищены +// собственным Cache::lock('autopodbor:study:tick', 10) внутри AutopodborStudyScheduler, +// а сам pickNextRunId() проверяет status='running' атомарно на момент чтения. +Schedule::command('autopodbor:study-tick') + ->everyMinute() + ->onSuccess(fn () => $hb->recordRunResult('autopodbor:study-tick', true, null, null)) + ->onFailure(fn () => $hb->recordRunResult('autopodbor:study-tick', false, 'Command failed', null)); diff --git a/app/tests/Feature/Autopodbor/StudyJobTriggersSchedulerTest.php b/app/tests/Feature/Autopodbor/StudyJobTriggersSchedulerTest.php new file mode 100644 index 00000000..18e36c21 --- /dev/null +++ b/app/tests/Feature/Autopodbor/StudyJobTriggersSchedulerTest.php @@ -0,0 +1,61 @@ +bind(CompetitorAgent::class, EmptyCompetitorAgent::class); + + $tenant = Tenant::factory()->create(['balance_rub' => '100000.00']); + DB::statement('SET app.current_tenant_id = '.$tenant->id); + SystemSetting::updateOrCreate(['key' => 'autopodbor_price_study_rub'], ['value' => '300', 'type' => 'decimal']); + + $comp1 = AutopodborCompetitor::create([ + 'tenant_id' => $tenant->id, 'name' => 'Окна А', 'origin' => 'manual', + 'dedup_key' => Str::slug('Окна А').'-'.Str::random(6), 'box' => 'field', + ]); + $comp2 = AutopodborCompetitor::create([ + 'tenant_id' => $tenant->id, 'name' => 'Окна Б', 'origin' => 'manual', + 'dedup_key' => Str::slug('Окна Б').'-'.Str::random(6), 'box' => 'field', + ]); + + $run1 = AutopodborRun::create([ + 'tenant_id' => $tenant->id, 'kind' => 'study', 'status' => 'queued', + 'competitor_id' => $comp1->id, 'region_code' => 1, 'params' => [], + ]); + AutopodborRun::whereKey($run1->id)->update(['created_at' => now()->subMinutes(5)]); + + $run2 = AutopodborRun::create([ + 'tenant_id' => $tenant->id, 'kind' => 'study', 'status' => 'queued', + 'competitor_id' => $comp2->id, 'region_code' => 1, 'params' => [], + ]); + AutopodborRun::whereKey($run2->id)->update(['created_at' => now()]); + + // Класс должен инстанцироваться с новой зависимостью в handle() без ошибок конструктора/резолва. + expect(new RunAutopodborStudyJob($run1->id, $tenant->id))->toBeInstanceOf(RunAutopodborStudyJob::class); + + // Прогоняем ПЕРВЫЙ прогон РЕАЛЬНО (через handle(), не через Queue) — Queue::fake() гарантирует, + // что единственный dispatch RunAutopodborStudyJob, который мы увидим, придёт ИЗНУТРИ handle() + // (от распорядителя), а не от нашего собственного вызова. + app()->call([new RunAutopodborStudyJob($run1->id, $tenant->id), 'handle']); + + expect($run1->fresh()->status)->toBe('empty'); + + Queue::assertPushed(RunAutopodborStudyJob::class, fn ($job) => $job->runId === $run2->id); +});