<?php

namespace App\Console\Commands;

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

use function tenant;

/**
 * Diagnostics for the AutoGrab GCS Parquet feed (SFSSA-114).
 *
 * Walks the whole chain a real import depends on and reports each step, so "0 rows dispatched"
 * can be traced to a concrete cause: config -> key readable -> DuckDB present -> GCS connect ->
 * partitions discovered per set -> part files -> a sample of rows/columns. Read-only; downloads
 * at most one shard (to sample) and removes it afterwards.
 */
class AutoGrabFeedStatus extends Command
{
    protected $signature = 'autograb:feed-status
        {--set=* : Limit to these file sets (default: backfill, incremental, reconcile)}
        {--rows=5 : Sample this many rows from the latest partition of each set (0 to skip)}';

    protected $description = 'Diagnose AutoGrab GCS feed connectivity, access and file discovery';

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

    public function handle(): int
    {
        $feed = config('services.autograb.feed');

        $this->line('Tenant:          ' . (tenant('id') ?? 'central'));
        $this->line('AutoGrab enabled: ' . (Settings::get('stock-autograb-enabled', false) ? 'yes' : 'no'));
        $this->line('Bucket/prefix:   gs://' . ($feed['bucket'] ?? '') . '/' . trim((string) ($feed['prefix'] ?? ''), '/'));
        $this->line('Project ID:      ' . ($feed['gcs_project_id'] ?? '(default from key)'));

        $keyFile = $feed['gcs_key_file'] ?? null;
        $keyReadable = !empty($keyFile) && is_readable($keyFile);
        $this->line('Key file:        ' . ($keyFile ?: '(not set)'));
        $this->line('Key readable:    ' . ($keyReadable ? 'yes' : 'NO'));
        $this->line('DuckDB:          ' . $this->duckdbVersion($feed['duckdb_bin'] ?? 'duckdb'));

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

        $storage = AutoGrabFeedStorage::fromConfig();
        $sets = $this->option('set') ?: self::ALL_SETS;
        $rows = (int) $this->option('rows');

        foreach ($sets as $set) {
            $this->reportSet($storage, $set, $rows);
        }

        return self::SUCCESS;
    }

    private function reportSet(AutoGrabFeedStorage $storage, string $set, int $rows): void
    {
        $this->newLine();
        $this->info("== {$set} ==");

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

        $this->line('  partitions found: ' . count($partitions));
        foreach (array_slice($partitions, -5) as $partition) {
            $this->line('    ' . $partition);
        }

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

        $latest = end($partitions);
        $parts = $storage->listParts($latest);
        $this->line("  latest partition: {$latest}");
        $this->line('  part files:       ' . count($parts));
        foreach (array_slice($parts, 0, 5) as $part) {
            $this->line('    ' . $part);
        }

        if ($rows > 0 && $parts !== []) {
            $this->sampleRows($storage, $parts[0], $rows);
        }
    }

    private function sampleRows(AutoGrabFeedStorage $storage, string $partObject, int $rows): void
    {
        $dir = storage_path('app/autograb-feed-status/' . uniqid('', true));

        try {
            $storage->downloadObject($partObject, $dir);
            $sample = ParquetFeedReader::fromConfig()->peek($dir, $rows);

            $this->line('  sample rows read: ' . count($sample));
            if ($sample !== []) {
                $this->line('  columns (' . count($sample[0]) . '): ' . implode(', ', array_keys($sample[0])));
            }
        } catch (Throwable $exception) {
            $this->error('  sample read failed: ' . $exception->getMessage());
        } finally {
            File::deleteDirectory($dir);
        }
    }

    private function duckdbVersion(string $bin): string
    {
        try {
            $result = Process::run([$bin, '--version']);

            return $result->successful()
                ? trim($result->output())
                : 'NOT AVAILABLE (' . trim($result->errorOutput() ?: 'exit ' . $result->exitCode()) . ')';
        } catch (Throwable $exception) {
            return 'NOT AVAILABLE (' . $exception->getMessage() . ')';
        }
    }
}
