<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Carbon;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use RuntimeException;
use Throwable;

class BusinessMirrorSynchronizer
{
    private string $driver = 'sqlsrv';
    public function __construct(
        private readonly BusinessMirrorDatasetRegistry $datasets,
        private readonly BusinessMirrorSchemaManager $schemas,
        private readonly GenericBusinessMirrorRepository $mirror,
        private readonly UpstreamReadOnlyQueryFactory $queries,
    ) {
    }

    /** @return array{status:string,mode:string,batches:int,rows:int,dataset:string} */
    public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array
    {
        $dataset = $this->datasets->get($datasetKey);
        $this->driver = app(UpstreamSourceRegistry::class)->driver($dataset->sourceKey());
        // A long-running scheduler may have retained a timed-out PDO instance
        // from the previous dataset. Each dataset starts with a fresh upstream
        // connection so one source failure cannot poison subsequent mirrors.
        UpstreamConnection::reset($dataset->sourceKey());
        $state = $this->mirror->state($dataset);

        if (! $full && ($state['initial_completed_at'] ?? null) === null) {
            return $this->result($dataset, 'requires_full_sync', 'incremental', 0, 0);
        }

        $columns = $this->schemas->ensure($dataset);

        $mode = $full ? 'full' : 'incremental';
        $resume = is_array($state)
            && ($state['run_mode'] ?? null) === $mode
            && in_array($state['status'] ?? null, ['running', 'failed'], true)
            && is_array($state['cursor'] ?? null);
        $generation = $resume && is_string($state['generation'] ?? null)
            ? (string) $state['generation']
            : (string) Str::uuid();
        [$windowStart, $windowEnd] = $this->window($dataset, $state, $mode, $resume);
        $state = $this->mirror->startRun(
            $dataset,
            $mode,
            $generation,
            $windowStart?->format('Y-m-d H:i:s.u'),
            $windowEnd?->format('Y-m-d H:i:s.u'),
            $resume,
        );
        $cursor = is_array($state['cursor'] ?? null) ? $state['cursor'] : null;
        $highWatermark = is_array($state['high_watermark'] ?? null) ? $state['high_watermark'] : [];
        // Routine runs keep the configured 500-row default. A larger explicit
        // batch is reserved for supervised first syncs so wide SQL Server views
        // need fewer repeated keyset scans without creating unbounded PHP arrays.
        $batchSize = min(max($batchSize ?? (int) config('upstream_mirrors.table_sync.batch_size', 500), 1), 5000);
        $maxBatches = max($maxBatches ?? (int) config('upstream_mirrors.table_sync.max_batches_per_run', 2), 0);
        $batches = 0;
        $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1);
        $rows = (int) ($state['last_upserted_rows'] ?? 0);

