CaiWuXiTong/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php

<?php

namespace Modules\EcommerceFinance\Services;

use InvalidArgumentException;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;

class BusinessMirrorDatasetRegistry
{
    /** @return array<string,BusinessMirrorDataset> */
    public function all(): array
    {
        $datasets = [];

        foreach ($this->definitions() as $key => $definition) {
            $datasets[(string) $key] = $this->make((string) $key, (array) $definition);
        }

        return $datasets;
    }

    public function get(string $key): BusinessMirrorDataset
    {
        $definitions = $this->definitions();
        $definition = (array) ($definitions[$key] ?? []);

        if ($definition === []) {
            throw new InvalidArgumentException('未知的业务镜像数据集：' . $key);
        }

        return $this->make($key, $definition);
    }

    private function definitions(): array
    {
        $definitions = (array) config('upstream_mirrors.datasets', []);
        if (Schema::hasTable('business_mirror_datasets')) {
            foreach (DB::table('business_mirror_datasets')->get() as $row) {
                // Existing code-owned datasets cannot be redirected by a new registration.
                $definitions[$row->dataset_key] ??= json_decode($row->definition, true, 32, JSON_THROW_ON_ERROR);
            }
        }
        return $definitions;
    }

    public function automaticInitialSyncAllowed(string $key): bool
    {
        return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets')
            ->where('dataset_key', $key)->where('sync_enabled', true)->where('initial_sync_approved', true)->exists();
    }

    public function scheduled(string $key): bool
    {
        $definition = $this->definitions()[$key] ?? null;
        if (! $definition || ! app(UpstreamSourceRegistry::class)->enabled($definition['source_key'])) return false;
        if (! app(BusinessDataPlacement::class)->shouldSchedule($key)) return false;
        if (array_key_exists($key, (array) config('upstream_mirrors.datasets', []))) return true;
        return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets')
            ->where('dataset_key', $key)->where('sync_enabled', true)->exists();
    }

    public function validate(string $key, array $definition): BusinessMirrorDataset
    {
        return $this->make($key, $definition);
    }

    public function columnSelection(string $key): ?array
    {
        return $this->definitions()[$key]['columns'] ?? null;
    }

    /** 读取登记数据集里的自定义键（例如自动跟随上游补进来的字段列表）。 */
    public function definitionValue(string $key, string $name, mixed $default = null): mixed
    {
        return $this->definitions()[$key][$name] ?? $default;
    }

    /** @return array<string,scalar|list<scalar>> */
    public function fixedFilters(string $key): array
    {
        $definitions = $this->definitions();
        $definition = (array) ($definitions[$key] ?? []);

        if ($definition === []) {
            throw new InvalidArgumentException('未知的业务镜像数据集：' . $key);
        }

        return $this->normalizeFixedFilters($key, (array) ($definition['fixed_filters'] ?? []));
    }

    private function make(string $key, array $definition): BusinessMirrorDataset
    {
        $source = trim((string) ($definition['source_key'] ?? ''));
        $schema = trim((string) ($definition['schema'] ?? ''));
        $object = trim((string) ($definition['object'] ?? ''));
        $target = trim((string) ($definition['target'] ?? ''));
        $keys = array_values(array_filter(array_map('strval', (array) ($definition['keys'] ?? []))));
        $watermark = trim((string) ($definition['watermark'] ?? ''));
        $indexes = array_values(array_map(
            static fn (mixed $columns): array => array_values(array_filter(array_map('strval', (array) $columns))),
            (array) ($definition['indexes'] ?? []),
        ));
        $indexColumns = array_merge(...($indexes ?: [[]]));
        $fixedFilters = $this->normalizeFixedFilters($key, (array) ($definition['fixed_filters'] ?? []));

        foreach ([$source, $target] as $identifier) {
            if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) {
                throw new InvalidArgumentException('业务镜像数据集包含不安全的标识符：' . $key);
            }
        }

        foreach ([$schema, $object, ...$keys, ...$indexColumns] as $identifier) {
            if ($identifier === '' || mb_strlen($identifier) > 128 || str_contains($identifier, "\0")) {
                throw new InvalidArgumentException('业务镜像数据集包含无效的标识符：' . $key);
            }
        }

        if ($keys === [] || in_array([], $indexes, true) || strlen($target) > 64) {
            throw new InvalidArgumentException('业务镜像数据集缺少稳定键或目标表名过长：' . $key);
        }

        if ($watermark !== '' && (mb_strlen($watermark) > 128 || str_contains($watermark, "\0"))) {
            throw new InvalidArgumentException('业务镜像数据集的增量字段不安全：' . $key);
        }

        return new readonly class($key, $source, $schema, $object, $target, $keys, $watermark ?: null, $fixedFilters, $indexes) implements BusinessMirrorDataset {
            public function __construct(
                private string $key,
                private string $source,
                private string $schema,
                private string $object,
                private string $target,
                private array $keys,
                private ?string $watermark,
                private array $fixedFilters,
                private array $indexes,
            ) {
            }

            public function key(): string { return $this->key; }
            public function sourceKey(): string { return $this->source; }
            public function sourceSchema(): string { return $this->schema; }
            public function sourceObject(): string { return $this->object; }
            public function targetTable(): string { return $this->target; }
            public function keyColumns(): array { return $this->keys; }
            public function watermarkColumn(): ?string { return $this->watermark; }
            public function fixedFilters(): array { return $this->fixedFilters; }
            public function indexDefinitions(): array { return $this->indexes; }
        };
    }

    /** @return array<string,scalar|list<scalar>> */
    private function normalizeFixedFilters(string $key, array $filters): array
    {
        $normalized = [];
        $seen = [];

        foreach ($filters as $column => $value) {
            $column = is_string($column) ? trim($column) : '';
            $canonical = strtolower($column);
            if ($column === '' || mb_strlen($column) > 128 || str_contains($column, "\0") || isset($seen[$canonical])) {
                throw new InvalidArgumentException('业务镜像数据集包含无效的固定过滤字段：' . $key);
            }
            if (is_array($value)) {
                if (! array_is_list($value) || $value === [] || collect($value)->contains(
                    static fn (mixed $item): bool => ! is_scalar($item)
                )) {
                    throw new InvalidArgumentException('业务镜像数据集包含无效的固定过滤值：' . $key);
                }
                $value = array_values($value);
            } elseif (! is_scalar($value)) {
                throw new InvalidArgumentException('业务镜像数据集包含无效的固定过滤值：' . $key);
            }

            $seen[$canonical] = true;
            $normalized[$column] = $value;
        }

        return $normalized;
    }
}