<?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()) {
            // Task 4.1: в окне 18:00→00:00 МСК онлайн НЕ шлёт поставщику немедленно —
            // он уже фиксирует заказ слепком в 21:00, мгновенная правка перезаписала бы
            // зафиксированное состояние. Откладываем в очередь; FlushDeferredOnlineSyncJob
            // (00:05 МСК) досылает вне окна.
            if ($this->isInDeferWindow()) {
                DB::connection(self::DB_CONNECTION)
                    ->table('supplier_deferred_sync')
                    ->insertOrIgnore([
                        'project_id' => $project->id,
                        'created_at' => now(),
                    ]);
                Log::info("SyncSupplierProjectJob: project {$project->id} deferred (online window 18:00-00:00 МСК)");

                return;
            }

            $this->handleOnline($project, $channel);
        } else {
            $this->handleBatch($project, $channel);
        }
    }

    /** Час МСК ≥ 18 (окно 18:00→00:00) — онлайн откладывает отправку до полуночи. */
    public const DEFER_WINDOW_START_HOUR_MSK = 18;

    private function isInDeferWindow(): bool
    {
        return Carbon::now('Europe/Moscow')->hour >= self::DEFER_WINDOW_START_HOUR_MSK;
    }

    // -------------------------------------------------------------------------
    // 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;
        }

        // R-17 (Stage 4 §4.4.1): unified agnostic key (was buildUniqueKey($p, $platform[0])
        // which diverged for SMS — B3 used sender alone while B2 used sender+keyword;
        // created orphan supplier_projects rows during sharing rebalance).
        $identifier = SupplierProjectGrouping::buildUniqueKeyAgnostic($project);

        // 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 = $identifier;
        $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 target_date → order/workdays (mirror nightly's eligibility window).
        // R-18 (Stage 4 §4.4.2): see ::targetWeekdayForNow().
        $targetWeekday = self::targetWeekdayForNow();
        $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()
        );
        // Fallback: tomorrow is a non-workday for everyone in the group (e.g. Mon-Fri masks
        // synced on Friday evening). With order=0 the portal would reject the save with
        // "Введите limit!". Use all active limits so the supplier-side identifier exists;
        // per-day eligibility is enforced by workdays we send below.
        if ($order === 0 && $groupActive) {
            $order = SupplierQuotaAllocator::computeOrder(
                $groupProjects->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;
        }

        // Platforms skipped for a transient reason (not escalation/defer) — non-empty at the
        // end (with an active group) means the supplier set is incomplete → throw to retry.
        $retryWorthy = [];

        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).
            $createResult = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $platforms);
            $idMap = $createResult['ids'];
            $retryWorthy = array_merge($retryWorthy, $createResult['failed']);

            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());
                $deadResult = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $deadPlatforms);
                $recreatedIdMap = $deadResult['ids'];
                $retryWorthy = array_merge($retryWorthy, $deadResult['failed']);

                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 !== []) {
                $missingResult = $this->createPerPlatform($client, $project, $identifier, $tag, $workdays, $allRegions, $shares, $missingPlatforms);
                $missingIdMap = $missingResult['ids'];
                $retryWorthy = array_merge($retryWorthy, $missingResult['failed']);

                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;
                }
                // Portal requires a non-zero `limit` even when status=paused — sending 0
                // returns "Введите limit!". When the whole group is paused, keep the previous
                // current_limit so the portal accepts the update; status=paused stops orders.
                $platformLimit = $groupActive
                    ? ($shares[$sp->platform] ?? 0)
                    : max(1, (int) $sp->current_limit);
                $perPlatformDto = new SupplierProjectDto(
                    platform: $sp->platform,
                    signalType: (string) $project->signal_type,
                    uniqueKey: $identifier,
                    limit: $platformLimit,
                    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' => $platformLimit,
                    '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();

        // Atomicity guard: the 3 platforms are created by 3 sequential supplier calls. If one
        // failed for a TRANSIENT reason (network/timeout/5xx/id-not-found), the others are
        // already persisted above (progress kept) — but the supplier set is incomplete and the
        // group under-orders ~1/N. Throw so Laravel retries (backoff 15/60/300s); on retry the
        // partial-set recovery branch fills the missing platform — closing the gap in minutes
        // instead of waiting for the nightly batch. Escalation/window-defer are NOT here (they
        // have their own recovery), so they never trigger a retry.
        if ($retryWorthy !== [] && $groupActive) {
            throw new \RuntimeException(sprintf(
                'SyncSupplierProjectJob: project %d incomplete platform set (transient miss: %s) — retrying for partial-set recovery',
                $project->id,
                implode(',', array_values(array_unique($retryWorthy))),
            ));
        }
    }

    // -------------------------------------------------------------------------
    // 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);

        // R-17 (Stage 4 §4.4.1): same agnostic key for all platforms in this batch run
        // (was per-platform divergence for SMS — created orphan rows).
        $uniqueKey = SupplierProjectGrouping::buildUniqueKeyAgnostic($project);
        foreach ($platforms as $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{ids: array<string, int>, failed: list<string>}
     *                                                              ids    — [platform => external_id] для успешно созданных;
     *                                                              failed — площадки, пропущенные по TRANSIENT-причине (сеть/таймаут/id-not-found),
     *                                                              НЕ из-за escalation/window-defer (у тех свой механизм восстановления).
     *                                                              Непустой failed → handleOnline бросит retry-исключение.
     */
    private function createPerPlatform(
        SupplierPortalClient $client,
        Project $project,
        string $identifier,
        string $tag,
        array $workdays,
        array $allRegions,
        array $shares,
        array $platformsToCreate,
    ): array {
        $idMap = [];
        $legitimateSkips = []; // escalation / window-defer — НЕ retry-worthy

        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}");
                $legitimateSkips[] = $platform;

                continue;
            } catch (WindowDeferredException) {
                Log::info("SyncSupplierProjectJob: project {$project->id} {$platform} deferred by portal window");
                $legitimateSkips[] = $platform;

                continue;
            } catch (\Throwable $e) {
                // Transient (network/timeout/portal 5xx). NOT added to legitimateSkips →
                // remains in `failed` → handleOnline throws → Laravel retry re-runs.
                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];
            }
            // else: save returned no id for this platform (id-not-found in listProjects) —
            // treat as transient: not in idMap, not in legitimateSkips → falls into `failed`.
        }

        $failed = array_values(array_diff($platformsToCreate, array_keys($idMap), $legitimateSkips));

        return ['ids' => $idMap, 'failed' => $failed];
    }

    /**
     * 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;
    }

    /**
     * R-18 (Stage 4 §4.4.2): ISO target weekday for online supplier sync.
     *
     * Slepok cut-off boundary is 21:00 МСК (matches supplier's snapshot fix-point), not midnight.
     *   hour <  21 МСК → target = today + 1 day
     *   hour >= 21 МСК → target = today + 2 days
     *
     * Before fix: `Carbon::tomorrow('Europe/Moscow')->isoWeekday()` flipped target at midnight
     * (Thu 23:59 → Fri; Fri 00:01 → Sat), mis-aligning portal sync with supplier's already-fixed
     * slepok. The post-21:00 portion of day N belongs to slepok dated N+1 (effective day N+2).
     */
    public static function targetWeekdayForNow(): int
    {
        $msk = Carbon::now('Europe/Moscow');

        return $msk->hour >= 21
            ? $msk->copy()->addDays(2)->startOfDay()->isoWeekday()
            : $msk->copy()->addDay()->startOfDay()->isoWeekday();
    }
}
