<?php

namespace App\Services\AutoGrab;

use Generator;
use Illuminate\Support\Facades\Process;
use RuntimeException;

/**
 * Streams rows out of a set of Parquet parts using the DuckDB CLI.
 *
 * PHP has no native Parquet reader, so we shell out to DuckDB, which reads all shards in
 * a partition together via a glob and writes newline-delimited JSON. We stream that file
 * line by line rather than decoding one large array, so a full daily snapshot (potentially
 * tens of thousands of active leads) never has to sit in memory at once.
 */
class ParquetFeedReader
{
    public function __construct(
        private readonly string $duckdbBin = 'duckdb',
        private readonly int $timeout = 900,
    ) {
    }

    public static function fromConfig(): self
    {
        return new self(
            duckdbBin: (string) (config('services.autograb.feed.duckdb_bin') ?? 'duckdb'),
        );
    }

    /**
     * Read every *.parquet in $directory and yield each row as an associative array.
     *
     * @return Generator<int, array<string, mixed>>
     */
    public function readDirectory(string $directory): Generator
    {
        $glob = rtrim($directory, '/') . '/*.parquet';

        $ndjsonPath = tempnam(sys_get_temp_dir(), 'autograb_ndjson_');
        if ($ndjsonPath === false) {
            throw new RuntimeException('Unable to create a temporary file for the Parquet export.');
        }

        try {
            $this->exportToNdjson($glob, $ndjsonPath);
            yield from $this->streamNdjson($ndjsonPath);
        } finally {
            @unlink($ndjsonPath);
        }
    }

    /**
     * Cheaply read the first $limit rows of the Parquet parts in $directory (uses a SQL
     * LIMIT so it never materialises the whole file). Intended for diagnostics.
     *
     * @return array<int, array<string, mixed>>
     */
    public function peek(string $directory, int $limit = 5): array
    {
        $glob = rtrim($directory, '/') . '/*.parquet';

        $sql = sprintf(
            'SELECT * FROM read_parquet(%s) LIMIT %d',
            $this->quote($glob),
            max(1, $limit),
        );

        $result = Process::timeout($this->timeout)->run([$this->duckdbBin, '-json', '-c', $sql]);

        if (!$result->successful()) {
            throw new RuntimeException(
                "DuckDB failed to read Parquet feed ({$glob}): " . trim($result->errorOutput() ?: $result->output())
            );
        }

        $decoded = json_decode(trim($result->output()) ?: '[]', true);

        return is_array($decoded) ? $decoded : [];
    }

    /**
     * Rows from the Parquet parts in $directory whose $column equals any of $values, compared
     * case-insensitively and ignoring spaces (VRM-style matching). Each row carries a
     * `filename` column identifying the shard it came from. Intended for diagnostics.
     *
     * @param string|array<int, string> $values
     * @return array<int, array<string, mixed>>
     */
    public function findByColumn(string $directory, string $column, string|array $values, int $limit = 50): array
    {
        $glob = rtrim($directory, '/') . '/*.parquet';

        $quoted = array_map(
            fn (string $value) => $this->quote(strtoupper(str_replace(' ', '', $value))),
            (array) $values,
        );

        $sql = sprintf(
            "SELECT * FROM read_parquet(%s, filename=true) "
                . "WHERE upper(replace(CAST(%s AS VARCHAR), ' ', '')) IN (%s) LIMIT %d",
            $this->quote($glob),
            $this->quoteIdentifier($column),
            implode(', ', $quoted),
            max(1, $limit),
        );

        $result = Process::timeout($this->timeout)->run([$this->duckdbBin, '-json', '-c', $sql]);

        if (!$result->successful()) {
            throw new RuntimeException(
                "DuckDB failed to search Parquet feed ({$glob}): " . trim($result->errorOutput() ?: $result->output())
            );
        }

        $decoded = json_decode(trim($result->output()) ?: '[]', true);

        return is_array($decoded) ? $decoded : [];
    }

    private function exportToNdjson(string $glob, string $ndjsonPath): void
    {
        // COPY (...) TO '<file>' (FORMAT JSON) writes one JSON object per line (NDJSON).
        $sql = sprintf(
            'COPY (SELECT * FROM read_parquet(%s)) TO %s (FORMAT JSON)',
            $this->quote($glob),
            $this->quote($ndjsonPath),
        );

        $result = Process::timeout($this->timeout)->run([$this->duckdbBin, '-c', $sql]);

        if (!$result->successful()) {
            throw new RuntimeException(
                "DuckDB failed to read Parquet feed ({$glob}): " . trim($result->errorOutput() ?: $result->output())
            );
        }
    }

    /**
     * @return Generator<int, array<string, mixed>>
     */
    private function streamNdjson(string $ndjsonPath): Generator
    {
        $handle = fopen($ndjsonPath, 'rb');
        if ($handle === false) {
            throw new RuntimeException("Unable to open Parquet export for reading: {$ndjsonPath}");
        }

        try {
            while (($line = fgets($handle)) !== false) {
                $line = trim($line);
                if ($line === '') {
                    continue;
                }

                $row = json_decode($line, true);
                if (is_array($row)) {
                    yield $row;
                }
            }
        } finally {
            fclose($handle);
        }
    }

    /** Quote a value as a DuckDB single-quoted string literal. */
    private function quote(string $value): string
    {
        return "'" . str_replace("'", "''", $value) . "'";
    }

    /** Quote a column name as a DuckDB double-quoted identifier. */
    private function quoteIdentifier(string $identifier): string
    {
        return '"' . str_replace('"', '""', $identifier) . '"';
    }
}
