<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use PDOException;
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);
        $started = microtime(true);
        $stage = 'read_upstream';

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

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

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

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

                if ($batchRows < $batchSize) {
                    $stage = 'complete_mirror';
                    $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) {
            // PDOException alone cannot distinguish an upstream timeout from a
            // local transaction failure. Keep the stage and numeric diagnostics,
            // never the exception message, SQL, bindings or connection details.
            $sqlState = null;
            $driverCode = null;
            for ($cause = $exception; $cause !== null; $cause = $cause->getPrevious()) {
                if ($cause instanceof PDOException) {
                    $code = (string) ($cause->errorInfo[0] ?? $cause->getCode());
                    $sqlState = preg_match('/^[A-Z0-9]{5}$/D', $code) === 1 ? $code : null;
                    $native = $cause->errorInfo[1] ?? null;
                    $driverCode = is_numeric($native) ? (int) $native : null;
                }
            }
            Log::warning('共享业务镜像同步阶段失败。', [
                'dataset' => $dataset->key(),
                'source' => $dataset->sourceKey(),
                'stage' => $stage,
                'sql_state' => $sqlState,
                'driver_code' => $driverCode,
                'elapsed_seconds' => round(microtime(true) - $started, 3),
                'exception' => $exception::class,
            ]);
            $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 = [];

        [$excludedDates, $excludedBindings] = $this->excludedDateRange($dataset);
        if ($excludedDates !== null) {
            $where[] = $excludedDates;
            array_push($bindings, ...$excludedBindings);
        }

        [$minimumDatetime, $minimumBindings] = $this->minimumDatetimeFilter($dataset, $columns);
        if ($minimumDatetime !== null) {
            $where[] = $minimumDatetime;
            array_push($bindings, ...$minimumBindings);
        }

        [$parentMinimumDatetime, $parentMinimumBindings] = $this->parentMinimumDatetimeFilter($dataset);
        if ($parentMinimumDatetime !== null) {
            $where[] = $parentMinimumDatetime;
            array_push($bindings, ...$parentMinimumBindings);
        }

        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,
        );

        if ($this->driver === 'sqlsrv' && $mode === 'incremental'
            && in_array($dataset->key(), (array) config('upstream_mirrors.table_sync.key_first_datasets', []), true)) {
            $keys = (new UpstreamSqlDialect($this->driver))->select(
                implode(', ', array_map(fn (string $name): string => $this->quote($name), $order)), $tail, $limit,
            );
            $keyColumn = fn (string $name): string => $this->quote('mirror_keys') . '.' . $this->quote($name);
            $rowColumn = fn (string $name): string => $this->quote('mirror_rows') . '.' . $this->quote($name);
            $sql = 'SELECT ' . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column, 'mirror_rows'), $columns))
                . ' FROM (' . $keys . ') AS ' . $this->quote('mirror_keys')
                . ' INNER LOOP JOIN ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject())
                . ' AS ' . $this->quote('mirror_rows')
                . ' ON ' . implode(' AND ', array_map(fn (string $name): string => $rowColumn($name) . ' = ' . $keyColumn($name), $order))
                . ' ORDER BY ' . implode(', ', array_map($keyColumn, $order));
        }

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

    /** @return array{0:?string,1:list<string>} */
    private function excludedDateRange(BusinessMirrorDataset $dataset): array
    {
        $range = $this->datasets->definitionValue($dataset->key(), 'excluded_date_range');
        if ($range === null) {
            return [null, []];
        }

        $from = (string) ($range['from'] ?? '');
        $to = (string) ($range['to'] ?? '');
        if (! preg_match('/^\d{4}-\d{2}-\d{2}$/D', $from)
            || ! preg_match('/^\d{4}-\d{2}-\d{2}$/D', $to) || $from >= $to) {
            throw new \InvalidArgumentException('镜像排除日期范围无效。');
        }
        $column = $this->quote((string) $range['column']);
        if (! isset($range['parent_dataset'])) {
            return ["({$column} IS NULL OR {$column} < ? OR {$column} >= ?)", [$from, $to]];
        }

        $parent = $this->datasets->get((string) $range['parent_dataset']);
        if ($parent->sourceKey() !== $dataset->sourceKey()) {
            throw new \InvalidArgumentException('镜像日期关联必须使用同一上游来源。');
        }
        $table = $this->quote($parent->sourceSchema()) . '.' . $this->quote($parent->sourceObject());
        $key = $this->quote((string) $range['key']);
        $outerKey = $this->quote($dataset->sourceObject()) . '.' . $key;
        $old = $this->quote('excluded_parent');
        $kept = $this->quote('retained_parent');

        // Keep orphan details and details shared with a retained parent. Only
        // details belonging exclusively to the excluded dates are removed.
        return [
            "(NOT EXISTS (SELECT 1 FROM {$table} AS {$old} WHERE {$old}.{$key} = {$outerKey}"
                . " AND {$old}.{$column} >= ? AND {$old}.{$column} < ?)"
                . " OR EXISTS (SELECT 1 FROM {$table} AS {$kept} WHERE {$kept}.{$key} = {$outerKey}"
                . " AND ({$kept}.{$column} IS NULL OR {$kept}.{$column} < ? OR {$kept}.{$column} >= ?)))",
            [$from, $to, $from, $to],
        ];
    }

    /** @return array{0:?string,1:list<string>} */
    private function minimumDatetimeFilter(BusinessMirrorDataset $dataset, array $columns): array
    {
        $minimum = $this->datasets->minimumDatetime($dataset->key());
        if ($minimum === null) {
            return [null, []];
        }

        $available = array_map(
            static fn (array $column): string => strtolower((string) ($column['column_name'] ?? '')),
            $columns,
        );
        if (! in_array(strtolower($minimum['column']), $available, true)) {
            throw new RuntimeException('business_mirror_minimum_datetime_column_missing');
        }

        return [
            $this->quote($minimum['column']) . ' >= ?',
            [$this->minimumDatetimeBinding($minimum['value'])],
        ];
    }

    /** @return array{0:?string,1:list<string>} */
    private function parentMinimumDatetimeFilter(BusinessMirrorDataset $dataset): array
    {
        $minimum = $this->datasets->parentMinimumDatetime($dataset->key());
        if ($minimum === null) {
            return [null, []];
        }

        $parent = $this->datasets->get($minimum['parent_dataset']);
        $parentTable = $this->quote($parent->sourceSchema()) . '.' . $this->quote($parent->sourceObject());
        $alias = $this->quote('minimum_parent');
        $parentKey = $alias . '.' . $this->quote($minimum['parent_key']);
        $outerKey = $this->quote($dataset->sourceObject()) . '.' . $this->quote($minimum['child_key']);
        $column = $alias . '.' . $this->quote($minimum['column']);

        return [
            "EXISTS (SELECT 1 FROM {$parentTable} AS {$alias}"
                . " WHERE {$parentKey} = {$outerKey} AND {$column} >= ?)",
            [$this->minimumDatetimeBinding($minimum['value'])],
        ];
    }

    private function minimumDatetimeBinding(string $value): string
    {
        $date = Carbon::createFromFormat('!Y-m-d H:i:s', $value);
        if ($date === false) {
            throw new RuntimeException('business_mirror_minimum_datetime_invalid');
        }

        return (string) $this->sqlServerDateTime($date);
    }

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

        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({$reference} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}";
            }
            if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') {
                return "encode({$reference}, 'hex') AS {$quoted}";
            }
            return $reference;
        }

        // 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), ' . $reference . ', 23) AS ' . $quoted,
            'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $reference . ', 126) AS ' . $quoted,
            default => $reference,
        };
    }

    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()];
    }
}