95 lines
5.3 KiB
PHP
95 lines
5.3 KiB
PHP
<?php
|
|
|
|
namespace App\Console\Commands;
|
|
|
|
use Illuminate\Console\Command;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
|
use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
|
|
use Throwable;
|
|
|
|
class RegisterBusinessMirror extends Command
|
|
{
|
|
protected $signature = 'business-mirror:register {source} {schema} {object}
|
|
{--keys= : 逗号分隔的稳定唯一键}
|
|
{--columns= : 逗号分隔的明确同步字段，必须包含稳定键及增量字段}
|
|
{--watermark= : 可靠更新时间字段；留空为仅按主键追加}
|
|
{--enable-sync : 启用可断点首次同步及后续增量}
|
|
{--disable-sync : 停用此数据集的自动同步，保留镜像数据}';
|
|
|
|
protected $description = '登记声明式镜像数据集；核对范围并备份后可直接启用同步';
|
|
|
|
public function handle(BusinessMirrorDatasetRegistry $datasets, UpstreamSourceRegistry $sources): int
|
|
{
|
|
if (! Schema::hasTable('business_mirror_datasets')) {
|
|
$this->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)))));
|
|
}
|
|
}
|