<?php

namespace App\Console\Commands;

use GuzzleHttp\Psr7\StreamWrapper;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Facades\Storage;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
use Modules\EcommerceFinance\Services\BusinessDataPlacement;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use Modules\EcommerceFinance\Services\ClickHouseClient;
use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
use Psr\Http\Message\StreamInterface;
use RuntimeException;
use Throwable;

class BackfillBusinessDataColumnar extends Command
{
    protected $signature = 'business-data:backfill-columnar
        {dataset : config/upstream_mirrors.php 中的数据集键}
        {--batch=5000 : 每批从现有 MariaDB 镜像读取的行数}
        {--max-batches=10 : 本轮最多批次数，0 表示直到完成}
        {--restart : 使用新代次从头重建，不删除旧代次}
        {--archive : 对账通过后将 32 个哈希分区分别写入 Parquet 归档}';
    protected $description = '从现有 MariaDB 镜像断点复制到 ClickHouse，不访问上游、不删除原表';

    public function handle(
        BusinessMirrorDatasetRegistry $datasets,
        ClickHouseSchemaManager $clickHouseSchemas,
        ColumnarBusinessMirrorRepository $columnar,
        BusinessDataPlacement $placements,
        ClickHouseClient $clickHouse,
    ): int {
        $datasetKey = (string) $this->argument('dataset');

        try {
            $dataset = $datasets->get($datasetKey);
            $targetStore = $placements->targetStore($datasetKey);
            if (! in_array($targetStore, ['clickhouse', 'parquet'], true)) {
                throw new RuntimeException('business_data_dataset_not_columnar');
            }
            if (! $clickHouse->enabled() || ! $clickHouse->ping()) {
                throw new RuntimeException('business_clickhouse_unavailable');
            }

            $connection = $placements->connection();
            $schema = Schema::connection($connection->getName());
            if (! $schema->hasTable('business_data_placements') || ! $schema->hasTable($dataset->targetTable())) {
                throw new RuntimeException('business_data_source_or_state_table_missing');
            }

            $columns = $clickHouseSchemas->ensure($dataset);
            $columnNames = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
            $state = $placements->state($datasetKey);
            $restart = (bool) $this->option('restart');
            $verified = ! $restart
                && is_array($state)
                && in_array((string) ($state['phase'] ?? ''), ['verified', 'active'], true)
                && ! empty($state['backfill_generation'])
                && ! empty($state['verified_at']);
            if ($verified) {
                $generation = (string) $state['backfill_generation'];
                $this->info(sprintf(
                    '%s：复用已对账代次 %s（%d 行）；MariaDB 原表仍保留。',
                    $datasetKey,
                    $generation,
                    (int) ($state['target_rows'] ?? 0),
                ));
                if ((bool) $this->option('archive')) {
                    $this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
                }

                return self::SUCCESS;
            }
            $resume = ! $restart
                && is_array($state)
                && in_array((string) ($state['phase'] ?? ''), ['backfilling', 'failed'], true)
                && ! empty($state['backfill_generation']);

            $generation = $resume ? (string) $state['backfill_generation'] : (string) Str::uuid();
            $cursor = $resume ? (string) ($state['backfill_cursor'] ?? '') : '';
            $cutoff = $resume && ! empty($state['source_cutoff_at'])
                ? (string) $state['source_cutoff_at']
                : now()->format('Y-m-d H:i:s.u');
            $sourceQuery = $connection->table($dataset->targetTable())
                ->where('_mirror_synced_at', '<=', $cutoff);
            $sourceRows = $resume ? (int) $state['source_rows'] : (int) (clone $sourceQuery)->count();
            $backfilledRows = $resume ? (int) $state['backfilled_rows'] : 0;

            $placements->save($datasetKey, [
                'target_store' => $targetStore,
                'phase' => 'backfilling',
                'backfill_generation' => $generation,
                'backfill_cursor' => $cursor === '' ? null : $cursor,
                'source_cutoff_at' => $cutoff,
                'source_rows' => $sourceRows,
                'backfilled_rows' => $backfilledRows,
                'target_rows' => 0,
                'verified_at' => null,
                'last_error_code' => null,
            ]);

            $batchSize = min(max((int) $this->option('batch'), 100), 10000);
            $maxBatches = max((int) $this->option('max-batches'), 0);
            $batches = 0;

            while ($maxBatches === 0 || $batches < $maxBatches) {
                $query = $connection->table($dataset->targetTable())
                    ->where('_mirror_synced_at', '<=', $cutoff)
                    ->orderBy('_mirror_key')
                    ->limit($batchSize);
                if ($cursor !== '') {
                    $query->where('_mirror_key', '>', $cursor);
                }
                $rows = $query->get(['_mirror_key', ...$columnNames, '_mirror_synced_at'])
                    ->map(static fn (object $row): array => (array) $row)
                    ->all();
                if ($rows === []) {
                    break;
                }

                $columnar->storeBackfillBatch($dataset, $columns, $rows, $generation);
                $cursor = (string) $rows[array_key_last($rows)]['_mirror_key'];
                $backfilledRows += count($rows);
                $batches++;
                $placements->save($datasetKey, [
                    'phase' => 'backfilling',
                    'backfill_generation' => $generation,
                    'backfill_cursor' => $cursor,
                    'source_cutoff_at' => $cutoff,
                    'source_rows' => $sourceRows,
                    'backfilled_rows' => $backfilledRows,
                    'last_error_code' => null,
                ]);

                if (count($rows) < $batchSize) {
                    break;
                }
            }

            if ($backfilledRows < $sourceRows) {
                $this->info(sprintf('%s：本轮 %d 批，已复制 %d/%d 行；可重复执行续传。', $datasetKey, $batches, $backfilledRows, $sourceRows));

                return self::SUCCESS;
            }

            $targetRows = $columnar->count($dataset, $generation);
            $distinctKeys = $columnar->distinctKeys($dataset, $generation);
            if ($targetRows !== $sourceRows || $distinctKeys !== $sourceRows) {
                $mismatch = sprintf(
                    'business_data_backfill_count_mismatch:%d:%d:%d',
                    $sourceRows,
                    $targetRows,
                    $distinctKeys,
                );
                $placements->save($datasetKey, [
                    'phase' => 'failed',
                    'target_rows' => $targetRows,
                    'last_error_code' => $mismatch,
                ]);
                throw new RuntimeException($mismatch);
            }

            $placements->save($datasetKey, [
                'phase' => 'verified',
                'target_rows' => $targetRows,
                'verified_at' => now(),
                'last_error_code' => null,
            ]);
            $this->info(sprintf('%s：MariaDB 与 ClickHouse 已对账，代次 %s 共 %d 行；尚未切换页面或删除原表。', $datasetKey, $generation, $targetRows));

            if ((bool) $this->option('archive')) {
                $this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
            }

            return self::SUCCESS;
        } catch (Throwable $exception) {
            $errorCode = preg_match('/^(business_|clickhouse_)[A-Za-z0-9_:.-]+$/', $exception->getMessage()) === 1
                ? $exception->getMessage()
                : $exception::class;
            if (isset($placements) && Schema::connection($placements->connection()->getName())->hasTable('business_data_placements')) {
                try {
                    $placements->save($datasetKey, [
                        'phase' => 'failed',
                        'last_error_code' => mb_substr($errorCode, 0, 120),
                    ]);
                } catch (Throwable) {
                    // Preserve the original failure when the control store is unavailable.
                }
            }
            report($exception);
            $this->error($datasetKey . '：列式回填失败（' . $errorCode . '），现有 MariaDB 数据未删除、页面未切换。');

            return self::FAILURE;
        }
    }

