Дмитрий
e7a1b10441
Закрывает замечания заказчика (22.05.2026) по проектам/поставщику. Все 4 куска имеют общий корень: online-синхронизация одного проекта работала с данными ЭТОГО проекта, а не пересчитывала всю «группу» (проекты разных tenant'ов с одним identifier) — отсюда переплата ×3 при изменении лимита, затирание регионов/дней группы, неотправленная пауза, и осиротевшие проекты при смене источника. 1. Групповой пересчёт в SyncSupplierProjectJob::handleOnline (#1 при изменении, #2 дни, #3 регионы, C2/C3): union regions, computeOrder eligible, distributeForPlatform — те же расчёты, что в ночном syncGroup. Online и ночной теперь дают идентичный supplier-state, расхождение устранено. 2. Пауза #10: - ProjectController::toggleActive — диспатчит SyncSupplierProjectJob; - ProjectService::bulkPauseResume — диспатчит sync per project; - DTO status вычисляется из groupActive (paused когда группа без активных); - sp.inactive_since пишется при пересинке (для UI/DTO консистентности). 3. Смена источника #8/#9 в ProjectService::update: - до update снимается старый buildUniqueKeyAgnostic; - если изменился — отвязываем старые supplier_projects от этого project (pivot + legacy FK), DeleteSupplierProjectJob удаляет их у поставщика при отсутствии других потребителей, либо пересинкает агрегат. 4. Перенос auto-link корня из feat/root-domain-auto-link: новый App\Support\SupplierIdentifier::extractRootDomain + блоки auto-link в обоих джобах (online + nightly). Тесты: TDD на каждый кусок. SyncSupplierProjectJobTest +2 (group recompute, pause). ProjectUpdateDedupTest +1 (source detach + cleanup dispatch). ProjectsActionsTest +2 (toggle + bulk pause dispatches). Регрессия: 186/186 passed (Project/Plan5/Projects + Supplier), 502 assertions. Деплой: дельтой на боевой (база = root-domain ветка; на боевом джобы СТАРЕЕ main, deliver через копию изменённых файлов + config:cache + restart queue). План: docs/superpowers/plans/2026-05-22-замечания-проекты-чеклист.md Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
488 lines
22 KiB
PHP
488 lines
22 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace App\Jobs;
|
||
|
||
use App\Models\Project;
|
||
use App\Models\SupplierProject;
|
||
use App\Services\Supplier\Channel\Exceptions\TierEscalatedException;
|
||
use App\Services\Supplier\Channel\Exceptions\WindowDeferredException;
|
||
use App\Services\Supplier\Channel\FailoverProjectChannel;
|
||
use App\Services\Supplier\Channel\SupplierProjectChannel;
|
||
use App\Services\Supplier\Dto\SupplierProjectDto;
|
||
use App\Services\Supplier\SupplierExportMode;
|
||
use App\Services\Supplier\SupplierPortalClient;
|
||
use App\Services\Supplier\SupplierProjectGrouping;
|
||
use App\Services\Supplier\SupplierQuotaAllocator;
|
||
use App\Support\RussianRegions;
|
||
use App\Support\SupplierIdentifier;
|
||
use Carbon\Carbon;
|
||
use Illuminate\Bus\Queueable;
|
||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||
use Illuminate\Foundation\Bus\Dispatchable;
|
||
use Illuminate\Queue\InteractsWithQueue;
|
||
use Illuminate\Queue\SerializesModels;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Log;
|
||
|
||
/**
|
||
* Синхронизирует Лидерра-проект с supplier_projects на B1/B2/B3
|
||
* в зависимости от signal_type и текущего SupplierExportMode.
|
||
*
|
||
* Режимы:
|
||
* online → для каждой (subject × platform-set) группы проекта:
|
||
* saveProjectMultiFlag с полными параметрами (limit, regions, tag)
|
||
* → upsert supplier_projects + pivot project_supplier_links.
|
||
* batch → «каркас»: создаёт supplier_projects с limit=0, без регионов
|
||
* (старый путь); ночной SyncSupplierProjectsJob дольёт полные параметры.
|
||
*
|
||
* Канал миграции:
|
||
* batch mode — SupplierProjectChannel (FailoverProjectChannel: ярус 1 AJAX
|
||
* → ярус 2 browser-form → ярус 3 manual queue) для createProject.
|
||
* online mode — multi-flag save идёт напрямую через SupplierPortalClient
|
||
* (tier-1 AJAX only — multi-flag нет в tier-2 form по архитектуре
|
||
* портала). При любом transient/auth fail → log warning + skip
|
||
* subject; Laravel retry (tries=3 backoff [15s,60s,300s]) → ночной
|
||
* SyncSupplierProjectsJob подберёт с полным failover каналом.
|
||
* updateProject в online остаётся через $channel (полная схема failover).
|
||
* При эскалации на ярус 3 / переносе по окну портала — platform/subject пропускается
|
||
* (FK/pivot остаётся пустым; ночной SyncSupplierProjectsJob восстанавливает).
|
||
*
|
||
* Retry: 3 попытки с backoff [15s, 60s, 300s].
|
||
*
|
||
* Spec: docs/superpowers/specs/2026-05-19-supplier-project-channel-failover-design.md §5
|
||
* Plan: docs/superpowers/plans/2026-05-20-project-migration-redesign-plan-3-export.md Task 6
|
||
*/
|
||
class SyncSupplierProjectJob implements ShouldQueue
|
||
{
|
||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||
|
||
public int $tries = 3;
|
||
|
||
/** @var array<int, int> */
|
||
public array $backoff = [15, 60, 300];
|
||
|
||
/**
|
||
* BYPASSRLS-роль crm_supplier_worker для всех DB-операций (как у всех supplier-flow
|
||
* джобов: SyncSupplierProjectsJob/DeleteSupplierProjectJob/CsvReconcileJob/…).
|
||
*
|
||
* Джоб запускается из очереди, где SetTenantContext-прослойка не отрабатывает и
|
||
* app.current_tenant_id GUC не установлен. Под обычной ролью crm_app_user первый же
|
||
* SELECT по projects падает 42704 (unrecognized configuration parameter
|
||
* "app.current_tenant_id"). На dev не всплывало — там DB_USERNAME=postgres (superuser,
|
||
* RLS обходится). Plan 3 Task 3 learning.
|
||
*/
|
||
public const DB_CONNECTION = 'pgsql_supplier';
|
||
|
||
public function __construct(public int $projectId) {}
|
||
|
||
public function handle(SupplierProjectChannel $channel): void
|
||
{
|
||
$project = Project::on(self::DB_CONNECTION)->find($this->projectId);
|
||
|
||
if ($project === null) {
|
||
Log::warning("SyncSupplierProjectJob: project {$this->projectId} not found — skipping");
|
||
|
||
return;
|
||
}
|
||
|
||
if (SupplierExportMode::isOnline()) {
|
||
$this->handleOnline($project, $channel);
|
||
} else {
|
||
$this->handleBatch($project, $channel);
|
||
}
|
||
}
|
||
|
||
// -------------------------------------------------------------------------
|
||
// Online mode: per-subject full-param sync
|
||
// -------------------------------------------------------------------------
|
||
|
||
private function handleOnline(Project $project, SupplierProjectChannel $channel): void
|
||
{
|
||
$client = app(SupplierPortalClient::class);
|
||
|
||
$platforms = SupplierProjectGrouping::resolvePlatforms($project);
|
||
if ($platforms === []) {
|
||
return;
|
||
}
|
||
|
||
$identifier = SupplierProjectGrouping::buildUniqueKey($project, $platforms[0]);
|
||
|
||
// GROUP recompute (multi-client): an online edit of ONE project must recompute the
|
||
// WHOLE group sharing this identifier — otherwise it overwrites siblings' regions/
|
||
// limit/days until the nightly batch. Mirrors SyncSupplierProjectsJob::syncGroup so
|
||
// online and nightly produce identical supplier state.
|
||
$agnostic = SupplierProjectGrouping::buildUniqueKeyAgnostic($project);
|
||
$groupProjects = Project::on(self::DB_CONNECTION)
|
||
->where('is_active', true)
|
||
->where('signal_type', (string) $project->signal_type)
|
||
->get()
|
||
->filter(fn (Project $gp) => SupplierProjectGrouping::buildUniqueKeyAgnostic($gp) === $agnostic)
|
||
->values();
|
||
|
||
// status: paused when the whole group has no active project (the pause was the last one).
|
||
$groupActive = $groupProjects->isNotEmpty();
|
||
$status = $groupActive ? 'active' : 'paused';
|
||
|
||
// eligible tomorrow → order/workdays (mirror nightly's eligibility window).
|
||
$targetWeekday = Carbon::tomorrow('Europe/Moscow')->isoWeekday();
|
||
$eligible = $groupProjects->filter(
|
||
fn (Project $gp) => ((int) $gp->delivery_days_mask & (1 << ($targetWeekday - 1))) !== 0
|
||
)->values();
|
||
|
||
$order = SupplierQuotaAllocator::computeOrder(
|
||
$eligible->map(fn (Project $gp) => (int) $gp->daily_limit_target)->all()
|
||
);
|
||
|
||
// union regions across the group (any project "all-RF" → whole group all-RF).
|
||
$hasAllRussia = false;
|
||
$merged = [];
|
||
foreach ($groupProjects as $gp) {
|
||
$r = array_map('intval', (array) ($gp->regions ?? []));
|
||
if ($r === []) {
|
||
$hasAllRussia = true;
|
||
$merged = [];
|
||
break;
|
||
}
|
||
$merged = array_values(array_unique(array_merge($merged, $r)));
|
||
}
|
||
$allRegions = $hasAllRussia ? [] : $merged;
|
||
sort($allRegions);
|
||
|
||
// union workdays of eligible projects (fallback to this project's mask if group empty).
|
||
$wd = [];
|
||
foreach ($eligible as $gp) {
|
||
foreach ($this->workdaysFromMask((int) $gp->delivery_days_mask) as $d) {
|
||
$wd[$d] = $d;
|
||
}
|
||
}
|
||
sort($wd);
|
||
$workdays = $wd !== [] ? $wd : $this->workdaysFromMask((int) $project->delivery_days_mask);
|
||
|
||
// count=0 → all-Russia; count=1 → named region; count>1 → merged → 'РФ'
|
||
$tag = count($allRegions) === 1
|
||
? (RussianRegions::CODE_TO_NAME[$allRegions[0]] ?? (string) $allRegions[0])
|
||
: 'РФ';
|
||
|
||
// Split the GROUP order across platforms so Σ per-platform == order (no ×N overspend).
|
||
$shares = SupplierQuotaAllocator::distributeForPlatform($order, $platforms);
|
||
|
||
// Idempotency: find existing by identifier regardless of subject_code (any previous run).
|
||
$existingSps = SupplierProject::on(self::DB_CONNECTION)
|
||
->where('unique_key', $identifier)
|
||
->where('signal_type', (string) $project->signal_type)
|
||
->whereIn('platform', $platforms)
|
||
->get();
|
||
|
||
// Fully-paused group with nothing yet at supplier — nothing to create.
|
||
if ($existingSps->isEmpty() && ! $groupActive) {
|
||
return;
|
||
}
|
||
|
||
if ($existingSps->isEmpty()) {
|
||
// Create path: one save PER platform with that platform's divided share
|
||
// (single-flag save → exactly one rt-project, reliable id via listProjects match).
|
||
$idMap = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $platforms);
|
||
|
||
foreach ($platforms as $platform) {
|
||
$externalId = $idMap[$platform] ?? null;
|
||
if ($externalId === null) {
|
||
continue;
|
||
}
|
||
|
||
$sp = SupplierProject::on(self::DB_CONNECTION)->create([
|
||
'platform' => $platform,
|
||
'signal_type' => (string) $project->signal_type,
|
||
'unique_key' => $identifier,
|
||
'subject_code' => null,
|
||
'supplier_external_id' => (string) $externalId,
|
||
'current_limit' => $shares[$platform] ?? 0,
|
||
'current_workdays' => $workdays,
|
||
'current_regions' => $allRegions,
|
||
'sync_status' => 'ok',
|
||
'last_synced_at' => now(),
|
||
]);
|
||
|
||
$existingSps->push($sp);
|
||
}
|
||
} else {
|
||
// External-deletion recovery: донор мог быть удалён на портале (вручную или
|
||
// прошлым hard-delete). Тогда external_id в нашей БД мёртв, а updateProject
|
||
// такого id портал молча принимает (no-op) — донор не пересоздаётся. Поэтому
|
||
// сверяемся со списком живых проектов портала и пересоздаём недостающих
|
||
// in-place (НЕ удаляя записи — на supplier_project могут висеть лиды/списания).
|
||
$livePortalIds = collect($client->listProjects())
|
||
->map(fn ($p) => (string) ($p['id'] ?? ''))
|
||
->filter()
|
||
->all();
|
||
|
||
$deadSps = $existingSps->filter(
|
||
fn (SupplierProject $sp) => $sp->supplier_external_id !== null
|
||
&& ! in_array((string) $sp->supplier_external_id, $livePortalIds, true)
|
||
);
|
||
|
||
if ($deadSps->isNotEmpty()) {
|
||
$deadPlatforms = array_values($deadSps->pluck('platform')->all());
|
||
$recreatedIdMap = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $deadPlatforms);
|
||
|
||
foreach ($deadSps as $sp) {
|
||
$newId = $recreatedIdMap[$sp->platform] ?? null;
|
||
if ($newId !== null) {
|
||
$sp->forceFill(['supplier_external_id' => (string) $newId])->save();
|
||
}
|
||
}
|
||
}
|
||
|
||
// Partial-set recovery: если предыдущий run создал не все platforms.
|
||
$existingPlatforms = $existingSps->pluck('platform')->all();
|
||
$missingPlatforms = array_values(array_diff($platforms, $existingPlatforms));
|
||
|
||
if ($missingPlatforms !== []) {
|
||
$missingIdMap = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $missingPlatforms);
|
||
|
||
foreach ($missingPlatforms as $platform) {
|
||
$externalId = $missingIdMap[$platform] ?? null;
|
||
if ($externalId === null) {
|
||
continue;
|
||
}
|
||
$sp = SupplierProject::on(self::DB_CONNECTION)->create([
|
||
'platform' => $platform,
|
||
'signal_type' => (string) $project->signal_type,
|
||
'unique_key' => $identifier,
|
||
'subject_code' => null,
|
||
'supplier_external_id' => (string) $externalId,
|
||
'current_limit' => $shares[$platform] ?? 0,
|
||
'current_workdays' => $workdays,
|
||
'current_regions' => $allRegions,
|
||
'sync_status' => 'ok',
|
||
'last_synced_at' => now(),
|
||
]);
|
||
$existingSps->push($sp);
|
||
}
|
||
}
|
||
|
||
// Update existing supplier projects with current regions/limit.
|
||
foreach ($existingSps as $sp) {
|
||
if ($sp->supplier_external_id === null) {
|
||
continue;
|
||
}
|
||
$perPlatformDto = new SupplierProjectDto(
|
||
platform: $sp->platform,
|
||
signalType: (string) $project->signal_type,
|
||
uniqueKey: $identifier,
|
||
limit: $shares[$sp->platform] ?? 0,
|
||
workdays: $workdays,
|
||
regions: $allRegions,
|
||
regionsReverse: false,
|
||
status: $status,
|
||
tag: $tag,
|
||
platforms: [$sp->platform],
|
||
);
|
||
$channel->updateProject((int) $sp->supplier_external_id, $perPlatformDto);
|
||
$sp->forceFill([
|
||
'current_limit' => $shares[$sp->platform] ?? 0,
|
||
'current_workdays' => $workdays,
|
||
'current_regions' => $allRegions,
|
||
'sync_status' => 'ok',
|
||
'last_synced_at' => now(),
|
||
'inactive_since' => $groupActive ? null : now(),
|
||
])->save();
|
||
}
|
||
}
|
||
|
||
// Pivot: project × each supplier_project → ON CONFLICT DO NOTHING
|
||
foreach ($existingSps as $sp) {
|
||
DB::connection(self::DB_CONNECTION)->table('project_supplier_links')->insertOrIgnore([
|
||
'project_id' => $project->id,
|
||
'supplier_project_id' => $sp->id,
|
||
'platform' => $sp->platform,
|
||
'subject_code' => null,
|
||
]);
|
||
}
|
||
|
||
// Auto-link to root domain (spec 2026-05-22-root-domain-auto-link-design §4.2).
|
||
// Когда identifier — субдомен (krasnoyarsk.carmoney.ru), доп. линкуем проект к
|
||
// supplier_projects корневого домена (carmoney.ru), если такие есть. Закрывает
|
||
// класс «поставщик шлёт корень — подписчики на субдомены не получают».
|
||
if ($project->signal_type === 'site') {
|
||
$rootIdentifier = SupplierIdentifier::extractRootDomain(
|
||
(string) $project->signal_identifier
|
||
);
|
||
if ($rootIdentifier !== null) {
|
||
$rootSps = SupplierProject::on(self::DB_CONNECTION)
|
||
->where('unique_key', $rootIdentifier)
|
||
->where('signal_type', 'site')
|
||
->get();
|
||
foreach ($rootSps as $rootSp) {
|
||
DB::connection(self::DB_CONNECTION)->table('project_supplier_links')->insertOrIgnore([
|
||
'project_id' => $project->id,
|
||
'supplier_project_id' => $rootSp->id,
|
||
'platform' => $rootSp->platform,
|
||
'subject_code' => null,
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Mirror the link into the legacy FK columns (supplier_b{1,2,3}_project_id) so the
|
||
// UI sync-status (ProjectResource → aggregateSyncStatus, which reads supplierB1/B2/B3)
|
||
// reflects the synced stack in online mode too — online primarily uses the pivot.
|
||
foreach ($existingSps as $sp) {
|
||
$column = 'supplier_'.strtolower((string) $sp->platform).'_project_id';
|
||
$project->{$column} = $sp->id;
|
||
}
|
||
$project->save();
|
||
}
|
||
|
||
// -------------------------------------------------------------------------
|
||
// Batch mode: каркас (limit=0, no regions) — backward-compat
|
||
// -------------------------------------------------------------------------
|
||
|
||
private function handleBatch(Project $project, SupplierProjectChannel $channel): void
|
||
{
|
||
$platforms = SupplierProjectGrouping::resolvePlatforms($project);
|
||
$workdays = $this->workdaysFromMask((int) $project->delivery_days_mask);
|
||
|
||
foreach ($platforms as $platform) {
|
||
$uniqueKey = SupplierProjectGrouping::buildUniqueKey($project, $platform);
|
||
$column = 'supplier_'.strtolower($platform).'_project_id';
|
||
|
||
// Idempotency: local supplier_projects-запись уже есть?
|
||
$existing = SupplierProject::on(self::DB_CONNECTION)
|
||
->where('platform', $platform)
|
||
->where('signal_type', $project->signal_type)
|
||
->where('unique_key', $uniqueKey)
|
||
->first();
|
||
|
||
if ($existing !== null) {
|
||
$project->{$column} = $existing->id;
|
||
|
||
continue;
|
||
}
|
||
|
||
$dto = new SupplierProjectDto(
|
||
platform: $platform,
|
||
signalType: (string) $project->signal_type,
|
||
uniqueKey: $uniqueKey,
|
||
limit: 0,
|
||
workdays: $workdays,
|
||
regions: [],
|
||
regionsReverse: false,
|
||
status: 'active',
|
||
);
|
||
|
||
try {
|
||
$externalId = $channel instanceof FailoverProjectChannel
|
||
? $channel->createProjectForLiderra($project, $dto)
|
||
: $channel->createProject($dto);
|
||
} catch (TierEscalatedException $e) {
|
||
Log::info("SyncSupplierProjectJob: project {$project->id} {$platform} escalated to manual queue #{$e->queueRowId}");
|
||
|
||
continue;
|
||
} catch (WindowDeferredException) {
|
||
Log::info("SyncSupplierProjectJob: project {$project->id} {$platform} deferred by portal window");
|
||
|
||
continue;
|
||
}
|
||
|
||
$sp = SupplierProject::on(self::DB_CONNECTION)->create([
|
||
'platform' => $platform,
|
||
'signal_type' => $project->signal_type,
|
||
'unique_key' => $uniqueKey,
|
||
'supplier_external_id' => (string) $externalId,
|
||
'current_limit' => 0,
|
||
'current_workdays' => $workdays,
|
||
'current_regions' => null,
|
||
'sync_status' => 'ok',
|
||
]);
|
||
|
||
$project->{$column} = $sp->id;
|
||
}
|
||
|
||
$project->save();
|
||
}
|
||
|
||
/**
|
||
* Создаёт проекты на портале ПО ОДНОМУ на платформу с её долей лимита ($shares).
|
||
*
|
||
* Один single-flag save = ровно один rt-проект → надёжный id через listProjects-матч.
|
||
* Так per-platform лимит = доля (Σ == заказу), а не полный лимит на каждой площадке.
|
||
* Per-platform tolerance: tier-escalation / window-defer / прочая ошибка одной площадки
|
||
* не валит остальные — пропускаем, следующий run (или ночной батч) подберёт недостающее.
|
||
*
|
||
* @param array<string, int> $shares [platform => лимит площадки]
|
||
* @param list<string> $platformsToCreate
|
||
* @return array<string, int> [platform => external_id] для успешно созданных
|
||
*/
|
||
private function createPerPlatform(
|
||
SupplierPortalClient $client,
|
||
Project $project,
|
||
string $identifier,
|
||
string $tag,
|
||
array $workdays,
|
||
array $allRegions,
|
||
array $shares,
|
||
array $platformsToCreate,
|
||
): array {
|
||
$idMap = [];
|
||
|
||
foreach ($platformsToCreate as $platform) {
|
||
$dto = new SupplierProjectDto(
|
||
platform: $platform,
|
||
signalType: (string) $project->signal_type,
|
||
uniqueKey: $identifier,
|
||
limit: $shares[$platform] ?? 0,
|
||
workdays: $workdays,
|
||
regions: $allRegions,
|
||
regionsReverse: false,
|
||
status: 'active',
|
||
tag: $tag,
|
||
platforms: [$platform],
|
||
);
|
||
|
||
try {
|
||
$result = $client->saveProjectMultiFlag($dto);
|
||
} catch (TierEscalatedException $e) {
|
||
Log::info("SyncSupplierProjectJob: project {$project->id} {$platform} escalated to manual queue #{$e->queueRowId}");
|
||
|
||
continue;
|
||
} catch (WindowDeferredException) {
|
||
Log::info("SyncSupplierProjectJob: project {$project->id} {$platform} deferred by portal window");
|
||
|
||
continue;
|
||
} catch (\Throwable $e) {
|
||
Log::warning("SyncSupplierProjectJob: online per-platform save failed for project {$project->id} {$platform} (".get_class($e).'): '.$e->getMessage());
|
||
|
||
continue;
|
||
}
|
||
|
||
if (isset($result[$platform])) {
|
||
$idMap[$platform] = $result[$platform];
|
||
}
|
||
}
|
||
|
||
return $idMap;
|
||
}
|
||
|
||
/**
|
||
* Bitmask → ISO weekday list. bit 0 = Mon (ISO 1) … bit 6 = Sun (ISO 7).
|
||
*
|
||
* Mirror of SyncSupplierProjectsJob::bitmaskToList(). Kept inline (not
|
||
* extracted to a shared helper) to keep this fix surgical.
|
||
*
|
||
* @return list<int>
|
||
*/
|
||
private function workdaysFromMask(int $mask): array
|
||
{
|
||
$out = [];
|
||
for ($i = 0; $i < 7; $i++) {
|
||
if (($mask & (1 << $i)) !== 0) {
|
||
$out[] = $i + 1;
|
||
}
|
||
}
|
||
|
||
return $out;
|
||
}
|
||
}
|