<?php

declare(strict_types=1);

namespace App\Jobs\Supplier;

use App\Jobs\RouteSupplierLeadJob;
use App\Mail\CsvDriftAlertMail;
use App\Models\SupplierLead;
use App\Services\Supplier\SupplierCsvParser;
use App\Services\Supplier\SupplierPortalClient;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Cache\LockProvider;
use Illuminate\Contracts\Mail\Mailer;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Database\QueryException;
use Illuminate\Foundation\Queue\Queueable as FoundationQueueable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Throwable;

/**
 * Hourly CSV reconciliation с порталом поставщика.
 *
 * Spec: docs/superpowers/specs/2026-05-11-plan4-billing-csv-admin-design.md §5.3
 *
 * Алгоритм:
 *   1. Cache::lock на 600s — overlap-защита.
 *   2. INSERT supplier_csv_reconcile_log (status='running').
 *   3. Download CSV за окно 25h.
 *   4. Parse → собираем ['vid' => row].
 *   5. SELECT existing vid'ы из supplier_leads (BYPASSRLS).
 *   6. Diff = missing.
 *   7. Для каждой missing — INSERT supplier_leads (recovered_from_csv_at) + dispatch RouteJob.
 *   8. UPDATE log с метриками + status.
 *   9. drift > 5% → CsvDriftAlertMail + alert_email_sent_at.
 *  10. На exception — status='failed', throw.
 */
final class CsvReconcileJob implements ShouldQueue
{
    use FoundationQueueable;
    use InteractsWithQueue;
    use Queueable;
    use SerializesModels;

    public int $tries = 1;

    public int $timeout = 300;

    private const DB_CONNECTION = 'pgsql_supplier';

    private const DRIFT_THRESHOLD = 0.05;

    private const WINDOW_HOURS = 25;

    private const LOCK_NAME = 'supplier:csv_reconcile';

    private const LOCK_TTL_SECONDS = 600;

    public function handle(
        SupplierPortalClient $portal,
        SupplierCsvParser $parser,
        Mailer $mailer,
    ): void {
        /** @var LockProvider $lockStore */
        $lockStore = Cache::store('redis');
        $lock = $lockStore->lock(self::LOCK_NAME, self::LOCK_TTL_SECONDS);
        if (! $lock->get()) {
            Log::info('csv_reconcile.skipped_overlap');

            return;
        }

        $windowEnd = Carbon::now();
        $windowStart = (clone $windowEnd)->subHours(self::WINDOW_HOURS);

        $logId = DB::connection(self::DB_CONNECTION)
            ->table('supplier_csv_reconcile_log')
            ->insertGetId([
                'started_at' => now(),
                'window_start' => $windowStart,
                'window_end' => $windowEnd,
                'status' => 'running',
                'created_at' => now(),
            ]);

        try {
            $csv = $portal->downloadLeadsCsv($windowStart, $windowEnd);

            /** @var array<string, array<string, mixed>> $csvByVid */
            $csvByVid = [];
            foreach ($parser->parse($csv) as $row) {
                $csvByVid[(string) $row['vid']] = $row;
            }
            $totalCsvRows = count($csvByVid);

            $existing = DB::connection(self::DB_CONNECTION)
                ->table('supplier_leads')
                ->where('received_at', '>=', $windowStart)
                ->where('received_at', '<', $windowEnd->copy()->addHour())
                ->pluck('vid')
                ->map(fn ($v) => (string) $v)
                ->all();

            $existingMap = array_flip($existing);
            $missing = array_diff_key($csvByVid, $existingMap);

            $recoveredCount = 0;
            foreach ($missing as $vid => $row) {
                $platform = $this->extractPlatform((string) ($row['project'] ?? ''));
                if ($platform === null) {
                    Log::warning('csv_reconcile.unparseable_project_skipped', [
                        'vid' => $vid,
                        'project' => $row['project'] ?? null,
                    ]);

                    continue;
                }

                try {
                    $lead = SupplierLead::create([
                        'vid' => (int) $vid,
                        'platform' => $platform,
                        'phone' => (string) $row['phone'],
                        'raw_payload' => $row,
                        'received_at' => Carbon::createFromTimestamp((int) $row['time']),
                        'recovered_from_csv_at' => now(),
                        'source' => 'csv_recovery',
                        'supplier_project_id' => null,  // ResolverStub разрезолвит при RouteJob run
                    ]);
                    RouteSupplierLeadJob::dispatch($lead->id);
                    $recoveredCount++;
                } catch (QueryException $e) {
                    if (str_contains($e->getMessage(), 'unique')) {
                        Log::info('csv_reconcile.duplicate_vid_skipped', ['vid' => $vid]);

                        continue;
                    }
                    throw $e;
                }
            }

            $matchedCount = $totalCsvRows - count($missing);
            $driftRatio = $totalCsvRows > 0 ? count($missing) / $totalCsvRows : 0.0;
            $status = $driftRatio > self::DRIFT_THRESHOLD ? 'drift_alert' : 'ok';

            $update = [
                'finished_at' => now(),
                'total_csv_rows' => $totalCsvRows,
                'matched_count' => $matchedCount,
                'recovered_count' => $recoveredCount,
                'drift_ratio' => $driftRatio,
                'status' => $status,
            ];

            if ($status === 'drift_alert') {
                $mailer->to((string) config('services.supplier.alert_email'))
                    ->send(new CsvDriftAlertMail(
                        reconcileLogId: $logId,
                        totalCsvRows: $totalCsvRows,
                        missingCount: count($missing),
                        recoveredCount: $recoveredCount,
                        driftRatio: $driftRatio,
                        windowStart: $windowStart,
                        windowEnd: $windowEnd,
                    ));
                $update['alert_email_sent_at'] = now();
            }

            DB::connection(self::DB_CONNECTION)
                ->table('supplier_csv_reconcile_log')
                ->where('id', $logId)
                ->update($update);

        } catch (Throwable $e) {
            DB::connection(self::DB_CONNECTION)
                ->table('supplier_csv_reconcile_log')
                ->where('id', $logId)
                ->update([
                    'finished_at' => now(),
                    'status' => 'failed',
                    'error_message' => substr($e->getMessage(), 0, 1000),
                ]);
            throw $e;
        } finally {
            $lock->release();
        }
    }

    /**
     * Извлекает platform (B1/B2/B3) из поля raw_payload['project'] CSV-строки.
     * Формат project: `B[123]_<rest>` (например `B1_a.com`, `B2_79991234567`).
     * Возвращает null если не парсится — caller пропустит строку с warning.
     */
    private function extractPlatform(string $project): ?string
    {
        if (preg_match('/^(B[123])_/', $project, $m) === 1) {
            return $m[1];
        }

        return null;
    }
}
