<?php

namespace App\Services\AutoGrab;

use Google\Cloud\Storage\StorageClient;
use RuntimeException;

/**
 * Read-only access to the AutoGrab Saxtons UK lead feed in Google Cloud Storage.
 *
 * The feed lives under a single AutoGrab-owned bucket (gs://<bucket>/<prefix>/) with
 * one prefix per file set (backfill, incremental, reconcile). Each export writes a
 * dated partition dir containing sharded part-*.parquet objects, e.g.
 *   <prefix>/reconcile/dt=2026-07-01/part-0.parquet
 *   <prefix>/backfill/dt=2026-07-01T120000Z/part-0.parquet
 *
 * We only ever list and download; the service account is granted objectViewer.
 */
class AutoGrabFeedStorage
{
    private ?StorageClient $client = null;

    public function __construct(
        private readonly ?string $keyFilePath,
        private readonly ?string $projectId,
        private readonly string $bucket,
        private readonly string $prefix,
    ) {
    }

    public static function fromConfig(): self
    {
        $feed = config('services.autograb.feed');

        return new self(
            keyFilePath: $feed['gcs_key_file'] ?? null,
            projectId: $feed['gcs_project_id'] ?? null,
            bucket: $feed['bucket'] ?? '',
            prefix: trim((string) ($feed['prefix'] ?? ''), '/'),
        );
    }

    public function keyFilePath(): ?string
    {
        return $this->keyFilePath;
    }

    public function keyFileReadable(): bool
    {
        return !empty($this->keyFilePath) && is_readable($this->keyFilePath);
    }

    /** gs://bucket/prefix, for diagnostics/logging. */
    public function location(): string
    {
        return "gs://{$this->bucket}/{$this->prefix}";
    }

    /**
     * All partition prefixes for a file set, sorted ascending. Partition dir names sort
     * chronologically, so the last element is the newest (dt=YYYY-MM-DD and
     * dt=YYYY-MM-DDTHHMMSSZ both hold this property).
     *
     * @return array<int, string>
     */
    public function listPartitions(string $set): array
    {
        $setPrefix = "{$this->prefix}/{$set}/";
        $partitions = [];

        foreach ($this->objects($setPrefix) as $name) {
            // Take the partition dir immediately beneath the set prefix (the dt=... segment).
            $relative = substr($name, strlen($setPrefix));
            $segment = strstr($relative, '/', true);

            if ($segment !== false && $segment !== '') {
                $partitions["{$setPrefix}{$segment}/"] = true;
            }
        }

        $keys = array_keys($partitions);
        sort($keys);

        return $keys;
    }

    /**
     * Full object prefix of the most recent partition for a file set, or null when the
     * set has no partitions yet.
     */
    public function latestPartition(string $set): ?string
    {
        $partitions = $this->listPartitions($set);

        return $partitions === [] ? null : end($partitions);
    }

    /**
     * Names of the part-*.parquet objects under a partition prefix (not downloaded).
     *
     * @return array<int, string>
     */
    public function listParts(string $partitionPrefix): array
    {
        $parts = [];

        foreach ($this->objects($partitionPrefix) as $name) {
            if (str_ends_with($name, '.parquet') && str_contains(basename($name), 'part-')) {
                $parts[] = $name;
            }
        }

        return $parts;
    }

    /**
     * Download every part-*.parquet object under a partition prefix into $localDir.
     *
     * @return array<int, string> Local file paths of the downloaded parts.
     */
    public function downloadParts(string $partitionPrefix, string $localDir): array
    {
        $this->ensureDir($localDir);

        $bucket = $this->client()->bucket($this->bucket);
        $paths = [];

        foreach ($this->listParts($partitionPrefix) as $name) {
            $local = rtrim($localDir, '/') . '/' . basename($name);
            $bucket->object($name)->downloadToFile($local);
            $paths[] = $local;
        }

        return $paths;
    }

    /**
     * Download a single object to $localDir, returning its local path. Used by diagnostics
     * to sample one shard without pulling a whole partition.
     */
    public function downloadObject(string $objectName, string $localDir): string
    {
        $this->ensureDir($localDir);

        $local = rtrim($localDir, '/') . '/' . basename($objectName);
        $this->client()->bucket($this->bucket)->object($objectName)->downloadToFile($local);

        return $local;
    }

    /**
     * @return iterable<int, string> Object names under a prefix.
     */
    private function objects(string $prefix): iterable
    {
        $bucket = $this->client()->bucket($this->bucket);

        foreach ($bucket->objects(['prefix' => $prefix]) as $object) {
            yield $object->name();
        }
    }

    private function ensureDir(string $dir): void
    {
        if (!is_dir($dir) && !mkdir($dir, 0770, true) && !is_dir($dir)) {
            throw new RuntimeException("Unable to create feed download directory: {$dir}");
        }
    }

    private function client(): StorageClient
    {
        if ($this->client instanceof StorageClient) {
            return $this->client;
        }

        if (!$this->keyFileReadable()) {
            throw new RuntimeException(
                'AutoGrab GCS key file is not configured or not readable. Set AUTOGRAB_GCS_KEY_FILE.'
            );
        }

        $config = ['keyFilePath' => $this->keyFilePath];
        if (!empty($this->projectId)) {
            $config['projectId'] = $this->projectId;
        }

        return $this->client = new StorageClient($config);
    }
}
