CaiWuXiTong/modules/EcommerceFinance/Repositories/ColumnarBusinessMirrorRepository.php

<?php

namespace Modules\EcommerceFinance\Repositories;

use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
use Modules\EcommerceFinance\Services\ClickHouseClient;
use Psr\Http\Message\StreamInterface;

class ColumnarBusinessMirrorRepository
{
    public function __construct(private readonly ClickHouseClient $clickHouse)
    {
    }

    /**
     * @param list<array<string,mixed>> $columns
     * @param list<array<string,mixed>> $rows
     */
    public function storeBackfillBatch(
        BusinessMirrorDataset $dataset,
        array $columns,
        array $rows,
        string $generation,
    ): void {
        if ($rows === []) {
            return;
        }

        $names = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
        $types = [];
        foreach ($columns as $column) {
            $types[strtolower((string) $column['column_name'])] = strtolower((string) ($column['data_type'] ?? ''));
        }

        $version = (int) floor(microtime(true) * 1_000_000);
        $mapped = [];
        foreach (array_values($rows) as $index => $row) {
            $values = ['_mirror_key' => (string) $row['_mirror_key']];
            foreach ($names as $name) {
                $values[$name] = $this->value($row[$name] ?? null, $types[strtolower($name)] ?? '');
            }
            $values['_mirror_generation'] = $generation;
            $values['_mirror_synced_at'] = $this->dateTime($row['_mirror_synced_at'] ?? null);
            $values['_mirror_version'] = $version + $index;
            $mapped[] = $values;
        }

        $this->clickHouse->insertJsonEachRow(
            $dataset->targetTable(),
            ['_mirror_key', ...$names, '_mirror_generation', '_mirror_synced_at', '_mirror_version'],
            $mapped,
        );
    }

    public function count(BusinessMirrorDataset $dataset, string $generation): int
    {
        return (int) $this->clickHouse->scalar(
            sprintf(
                'SELECT count() FROM %s FINAL WHERE _mirror_generation = {generation:UUID}',
                $this->clickHouse->qualifiedTable($dataset->targetTable()),
            ),
            ['generation' => $generation],
        );
    }

    public function distinctKeys(BusinessMirrorDataset $dataset, string $generation): int
    {
        return (int) $this->clickHouse->scalar(
            sprintf(
                'SELECT uniqExact(_mirror_key) FROM %s FINAL WHERE _mirror_generation = {generation:UUID}',
                $this->clickHouse->qualifiedTable($dataset->targetTable()),
            ),
            ['generation' => $generation],
        );
    }

    /** @return array{row_count:int,logical_checksum:string} */
    public function partitionStats(BusinessMirrorDataset $dataset, string $generation, int $partition): array
    {
        $rows = $this->clickHouse->rows(
            sprintf(
                'SELECT count() AS row_count, hex(groupBitXor(cityHash64(_mirror_key))) AS logical_checksum '
                . 'FROM %s FINAL WHERE _mirror_generation = {generation:UUID} '
                . 'AND modulo(cityHash64(_mirror_key), 32) = {partition:UInt8}',
                $this->clickHouse->qualifiedTable($dataset->targetTable()),
            ),
            ['generation' => $generation, 'partition' => $this->partition($partition)],
        );
        $row = $rows[0] ?? [];

        return [
            'row_count' => (int) ($row['row_count'] ?? 0),
            'logical_checksum' => (string) ($row['logical_checksum'] ?? ''),
        ];
    }

    public function parquetPartition(
        BusinessMirrorDataset $dataset,
        string $generation,
        int $partition,
    ): StreamInterface
    {
        return $this->clickHouse->stream(
            sprintf(
                'SELECT * EXCEPT (_mirror_version) FROM %s FINAL '
                . 'WHERE _mirror_generation = {generation:UUID} '
                . 'AND modulo(cityHash64(_mirror_key), 32) = {partition:UInt8} ORDER BY _mirror_key',
                $this->clickHouse->qualifiedTable($dataset->targetTable()),
            ),
            'Parquet',
            ['generation' => $generation, 'partition' => $this->partition($partition)],
        );
    }

    private function partition(int $partition): int
    {
        if ($partition < 0 || $partition > 31) {
            throw new \InvalidArgumentException('business_data_invalid_archive_partition');
        }

        return $partition;
    }

    private function value(mixed $value, string $type): mixed
    {
        if ($value === null) {
            return null;
        }
        if (is_resource($value)) {
            $value = stream_get_contents($value);
        }
        if (is_string($value) && in_array($type, ['binary', 'varbinary', 'image', 'timestamp', 'rowversion', 'bytea'], true)) {
            return base64_encode($value);
        }
        if (is_string($value) && ! mb_check_encoding($value, 'UTF-8')) {
            return mb_scrub($value, 'UTF-8');
        }

        return $value;
    }

    private function dateTime(mixed $value): string
    {
        $value = trim((string) ($value ?? ''));

        return $value === '' ? now()->format('Y-m-d H:i:s.u') : str_replace('T', ' ', $value);
    }
}