<?php

namespace App\Modules\Stock;

use App\Contracts\StockProvider;
use App\Events\StockSyncFinished;
use App\Facades\Settings;
use App\Imports\AutoGrabToVehicleImport;
use App\Jobs\ImportAutoGrabFeedBatch;
use App\Services\AutoGrab\AutoGrabFeedStorage;
use App\Services\AutoGrab\ParquetFeedReader;
use App\Traits\TracksImportStats;
use Illuminate\Foundation\Bus\DispatchesJobs;
use Illuminate\Support\Facades\Event;
use Illuminate\Support\Facades\File;
use Illuminate\Support\Facades\Log;
use Mtc\MercuryDataModels\Vehicle;

use function tenant;

/**
 * AutoGrab is a pass-through stock provider driven by AutoGrab's Saxtons UK Parquet feed.
 *
 * Real-time removals still arrive via the DEACTIVATE-only webhook (see
 * AutoGrabNotificationController); new stock and updates come from the daily Reconcile file,
 * which is the authoritative snapshot of currently-active leads (SFSSA-114).
 *
 * Like Motors, the feed is pulled to local disk, processed off a batch queue and the local
 * copy removed. It runs inside the scheduled stock-provider workflow (RunScheduledSyncTask
 * via sync:vehicles), not a standalone cron. Because that workflow runs hourly while the
 * Reconcile is produced daily, each partition is processed at most once.
 */
class AutoGrab implements StockProvider
{
    use DispatchesJobs;
    use TracksImportStats;

    private const PROVIDER = 'autograb';

    /** Rows per dispatched batch job (matches the Motors batch size). */
    private const BATCH_SIZE = 200;

    /** Setting holding the last Reconcile partition processed, so hourly runs are idempotent. */
    public const LAST_PARTITION_SETTING = 'stock-autograb-feed-last-partition';

    private ?AutoGrabFeedStorage $storage = null;

    public function enabled(): bool
    {
        return Settings::get('stock-autograb-enabled') ?? false;
    }

    public function name(): string
    {
        return 'AutoGrab';
    }

    /**
     * Scheduled workflow entry point: process today's Reconcile snapshot once.
     */
    public function runScheduledImport(bool $fullSync = true): void
    {
        $this->resetStats();

        if (!$this->feedConfigured()) {
            $this->log('skipped: GCS key file not configured (AUTOGRAB_GCS_KEY_FILE missing/unreadable)');
            return;
        }

        $partition = $this->storage()->latestPartition('reconcile');
        if ($partition === null) {
            $this->log('reconcile: no partitions found under the feed prefix');

            // Diagnostic: the scheduled sync only consumes reconcile, but if the feed is currently
            // only producing incremental exports, surface that so an empty run isn't mistaken for a
            // bad GCS connection or wrong prefix.
            $incremental = $this->storage()->listPartitions('incremental');
            if ($incremental !== []) {
                $this->log('incremental partitions present but not consumed by the scheduled sync', [
                    'count' => count($incremental),
                    'latest' => end($incremental),
                ]);
            } else {
                $this->log('incremental: no partitions found under the feed prefix either');
            }

            return;
        }

        // The Reconcile is a daily file but this workflow runs hourly; skip a partition we
        // have already applied so we don't re-download and re-converge every hour.
        if (Settings::get(self::LAST_PARTITION_SETTING) === $partition) {
            $this->log('reconcile: partition already processed, skipping', ['partition' => $partition]);
            return;
        }

        $this->log('reconcile: processing partition', ['partition' => $partition]);
        $dispatched = $this->importPartition($partition, converge: true);

        Settings::update(self::LAST_PARTITION_SETTING, $partition);
        $this->log('reconcile: complete', ['partition' => $partition, 'dispatched' => $dispatched]);
        Event::dispatch(new StockSyncFinished(self::PROVIDER));
    }

