portal/app/app/Jobs/Autopodbor/RunAutopodborStudyJob.php

<?php

declare(strict_types=1);

namespace App\Jobs\Autopodbor;

use App\Models\AutopodborCompetitor;
use App\Models\AutopodborRun;
use App\Models\AutopodborSource;
use App\Services\Autopodbor\Agent\CompetitorAgent;
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;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\Middleware\WithoutOverlapping;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\DB;

class RunAutopodborStudyJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;

    public array $backoff = [15, 60, 300];

    public function __construct(public int $runId, public ?int $tenantId = null)
    {
        // Шаг 2 «изучение / дайте источники» — тупой слой (без ИИ). ОТДЕЛЬНАЯ очередь + свой воркер,
        // чтобы изучение шло параллельно поиску, а не ждало его в общей линии.
        $this->onQueue('autopodbor-study');
    }

    /**
     * Защита от нахлёста: один прогон изучения не обрабатывается двумя воркерами разом (ретрай).
     *
     * @return array<int, object>
     */
    public function middleware(): array
    {
        return [new WithoutOverlapping((string) $this->runId)];
    }

    public function handle(
        CompetitorAgent $agent,
        AutopodborDedup $dedup,
        AutopodborChargeService $charge,
        AutopodborNormalizer $norm,
        RunProgressChannel $progress,
        AutopodborStudyScheduler $scheduler,
    ): void {
        // Внешний try/finally: распорядитель должен взять следующего кандидата ПОСЛЕ ЛЮБОГО
        // завершения этого прогона — done/empty/failed, включая ранние return'ы и проброшенное
        // исключение из внутреннего catch. Дорожка шага 2 одна на весь портал (см.
        // AutopodborStudyScheduler) — если не звать tick() здесь, освободившаяся дорожка простаивала
        // бы до следующего крон-тика (autopodbor:study-tick, страховка раз в минуту).
        try {
            // Сначала представляемся базе, ПОТОМ читаем прогон: на боевом роль 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 {
                $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,
                    ],
                    regionCode: (int) $run->region_code,
                ));

                $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;
            }
        } finally {
            $scheduler->tick();
        }
    }
}