CaiWuXiTong/modules/EcommerceFinance/Services/UpstreamOdbcConnection.php

<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Facades\Log;
use PDO;
use PDOException;
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;

/**
 * 按来源隔离的 SQL Server PDO_ODBC 连接管理。
 */
class UpstreamOdbcConnection
{
    /** @var array<string,PDO> */
    private static array $instances = [];

    public static function get(string $source): PDO
    {
        $definition = app(UpstreamSourceRegistry::class)->definition($source);
        $key = $definition['key'];

        return self::$instances[$key] ??= self::create($key);
    }

    public static function create(string $source): PDO
    {
        return self::connect($source, true);
    }

    public static function probe(string $source): void
    {
        $sources = app(UpstreamSourceRegistry::class);
        $definition = $sources->definition($source);

        self::probeConfiguration(
            (array) $sources->value($source, 'connection', []),
            $definition['label'],
            $definition['key'],
        );
    }

    public static function probeConfiguration(array $connection, string $label, string $source = 'custom'): void
    {
        $pdo = self::open($connection, $source, $label);

        try {
            $statement = $pdo->query('SELECT 1');
            $statement->fetchColumn();
        } catch (PDOException $exception) {
            Log::warning('上游数据库只读连接探测失败。', [
                'source' => $source,
                'source_label' => $label,
                'sql_state' => (string) $exception->getCode(),
                'exception' => $exception::class,
            ]);

            throw new ServiceUnavailableHttpException(60, '上游数据库连接探测失败，请检查配置。');
        }
    }

    private static function connect(string $source, bool $requireEnabled): PDO
    {
        $sources = app(UpstreamSourceRegistry::class);
        $definition = $sources->definition($source);

        if ($requireEnabled && ! $sources->enabled($source)) {
            throw new ServiceUnavailableHttpException(60, '该上游数据同步尚未启用，请联系管理员。');
        }

        return self::open(
            (array) $sources->value($source, 'connection', []),
            $definition['key'],
            $definition['label'],
        );
    }

    private static function open(array $connection, string $source, string $label): PDO
    {
        $host = trim((string) ($connection['host'] ?? ''));
        $port = min(max((int) ($connection['port'] ?? 1433), 1), 65535);
        $db = trim((string) ($connection['database'] ?? ''));
        $user = trim((string) ($connection['username'] ?? ''));
        $pass = (string) ($connection['password'] ?? '');
        $driver = trim((string) ($connection['odbc_driver'] ?? 'FreeTDS'), '{} ');
        $tdsVersion = trim((string) ($connection['tds_version'] ?? '7.4'));
        $connectTimeout = min(max((int) ($connection['connect_timeout'] ?? 5), 1), 30);
        $queryTimeout = min(max((int) ($connection['query_timeout'] ?? 15), 1), 120);
        $applicationIntent = trim((string) ($connection['application_intent'] ?? ''));

        if (
            $host === '' || $db === '' || $user === '' || $pass === '' || $driver === ''
            || preg_match('/[;\x00-\x1F]/', $driver) === 1
            || preg_match('/^\d+(?:\.\d+)?$/', $tdsVersion) !== 1
        ) {
            throw new ServiceUnavailableHttpException(60, '上游数据库配置不完整，请联系管理员。');
        }

        if ($applicationIntent !== '' && strcasecmp($applicationIntent, 'ReadOnly') !== 0) {
            throw new ServiceUnavailableHttpException(60, '上游数据库只允许只读连接意图。');
        }

        $applicationIntent = $applicationIntent === '' ? '' : 'ReadOnly';

        $dsn = sprintf(
            'odbc:Driver={%s};Server=%s;Port=%d;Database=%s;UID=%s;PWD=%s;TDS_Version=%s;ClientCharset=UTF-8;Connect Timeout=%d',
            str_replace('}', '}}', $driver),
            self::quoteOdbcValue($host),
            $port,
            self::quoteOdbcValue($db),
            self::quoteOdbcValue($user),
            self::quoteOdbcValue($pass),
            $tdsVersion,
            $connectTimeout
        );

        if ($applicationIntent !== '') {
            $dsn .= ';ApplicationIntent=' . self::quoteOdbcValue($applicationIntent);
        }

        try {
            return new PDO($dsn, null, null, [
                PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
                PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
                PDO::ATTR_TIMEOUT => $queryTimeout,
                PDO::ATTR_PERSISTENT => false,
            ]);
        } catch (PDOException $exception) {
            Log::warning('上游数据库连接失败。', [
                'source' => $source,
                'source_label' => $label,
                'sql_state' => (string) $exception->getCode(),
                'exception' => $exception::class,
            ]);

            // 不传递原异常，防止异常链携带含凭据的 DSN。
            throw new ServiceUnavailableHttpException(60, '上游数据库暂时无法连接，请稍后再试。');
        }
    }

    public static function reset(?string $source = null): void
    {
        if ($source === null) {
            self::$instances = [];

            return;
        }

        $key = app(UpstreamSourceRegistry::class)->definition($source)['key'];
        unset(self::$instances[$key]);
    }

    private static function quoteOdbcValue(string $value): string
    {
        return '{' . str_replace('}', '}}', $value) . '}';
    }
}