    /**
     * One-off onboarding seed from the Backfill file set (run manually via the console
     * command). Backfill carries recent removals inline (removed_at), so it does not run the
     * absence-based convergence pass.
     *
     * @return int Rows dispatched for processing.
     */
    public function importBackfill(): int
    {
        $this->resetStats();

        if (!$this->feedConfigured()) {
            $this->log('backfill skipped: GCS key file not configured (AUTOGRAB_GCS_KEY_FILE missing/unreadable)');
            return 0;
        }

        $partition = $this->storage()->latestPartition('backfill');
        if ($partition === null) {
            $this->log('backfill: no partitions found under the feed prefix');
            return 0;
        }

        $this->log('backfill: processing partition', ['partition' => $partition]);
        $dispatched = $this->importPartition($partition, converge: false);

        $this->log('backfill: complete', ['partition' => $partition, 'dispatched' => $dispatched]);
        Event::dispatch(new StockSyncFinished(self::PROVIDER));

        return $dispatched;
    }

    /**
     * Download the parts of a partition to local disk, dispatch batch jobs to process them,
     * remove the local copy, then (for the Reconcile) converge the active set.
     *
     * @return int Rows dispatched for processing.
     */
    private function importPartition(string $partition, bool $converge): int
    {
        if ($converge) {
            // Clear both flags up front. Each in-feed vehicle is re-marked pending before its
            // batch is dispatched and synced by the batch job; anything left unflagged after
            // dispatch was absent from the snapshot and is removed by removeOld().
            Vehicle::query()
                ->where('stock_provider', self::PROVIDER)
                ->update(['was_recently_synced' => false, 'pending_stock_sync' => null]);
        }

        $localDir = $this->downloadDir();
        $dispatched = 0;

        try {
            $parts = $this->storage()->downloadParts($partition, $localDir);
            $this->log('downloaded parts to local disk', ['partition' => $partition, 'parts' => count($parts)]);

            if ($parts === []) {
                $this->log('no part-*.parquet objects in partition; nothing to import', ['partition' => $partition]);
            } else {
                $batch = [];

                foreach (ParquetFeedReader::fromConfig()->readDirectory($localDir) as $row) {
                    $batch[] = $row;

                    if (count($batch) >= self::BATCH_SIZE) {
                        $dispatched += $this->dispatchBatch($batch);
                        $batch = [];
                    }
                }

                if ($batch !== []) {
                    $dispatched += $this->dispatchBatch($batch);
                }

                $this->log('dispatched batch jobs', ['partition' => $partition, 'rows' => $dispatched]);
            }
        } finally {
            // The Parquet parts contain UK personal data (contact_name, postcode, rego);
            // remove the local copy once read.
            File::deleteDirectory($localDir);
        }

        // Only converge when rows were actually dispatched, so a failed download or an empty
        // partition can never wipe live stock.
        if ($converge && $dispatched > 0) {
            $deactivated = new AutoGrabToVehicleImport()->deactivateNotRecentlySynced();
            $this->log('convergence complete', ['deactivated' => $deactivated]);
        }

        return $dispatched;
    }

    private function log(string $message, array $context = []): void
    {
        Log::info('AutoGrab feed: ' . $message, array_merge([
            'tenant' => tenant('id'),
            'bucket' => config('services.autograb.feed.bucket'),
            'prefix' => config('services.autograb.feed.prefix'),
        ], $context));
    }

    /**
     * Protect the batch's in-feed vehicles from convergence and hand the rows to a queue job.
     *
     * @param array<int, array<string, mixed>> $rows
     * @return int Rows dispatched.
     */
    private function dispatchBatch(array $rows): int
    {
        $ids = array_values(array_filter(
            array_map(fn (array $row) => $row['id'] ?? null, $rows),
            fn ($id) => $id !== null,
        ));

        if ($ids !== []) {
            Vehicle::query()
                ->where('stock_provider', self::PROVIDER)
                ->whereIn('uuid', $ids)
                ->update(['pending_stock_sync' => 1]);
        }

        foreach ($rows as $ignored) {
            $this->incrementProcessed();
        }

        $this->dispatch(new ImportAutoGrabFeedBatch($rows));

        return count($rows);
    }

    private function feedConfigured(): bool
    {
        return !empty(config('services.autograb.feed.gcs_key_file'));
    }

    private function storage(): AutoGrabFeedStorage
    {
        return $this->storage ??= AutoGrabFeedStorage::fromConfig();
    }

    private function downloadDir(): string
    {
        return storage_path(implode('/', [
            'app',
            'autograb-feed',
            (string) (tenant('id') ?? 'central'),
            now()->format('YmdHis'),
        ]));
    }
}