        try {
            while (true) {
                $batch = $this->readBatch(
                    $dataset,
                    $columns,
                    $mode,
                    $cursor,
                    $highWatermark,
                    $windowStart,
                    $windowEnd,
                    $batchSize,
                );

                if ($batch === []) {
                    $this->mirror->complete($dataset, $mode, $generation, $cursor ?? $highWatermark, $rows);

                    return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
                }

                $cursor = $this->cursor($dataset, $batch[array_key_last($batch)]);
                $rows += count($batch);
                $batches++;
                $this->mirror->storeBatch($dataset, $columns, $this->normalize($columns, $batch), $cursor, $generation, $rows);

                if (count($batch) < $batchSize) {
                    $this->mirror->complete($dataset, $mode, $generation, $cursor, $rows);

                    return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
                }

                if ($maxBatches > 0 && ($batches >= $maxBatches || microtime(true) >= $deadline)) {
                    return $this->result($dataset, 'running', $mode, $batches, $rows);
                }
            }
        } catch (Throwable $exception) {
            $this->mirror->fail($dataset, $exception::class);
            throw $exception;
        }
    }

    /** @return list<array<string,mixed>> */
    private function readBatch(
        BusinessMirrorDataset $dataset,
        array $columns,
        string $mode,
        ?array $cursor,
        array $highWatermark,
        ?Carbon $windowStart,
        ?Carbon $windowEnd,
        int $limit,
    ): array {
        $order = $this->orderColumns($dataset);
        $where = [];
        $bindings = [];

        foreach ($dataset->fixedFilters() as $column => $value) {
            if (is_array($value)) {
                $where[] = $this->quote($column) . ' IN (' . implode(', ', array_fill(0, count($value), '?')) . ')';
                array_push($bindings, ...$value);
            } else {
                $where[] = $this->quote($column) . ' = ?';
                $bindings[] = $value;
            }
        }

        if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) {
            $watermark = $this->quote($dataset->watermarkColumn());
            $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?';
            $timezoneAware = $this->driver === 'pgsql' && collect($columns)->contains(fn ($c) =>
                $c['column_name'] === $dataset->watermarkColumn() && $c['data_type'] === 'timestamp with time zone'
            );
            $bindings[] = $timezoneAware ? $windowStart?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowStart);
            $bindings[] = $timezoneAware ? $windowEnd?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowEnd);
        } elseif ($mode === 'incremental' && $cursor === null && $highWatermark !== []) {
            [$predicate, $values] = $this->keyset($dataset->keyColumns(), $highWatermark);
            $where[] = $predicate;
            array_push($bindings, ...$values);
        }

        if ($cursor !== null) {
            [$predicate, $values] = $this->keyset($order, $cursor);
            $where[] = $predicate;
            array_push($bindings, ...$values);
        }

        $tail = 'FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject())
            . ($where === [] ? '' : ' WHERE (' . implode(') AND (', $where) . ')')
            . ' ORDER BY ' . implode(', ', array_map(fn (string $name): string => $this->quote($name), $order));
        $sql = (new UpstreamSqlDialect($this->driver))->select(
            implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit,
        );

        return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings);
    }

    /** @param array<string,mixed> $column */
    private function selectExpression(array $column): string
    {
        $name = (string) $column['column_name'];
        $quoted = $this->quote($name);

        if ($this->driver !== 'sqlsrv') {
            if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') {
                // Store timezone-aware PostgreSQL instants as UTC in the shared mirror.
                return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}";
            }
            if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') {
                return "encode({$quoted}, 'hex') AS {$quoted}";
            }
            return $quoted;
        }

        // FreeTDS may return SQL Server date values using the connection locale
        // (for example "Jan 1 2026"). Convert temporal columns at the source so
        // every mirror receives deterministic ISO values that MariaDB accepts.
        return match (strtolower((string) ($column['data_type'] ?? ''))) {
            'date' => 'CONVERT(varchar(10), ' . $quoted . ', 23) AS ' . $quoted,
            'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $quoted . ', 126) AS ' . $quoted,
            default => $quoted,
        };
    }

    private function sqlServerDateTime(?Carbon $value): ?string
    {
        // ISO 8601 with millisecond precision is accepted consistently by SQL Server
        // through FreeTDS. Six-digit fractions were interpreted as an invalid date by
        // some upstream datetime columns and broke every incremental run.
        return $value?->format('Y-m-d\TH:i:s.v');
    }

    /** @return array{0:string,1:list<mixed>} */
    private function keyset(array $columns, array $cursor): array
    {
        $parts = [];
        $bindings = [];

        foreach (array_values($columns) as $index => $column) {
            $and = [];
            for ($previous = 0; $previous < $index; $previous++) {
                $name = $columns[$previous];
                $and[] = $this->quote($name) . ' = ?';
                $bindings[] = $this->cursorValue($cursor, $name);
            }

            $and[] = $this->quote($column) . ' > ?';
            $bindings[] = $this->cursorValue($cursor, $column);
            $parts[] = '(' . implode(' AND ', $and) . ')';
        }

        return ['(' . implode(' OR ', $parts) . ')', $bindings];
    }

    /** @return list<string> */
    private function orderColumns(BusinessMirrorDataset $dataset): array
    {
        // The watermark belongs in the incremental WHERE window only. Sorting
        // a very large SQL Server view by an unindexed timestamp forces an
        // expensive full-window sort on every keyset page and eventually times
        // out. Dataset keys are the stable, unique pagination contract.
        return array_values(array_unique($dataset->keyColumns()));
    }

    /** @return array<string,mixed> */
    private function cursor(BusinessMirrorDataset $dataset, array $row): array
    {
        $cursor = [];
        foreach ($this->orderColumns($dataset) as $column) {
            $cursor[$column] = $this->rowValue($row, $column);
        }

        return $cursor;
    }

    /** @return array{0:?Carbon,1:?Carbon} */
    private function window(BusinessMirrorDataset $dataset, ?array $state, string $mode, bool $resume): array
    {
        if ($mode === 'full' || $dataset->watermarkColumn() === null) {
            return [null, null];
        }

        if ($resume && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null) {
            return [Carbon::parse((string) $state['window_started_at']), Carbon::parse((string) $state['window_ends_at'])];
        }

        $overlap = max((int) config('upstream_mirrors.table_sync.overlap_hours', 48), 1);
        $last = Carbon::parse((string) ($state['last_succeeded_at'] ?? now()->subHours($overlap)));

        return [$last->subHours($overlap), now()->addSecond()];
    }

    /** @return list<array<string,mixed>> */
    private function normalize(array $columns, array $rows): array
    {
        $types = [];
        foreach ($columns as $column) {
            $types[strtolower((string) $column['column_name'])] = strtolower((string) $column['data_type']);
        }

        return array_map(function (array $row) use ($types): array {
            foreach ($row as $name => $value) {
                if (! is_string($value)) {
                    continue;
                }

                $type = $types[strtolower((string) $name)] ?? '';
                if ($type === 'date') {
                    $row[$name] = substr($value, 0, 10);
                } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone'], true)) {
                    $row[$name] = preg_replace('/(\.\d{6})\d+/', '$1', str_replace('T', ' ', $value));
                } elseif (in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'text', 'ntext', 'xml'], true)) {
                    $row[$name] = $this->normalizeText($value);
                }
            }

            return $row;
        }, $rows);
    }

    private function normalizeText(string $value): string
    {
        // A small number of legacy SQL Server rows contain embedded NUL bytes or
        // malformed byte sequences. MariaDB correctly rejects those values as
        // invalid utf8mb4. Do not guess a source encoding: preserve all valid
        // UTF-8 and mark only irrecoverable bytes with the Unicode replacement
        // character so one corrupt label cannot block the entire table mirror.
        $value = str_replace("\0", '', $value);

        return mb_check_encoding($value, 'UTF-8')
            ? $value
            : mb_scrub($value, 'UTF-8');
    }

    private function quote(string $identifier): string
    {
        return (new UpstreamSqlDialect($this->driver))->quote($identifier);
    }

    private function cursorValue(array $cursor, string $column): mixed
    {
        foreach ($cursor as $key => $value) {
            if (strcasecmp((string) $key, $column) === 0) {
                return $value;
            }
        }

        throw new RuntimeException('business_mirror_cursor_incomplete');
    }

    private function rowValue(array $row, string $column): mixed
    {
        foreach ($row as $key => $value) {
            if (strcasecmp((string) $key, $column) === 0) {
                return $value;
            }
        }

        throw new RuntimeException('business_mirror_row_key_missing');
    }

    private function result(BusinessMirrorDataset $dataset, string $status, string $mode, int $batches, int $rows): array
    {
        return ['status' => $status, 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset->key()];
    }
}