feat(autopodbor): джоба зовёт распорядителя после завершения + крон autopodbor:study-tick

Задача 5: RunAutopodborStudyJob::handle() оборачивает существующее тело в try/finally — после ЛЮБОГО завершения прогона (done/empty/failed, включая проброшенное исключение) распорядитель AutopodborStudyScheduler::tick() берёт следующего кандидата с освободившейся дорожки шага 2. Задача 6: artisan autopodbor:study-tick — крон-страховка раз в минуту на случай, если дорожка освободилась не через завершение джобы. Без withoutOverlapping() — требует таблицу cache_locks, которой нет в схеме (см. комментарий в routes/console.php); overlap уже защищён внутренним Cache::lock в AutopodborStudyScheduler::tick().
2026-07-08 18:15:23 +03:00
parent 097f7673ee
commit d228a501dc
4 changed files with 207 additions and 96 deletions
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Console\Commands;
+
+use App\Services\Autopodbor\AutopodborStudyScheduler;
+use Illuminate\Console\Command;
+
+class AutopodborStudyTickCommand extends Command
+{
+    protected $signature = 'autopodbor:study-tick';
+
+    protected $description = 'Страховка: если дорожка шага 2 свободна, а очередь не пуста — запустить следующего.';
+
+    public function handle(AutopodborStudyScheduler $scheduler): int
+    {
+        $scheduler->tick();
+
+        return self::SUCCESS;
+    }
+}
@@ -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();
        }
    }
 }
@@ -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));
@@ -0,0 +1,61 @@
+<?php
+
+declare(strict_types=1);
+
+use App\Jobs\Autopodbor\RunAutopodborStudyJob;
+use App\Models\AutopodborCompetitor;
+use App\Models\AutopodborRun;
+use App\Models\SystemSetting;
+use App\Models\Tenant;
+use App\Services\Autopodbor\Agent\CompetitorAgent;
+use Illuminate\Foundation\Testing\DatabaseTransactions;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Queue;
+use Illuminate\Support\Str;
+use Tests\Concerns\SharesSupplierPdo;
+use Tests\Doubles\EmptyCompetitorAgent;
+
+uses(DatabaseTransactions::class, SharesSupplierPdo::class);
+
+it('после завершения прогона джоба сама зовёт распорядителя, и тот диспатчит следующий queued', function () {
+    Queue::fake();
+    // Детерминированная заглушка агента (без реальной сети) — прогон уйдёт в `empty` быстро.
+    app()->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);
+});