CaiWuXiTong/app/Console/Commands/FollowUpstreamColumns.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
use Throwable;

/**
 * 让登记过的数据集自动跟随上游新增字段。
 *
 * 上游视图/表新增列时，数据字典（upstream_schema_columns）会先刷新出来，这里再把新列
 * 补进数据集定义、建好镜像列；下一次常规同步就会把数据带进来，页面也会自动多出这一列。
 * 只做"新增"：不删除、不改写已有字段，敏感字段名一律跳过。
 */
class FollowUpstreamColumns extends Command
{
    protected $signature = 'business-mirror:follow-upstream-columns
        {--dataset= : 只处理指定数据集}';

    protected $description = '把上游新增字段补进登记过的镜像数据集（页面字段自动跟随）';

    private const DENIED_KEYWORDS = ['password', 'passwd', 'secret', 'token', 'cipher', 'credential', 'private'];

    public function handle(
        BusinessMirrorDatasetRegistry $datasets,
        BusinessMirrorSchemaManager $schemas,
    ): int {
        $keys = array_values(array_filter(array_map(
            static fn (mixed $key): string => trim((string) $key),
            (array) ($this->option('dataset') ?: array_values((array) config('upstream_mirrors.auto_follow_columns', []))),
        ), static fn (string $key): bool => $key !== ''));

        if ($keys === []) {
            $this->info('没有登记需要自动跟随上游字段的数据集。');

            return self::SUCCESS;
        }

        // 先刷新数据字典，否则新字段根本不在目录里。该命令自带 24 小时新鲜度判断，
        // 已经同步过会直接跳过。
        foreach (array_unique($this->sourcesOf($keys)) as $source) {
            try {
                // --due：数据字典新鲜（24 小时内同步过）就不再重复扫上游。
                Artisan::call('upstream:schema:sync', ['source' => $source, '--due' => true]);
            } catch (Throwable $exception) {
                report($exception);
                $this->warn($source .'：数据字典刷新失败，本轮跳过新字段检测。');
            }
        }

        $followed = 0;
        foreach ($keys as $key) {
            try {
                $added = $this->follow($key, $datasets, $schemas);
                if ($added !== []) {
                    $followed += count($added);
                    $this->info($key . '：新增 ' . implode('、', $added) . ' 并已建列，下次同步即带回数据。');
                    Log::info('业务镜像自动跟随上游新增字段。', ['dataset' => $key, 'columns' => $added]);
                }
            } catch (Throwable $exception) {
                report($exception);
                $this->error($key . '：跟随失败 —— ' . $exception->getMessage());
            }
        }

        $this->info($followed > 0 ? sprintf('共补入 %d 个新字段。', $followed) : '没有发现上游新增字段。');

        return self::SUCCESS;
    }

    /** @param list<string> $keys @return list<string> */
    private function sourcesOf(array $keys): array
    {
        $sources = [];
        foreach (DB::table('business_mirror_datasets')->whereIn('dataset_key', $keys)->get() as $row) {
            $definition = json_decode((string) $row->definition, true) ?: [];
            $source = trim((string) ($definition['source_key'] ?? ''));
            if ($source !== '') {
                $sources[] = $source;
            }
        }

        return $sources;
    }

    /**
     * @return list<string> 本次补入的字段
     */
    private function follow(string $key, BusinessMirrorDatasetRegistry $datasets, BusinessMirrorSchemaManager $schemas): array
    {
        $row = DB::table('business_mirror_datasets')->where('dataset_key', $key)->first();
        if ($row === null) {
            return [];
        }

        $definition = json_decode((string) $row->definition, true) ?: [];
        $source = trim((string) ($definition['source_key'] ?? ''));
        $schema = trim((string) ($definition['schema'] ?? ''));
        $object = trim((string) ($definition['object'] ?? ''));
        $current = array_values(array_filter(array_map(
            static fn (mixed $column): string => trim((string) $column),
            (array) ($definition['columns'] ?? []),
        ), static fn (string $column): bool => $column !== ''));

        if ($source === '' || $schema === '' || $object === '' || $current === []) {
            return [];
        }

        $upstream = DB::table('upstream_schema_columns')
            ->where('source_key', $source)
            ->whereRaw('LOWER(schema_name) = ?', [strtolower($schema)])
            ->whereRaw('LOWER(object_name) = ?', [strtolower($object)])
            ->orderBy('ordinal_position')
            ->pluck('column_name')
            ->map(static fn (mixed $name): string => (string) $name)
            ->all();

        $added = [];
        foreach ($upstream as $column) {
            if (in_array($column, $current, true) || str_starts_with($column, '_mirror') || $this->isSensitive($column)) {
                continue;
            }
            $added[] = $column;
        }

        if ($added === []) {
            return [];
        }

        $definition['columns'] = array_values(array_merge($current, $added));
        // 记下"跟随进来的字段"：页面据此自动补列，不会把历史上刻意没展示的字段带出来。
        $definition['followed_columns'] = array_values(array_unique(array_merge(
            (array) ($definition['followed_columns'] ?? []),
            $added,
        )));

        DB::table('business_mirror_datasets')->where('dataset_key', $key)->update([
            'definition' => json_encode($definition, JSON_THROW_ON_ERROR),
            'updated_at' => now(),
        ]);

        // 立刻建列，下一次同步就能写入；页面在数据到达前显示为空即可。
        $schemas->ensure($datasets->get($key));

        return $added;
    }

    private function isSensitive(string $column): bool
    {
        $name = strtolower($column);
        foreach (self::DENIED_KEYWORDS as $keyword) {
            if (str_contains($name, $keyword)) {
                return true;
            }
        }

        return false;
    }
}