error('尚未执行镜像数据集登记表迁移。'); return self::FAILURE; } $source = (string) $this->argument('source'); $schema = (string) $this->argument('schema'); $object = (string) $this->argument('object'); $key = $source . '.' . $schema . '.' . $object; try { $sources->definition($source); if (strlen($key) > 191 || isset(config('upstream_mirrors.datasets', [])[$key])) { throw new \RuntimeException('数据集键过长或由现有代码维护,不能覆盖。'); } if ($this->option('disable-sync')) { if ($this->option('enable-sync')) throw new \RuntimeException('不能同时启用和停用。'); if (DB::table('business_mirror_datasets')->where('dataset_key', $key)->update(['sync_enabled' => false, 'updated_at' => now()]) !== 1) { throw new \RuntimeException('未找到该登记数据集。'); } $this->info('已停止自动同步,现有镜像保留。'); return self::SUCCESS; } $keys = $this->csv((string) $this->option('keys')); $columns = $this->csv((string) $this->option('columns')); $watermark = trim((string) $this->option('watermark')); if ($keys === [] || $columns === [] || array_diff([...$keys, ...($watermark === '' ? [] : [$watermark])], $columns)) { throw new \RuntimeException('请明确字段和稳定键;同步字段必须包含稳定键及增量字段。'); } $catalog = DB::table('upstream_schema_columns')->where('source_key', $source) ->where('schema_name', $schema)->where('object_name', $object)->pluck('column_name')->all(); if (array_diff($columns, $catalog)) { throw new \RuntimeException('字段不在数据字典中,请先同步该来源字典并核实字段。'); } $definition = ['source_key' => $source, 'schema' => $schema, 'object' => $object, 'target' => 'mirror_' . substr(hash('sha256', $key), 0, 24), 'keys' => $keys, 'columns' => $columns, 'watermark' => $watermark]; $datasets->validate($key, $definition); $enable = (bool) $this->option('enable-sync'); DB::transaction(function () use ($key, $source, $definition, $enable): void { $existing = DB::table('business_mirror_datasets')->where('dataset_key', $key)->lockForUpdate()->first(); if ($existing !== null && json_decode($existing->definition, true) != $definition) { throw new \RuntimeException('已有数据集定义不同,不能覆盖稳定键、字段或同步游标。请先完成变更评估。'); } if ($existing !== null) { if ($enable) DB::table('business_mirror_datasets')->where('dataset_key', $key)->update([ 'sync_enabled' => true, 'initial_sync_approved' => true, 'updated_at' => now(), ]); return; } DB::table('business_mirror_datasets')->insert([ 'dataset_key' => $key, 'source_key' => $source, 'definition' => json_encode($definition, JSON_THROW_ON_ERROR), 'sync_enabled' => $enable, 'initial_sync_approved' => $enable, 'created_at' => now(), 'updated_at' => now(), ]); }); $this->info($key . ($enable ? ':已启用,生产调度将自动执行首次同步并持续增量。' : ':已登记,自动同步未启用。')); return self::SUCCESS; } catch (\Illuminate\Database\QueryException) { $this->error('登记失败,请检查系统数据库状态;原数据集保留。'); } catch (\RuntimeException|\InvalidArgumentException $e) { $this->error($e->getMessage()); } catch (Throwable) { $this->error('登记失败,未启用新同步。请检查系统数据库状态。'); } return self::FAILURE; } private function csv(string $value): array { return array_values(array_unique(array_filter(array_map('trim', explode(',', $value))))); } }