CaiWuXiTong/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php

<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use RuntimeException;

/**
 * 将上游 SQL Server 的表、视图和字段结构同步到共享系统数据库。
 *
 * 这里只读取 INFORMATION_SCHEMA 元数据，不读取业务行，也不保存连接资料。
 * 获取完整结果成功后才替换单个来源的旧目录，避免网络故障清空现有字典。
 */
class UpstreamSchemaCatalogSynchronizer
{
    public function __construct(
        private readonly UpstreamSourceRegistry $sources,
        private readonly UpstreamReadOnlyQueryFactory $queries,
    ) {
    }

    public function sync(string $source): array
    {
        $definition = $this->sources->definition($source);

        if (! $this->sources->enabled($source)) {
            throw new RuntimeException('该上游数据库尚未启用。');
        }

        $rows = $this->queries->forSource($source)->select(
            "SELECT
                c.TABLE_SCHEMA AS schema_name,
                c.TABLE_NAME AS object_name,
                CASE WHEN t.TABLE_TYPE = 'VIEW' THEN 'view' ELSE 'table' END AS object_type,
                c.COLUMN_NAME AS column_name,
                c.ORDINAL_POSITION AS ordinal_position,
                c.DATA_TYPE AS data_type,
                c.CHARACTER_MAXIMUM_LENGTH AS max_length,
                c.NUMERIC_PRECISION AS numeric_precision,
                c.NUMERIC_SCALE AS numeric_scale,
                c.IS_NULLABLE AS is_nullable
            FROM INFORMATION_SCHEMA.COLUMNS c
            INNER JOIN INFORMATION_SCHEMA.TABLES t
                ON t.TABLE_SCHEMA = c.TABLE_SCHEMA
                AND t.TABLE_NAME = c.TABLE_NAME
            WHERE c.TABLE_SCHEMA NOT IN ('sys', 'INFORMATION_SCHEMA')
            ORDER BY c.TABLE_SCHEMA, c.TABLE_NAME, c.ORDINAL_POSITION"
        );

        if ($rows === []) {
            throw new RuntimeException('上游数据库没有返回可用的表字段目录。');
        }

        $now = now();
        $normalized = $this->normalizeAndDeduplicate($definition['key'], $rows, $now);
        $objectCount = collect($normalized)
            ->map(fn (array $row): string => $row['schema_name'] . '.' . $row['object_name'])
            ->unique()
            ->count();

        DB::transaction(function () use ($definition, $normalized): void {
            DB::table('upstream_schema_columns')
                ->where('source_key', $definition['key'])
                ->delete();

            foreach (array_chunk($normalized, 500) as $chunk) {
                DB::table('upstream_schema_columns')->insert($chunk);
            }
        });

        return [
            'source' => $definition['key'],
            'objects' => $objectCount,
            'columns' => count($normalized),
            'synchronized_at' => $now,
        ];
    }

    public function ready(): bool
    {
        return Schema::hasTable('upstream_schema_columns');
    }

    private function normalize(string $source, array $row, $now): array
    {
        $row = array_change_key_case($row, CASE_LOWER);
        $schema = trim((string) ($row['schema_name'] ?? ''));
        $object = trim((string) ($row['object_name'] ?? ''));
        $column = trim((string) ($row['column_name'] ?? ''));
        $type = strtolower(trim((string) ($row['object_type'] ?? 'table')));
        $dataType = trim((string) ($row['data_type'] ?? ''));

        if ($schema === '' || $object === '' || $column === '' || $dataType === '') {
            throw new RuntimeException('上游数据库返回了不完整的字段目录。');
        }

        return [
            'source_key' => $source,
            'schema_name' => $schema,
            'object_name' => $object,
            'object_type' => $type === 'view' ? 'view' : 'table',
            'column_name' => $column,
            'ordinal_position' => max((int) ($row['ordinal_position'] ?? 0), 1),
            'data_type' => $dataType,
            'max_length' => $this->nullableInteger($row['max_length'] ?? null),
            'numeric_precision' => $this->nullableInteger($row['numeric_precision'] ?? null),
            'numeric_scale' => $this->nullableInteger($row['numeric_scale'] ?? null),
            'is_nullable' => strtoupper(trim((string) ($row['is_nullable'] ?? 'YES'))) === 'YES',
            'synchronized_at' => $now,
            'created_at' => $now,
            'updated_at' => $now,
        ];
    }

    /**
     * 部分旧 SQL Server 库的 INFORMATION_SCHEMA 会返回重复字段。
     * 共享库的目录键不允许重复，因此写入前按数据库常见的不区分大小写规则去重，
     * 避免少量脏元数据让整个来源的目录同步回滚。
     */
    private function normalizeAndDeduplicate(string $source, array $rows, $now): array
    {
        $unique = [];

        foreach ($rows as $row) {
            $normalized = $this->normalize($source, $row, $now);
            $identity = implode("\0", [
                mb_strtolower($normalized['source_key'], 'UTF-8'),
                mb_strtolower($normalized['schema_name'], 'UTF-8'),
                mb_strtolower($normalized['object_name'], 'UTF-8'),
                mb_strtolower($normalized['column_name'], 'UTF-8'),
            ]);

            $unique[$identity] ??= $normalized;
        }

        return array_values($unique);
    }

    private function nullableInteger(mixed $value): ?int
    {
        return $value === null || $value === '' ? null : (int) $value;
    }
}