<?php

namespace App\Console\Commands;

use App\Facades\Settings;
use App\Modules\Stock\AutoGrab;
use App\Services\AutoGrab\AutoGrabFeedStorage;
use App\Services\AutoGrab\ParquetFeedReader;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\File;
use Throwable;

/**
 * Debug tool for the AutoGrab GCS Parquet feed (SFSSA-114): find which feed files, if any,
 * contain a given VRM.
 *
 * Walks the partitions of each file set (newest first), downloads their part-*.parquet shards
 * to a temp dir and queries them with DuckDB for rows whose `rego` matches the VRM (case- and
 * space-insensitive). Reconcile partitions are annotated against the last-processed marker so
 * you can tell whether a match sits in a partition the import has actually consumed. Read-only
 * against GCS; every download is removed afterwards.
 */
class AutoGrabFeedFindVrm extends Command
{
    protected $signature = 'autograb:feed-find-vrm
        {vrm* : Registration(s) to search for (case and spaces are ignored)}
        {--set=* : Limit to these file sets (default: backfill, incremental, reconcile)}
        {--partitions=0 : Search only the newest N partitions per set (0 = all)}
        {--column=rego : Feed column to match the value against}
        {--json : Print full matching rows as JSON instead of the summary line}';

    protected $description = 'Search the AutoGrab feed files for specific VRMs';

    /** @var array<string, array<int, string>> Partitions each requested VRM was seen in. */
    private array $seenIn = [];

    private const ALL_SETS = ['backfill', 'incremental', 'reconcile'];

    /** Columns shown per match in summary mode, when present in the row. */
    private const SUMMARY_COLUMNS = ['id', 'rego', 'vin', 'make', 'model', 'updated_at'];

    public function handle(): int
    {
        $storage = AutoGrabFeedStorage::fromConfig();

        if (!$storage->keyFileReadable()) {
            $this->error('Cannot reach GCS: set AUTOGRAB_GCS_KEY_FILE to a readable service-account key.');
            return self::FAILURE;
        }

        $vrms = array_map(
            fn (string $vrm) => strtoupper(str_replace(' ', '', $vrm)),
            (array) $this->argument('vrm'),
        );
        $this->seenIn = array_fill_keys($vrms, []);

        $column = (string) $this->option('column');
        $limit = (int) $this->option('partitions');
        $sets = $this->option('set') ?: self::ALL_SETS;
        $lastProcessed = Settings::get(AutoGrab::LAST_PARTITION_SETTING);

        $this->line("Searching {$storage->location()} for {$column} in [" . implode(', ', $vrms) . ']');

        $reader = ParquetFeedReader::fromConfig();
        $totalMatches = 0;
        $searched = 0;

        foreach ($sets as $set) {
            $this->newLine();
            $this->info("== {$set} ==");

            try {
                $partitions = $storage->listPartitions($set);
            } catch (Throwable $exception) {
                $this->error('  listing failed: ' . $exception->getMessage());
                continue;
            }

            if ($partitions === []) {
                $this->warn('  no partitions under this set prefix.');
                continue;
            }

            // Newest first: recent partitions are the ones a live debugging session cares about.
            $partitions = array_reverse($partitions);
            if ($limit > 0) {
                $skipped = count($partitions) - $limit;
                $partitions = array_slice($partitions, 0, $limit);

                if ($skipped > 0) {
                    $this->line("  (skipping {$skipped} older partition(s); raise --partitions to include them)");
                }
            }

            foreach ($partitions as $partition) {
                $totalMatches += $this->searchPartition($storage, $reader, $partition, $column, $vrms, $lastProcessed);
                $searched++;
            }
        }

        $this->newLine();
        $this->info("Done: {$totalMatches} matching row(s) across {$searched} partition(s) searched.");

        if (count($vrms) > 1) {
            $this->newLine();
            foreach ($this->seenIn as $vrm => $partitions) {
                $partitions = array_unique($partitions);
                $this->line($partitions === []
                    ? "  {$vrm}: NOT FOUND in any searched partition"
                    : "  {$vrm}: in " . count($partitions) . ' partition(s), latest: ' . max($partitions));
            }
        }

        return self::SUCCESS;
    }

    /**
     * @param array<int, string> $vrms
     */
    private function searchPartition(
        AutoGrabFeedStorage $storage,
        ParquetFeedReader $reader,
        string $partition,
        string $column,
        array $vrms,
        ?string $lastProcessed,
    ): int {
        $label = $partition === $lastProcessed ? "{$partition}  <- last processed reconcile" : $partition;
        $localDir = storage_path('app/autograb-feed-find/' . uniqid('', true));

        try {
            $parts = $storage->downloadParts($partition, $localDir);

            if ($parts === []) {
                $this->line("  {$label}: no part files");
                return 0;
            }

            $matches = $reader->findByColumn($localDir, $column, $vrms);
        } catch (Throwable $exception) {
            $this->error("  {$label}: search failed: " . $exception->getMessage());
            return 0;
        } finally {
            File::deleteDirectory($localDir);
        }

        if ($matches === []) {
            $this->line('  ' . $label . ': no match (' . count($parts) . ' part(s))');
            return 0;
        }

        $this->info('  ' . $label . ': ' . count($matches) . ' match(es)');
        foreach ($matches as $row) {
            $this->printMatch($row);

            $matched = strtoupper(str_replace(' ', '', (string) ($row[$column] ?? '')));
            if (array_key_exists($matched, $this->seenIn)) {
                $this->seenIn[$matched][] = $partition;
            }
        }

        return count($matches);
    }

    private function printMatch(array $row): void
    {
        if ($this->option('json')) {
            $this->line(json_encode($row, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES));
            return;
        }

        $summary = [];
        foreach (self::SUMMARY_COLUMNS as $key) {
            if (array_key_exists($key, $row)) {
                $summary[] = "{$key}=" . (is_scalar($row[$key]) ? $row[$key] : json_encode($row[$key]));
            }
        }
        $summary[] = 'file=' . basename((string) ($row['filename'] ?? ''));

        $this->line('    ' . implode('  ', $summary));
    }
}
