CaiWuXiTong/modules/EcommerceFinance/Services/ClickHouseClient.php

<?php

namespace Modules\EcommerceFinance\Services;

use GuzzleHttp\Client;
use GuzzleHttp\ClientInterface;
use GuzzleHttp\Exception\GuzzleException;
use Psr\Http\Message\StreamInterface;
use RuntimeException;

class ClickHouseClient
{
    private readonly ClientInterface $http;

    /** @param array<string,mixed>|null $configuration */
    public function __construct(?ClientInterface $http = null, private readonly ?array $configuration = null)
    {
        $config = $this->config();
        $this->http = $http ?? new Client([
            'base_uri' => rtrim((string) $config['url'], '/') . '/',
            'connect_timeout' => (float) $config['connect_timeout_seconds'],
            'timeout' => (float) $config['query_timeout_seconds'],
        ]);
    }

    public function enabled(): bool
    {
        return (bool) $this->config()['enabled'];
    }

    public function ping(): bool
    {
        if (! $this->enabled()) {
            return false;
        }

        try {
            $response = $this->http->request('GET', 'ping', $this->authOptions());

            return $response->getStatusCode() === 200
                && trim((string) $response->getBody()) === 'Ok.';
        } catch (GuzzleException) {
            return false;
        }
    }

    /** @param array<string,scalar|null> $parameters */
    public function command(string $sql, array $parameters = []): void
    {
        $this->request($sql, $parameters);
    }

    /** @param array<string,scalar|null> $parameters @return list<array<string,mixed>> */
    public function rows(string $sql, array $parameters = []): array
    {
        $body = (string) $this->request(rtrim($sql, "; \t\n\r") . ' FORMAT JSONEachRow', $parameters);
        if (trim($body) === '') {
            return [];
        }

        $rows = [];
        foreach (preg_split('/\r?\n/', trim($body)) ?: [] as $line) {
            $decoded = json_decode($line, true, 64, JSON_THROW_ON_ERROR);
            if (! is_array($decoded)) {
                throw new RuntimeException('clickhouse_invalid_json_row');
            }
            $rows[] = $decoded;
        }

        return $rows;
    }

    /** @param array<string,scalar|null> $parameters */
    public function scalar(string $sql, array $parameters = []): mixed
    {
        $body = (string) $this->request(rtrim($sql, "; \t\n\r") . ' FORMAT TabSeparatedRaw', $parameters);

        return trim($body);
    }

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

        $table = $this->qualifiedTable($table);
        $columns = array_map([$this, 'identifier'], $columns);
        $payload = '';
        foreach ($rows as $row) {
            $payload .= json_encode($row, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR) . "\n";
        }

        $sql = sprintf('INSERT INTO %s (%s) FORMAT JSONEachRow', $table, implode(', ', $columns));
        $this->request($sql, [], $payload, 'application/x-ndjson');
    }

    /** @param array<string,scalar|null> $parameters */
    public function stream(string $sql, string $format, array $parameters = []): StreamInterface
    {
        if (preg_match('/^[A-Za-z][A-Za-z0-9]*$/', $format) !== 1) {
            throw new RuntimeException('clickhouse_invalid_format');
        }

        return $this->request(
            rtrim($sql, "; \t\n\r") . ' FORMAT ' . $format,
            $parameters,
            null,
            null,
            true,
        );
    }

    public function qualifiedTable(string $table): string
    {
        return $this->identifier((string) $this->config()['database']) . '.' . $this->identifier($table);
    }

    public function identifier(string $value): string
    {
        if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $value) !== 1) {
            throw new RuntimeException('clickhouse_invalid_identifier');
        }

        return '`' . $value . '`';
    }

    /**
     * @param array<string,scalar|null> $parameters
     * @return StreamInterface|string
     */
    private function request(
        string $sql,
        array $parameters = [],
        ?string $body = null,
        ?string $contentType = null,
        bool $stream = false,
    ): StreamInterface|string {
        if (! $this->enabled()) {
            throw new RuntimeException('clickhouse_disabled');
        }

        $query = [
            'database' => (string) $this->config()['database'],
            'wait_end_of_query' => 1,
            'query' => $sql,
        ];
        foreach ($parameters as $name => $value) {
            if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $name) !== 1 || (! is_scalar($value) && $value !== null)) {
                throw new RuntimeException('clickhouse_invalid_parameter');
            }
            $query['param_' . $name] = $value;
        }

        $options = $this->authOptions() + ['query' => $query, 'stream' => $stream];
        if ($stream) {
            $options['timeout'] = (float) $this->config()['stream_timeout_seconds'];
        }
        if ($body !== null) {
            $options['body'] = $body;
        }
        if ($contentType !== null) {
            $options['headers'] = ['Content-Type' => $contentType];
        }

        try {
            $response = $this->http->request('POST', '', $options);
        } catch (GuzzleException $exception) {
            throw new RuntimeException('clickhouse_request_failed', 0, $exception);
        }

        if ($response->getStatusCode() < 200 || $response->getStatusCode() >= 300) {
            throw new RuntimeException('clickhouse_unexpected_status');
        }

        return $stream ? $response->getBody() : (string) $response->getBody();
    }

    /** @return array<string,mixed> */
    private function authOptions(): array
    {
        $username = (string) $this->config()['username'];

        return $username === '' ? [] : ['auth' => [$username, (string) $this->config()['password']]];
    }

    /** @return array<string,mixed> */
    private function config(): array
    {
        return ($this->configuration ?? (array) config('business_data.clickhouse', [])) + [
            'enabled' => false,
            'url' => 'http://clickhouse:8123',
            'database' => 'caiwu_business_analytics',
            'username' => 'caiwu_app',
            'password' => '',
            'connect_timeout_seconds' => 3,
            'query_timeout_seconds' => 30,
            'stream_timeout_seconds' => 1800,
        ];
    }
}