    private function archive(
        string $datasetKey,
        BusinessMirrorDataset $dataset,
        string $generation,
        ColumnarBusinessMirrorRepository $columnar,
        BusinessDataPlacement $placements,
    ): void {
        $diskName = (string) config('business_data.archive.disk', 'business_uploads_qiniu');
        $prefix = trim((string) config('business_data.archive.prefix', 'business-data-archive'), '/');
        $slug = trim((string) preg_replace('/[^A-Za-z0-9_.-]+/', '-', $datasetKey), '-');
        $disk = Storage::disk($diskName);

        for ($partition = 0; $partition < 32; $partition++) {
            $partitionKey = sprintf('hash-%02d', $partition);
            $objectKey = sprintf('%s/%s/%s/%s.parquet', $prefix, $slug, $generation, $partitionKey);
            $existing = $placements->connection()->table('business_data_archives')
                ->where('dataset_key', $datasetKey)
                ->where('generation', $generation)
                ->where('partition_key', $partitionKey)
                ->first();
            if ($existing !== null
                && (string) ($existing->status ?? '') === 'verified'
                && $disk->exists($objectKey)
                && (int) $disk->size($objectKey) === (int) ($existing->byte_size ?? 0)) {
                continue;
            }
            $stream = $columnar->parquetPartition($dataset, $generation, $partition);
            $resource = $this->seekableResource($stream);
            try {
                $disk->writeStream($objectKey, $resource);
            } finally {
                fclose($resource);
            }

            $bytes = (int) $disk->size($objectKey);
            $stats = $columnar->partitionStats($dataset, $generation, $partition);
            $rowCount = $stats['row_count'];
            if ($bytes <= 0 || ($rowCount > 0 && $bytes < 8)) {
                throw new RuntimeException('business_data_archive_verification_failed');
            }
            $placements->connection()->table('business_data_archives')->updateOrInsert(
                ['dataset_key' => $datasetKey, 'generation' => $generation, 'partition_key' => $partitionKey],
                [
                    'object_key' => $objectKey,
                    'row_count' => $rowCount,
                    'byte_size' => $bytes,
                    'logical_checksum' => $stats['logical_checksum'],
                    'status' => 'verified',
                    'verified_at' => now(),
                    'created_at' => now(),
                    'updated_at' => now(),
                ],
            );
        }

        $this->info($datasetKey . '：32 个 Parquet 哈希分区已写入对象存储并登记清单；MariaDB 原表仍保留。');
    }

    /** @return resource */
    private function seekableResource(StreamInterface $stream): mixed
    {
        $source = StreamWrapper::getResource($stream);
        $target = tmpfile();
        if ($target === false) {
            fclose($source);
            throw new RuntimeException('business_data_archive_temp_file_unavailable');
        }

        try {
            if (stream_copy_to_stream($source, $target) === false || ! rewind($target)) {
                throw new RuntimeException('business_data_archive_spool_failed');
            }
        } catch (Throwable $exception) {
            fclose($target);
            throw $exception;
        } finally {
            fclose($source);
        }

        return $target;
    }
}