90 lines
2.8 KiB
PHP
90 lines
2.8 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Illuminate\Cache\CacheManager;
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Modules\EcommerceFinance\Services\UpstreamSchemaCatalogSynchronizer;
|
||
use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
|
||
use Throwable;
|
||
|
||
class UpstreamSyncSchemaCatalog extends Command
|
||
{
|
||
protected $signature = 'upstream:schema:sync
|
||
{source? : 仅同步指定来源}
|
||
{--due : 仅同步超过 24 小时未更新的来源}';
|
||
|
||
protected $description = '低频同步上游表、视图和字段目录到共享系统数据库';
|
||
|
||
public function handle(
|
||
UpstreamSchemaCatalogSynchronizer $synchronizer,
|
||
UpstreamSourceRegistry $sources,
|
||
CacheManager $cacheManager,
|
||
): int {
|
||
if (! $synchronizer->ready()) {
|
||
$this->warn('共享系统数据库尚未执行数据字典迁移。');
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
$requested = $this->argument('source');
|
||
$sourceKeys = $requested
|
||
? [(string) $requested]
|
||
: array_keys($sources->all());
|
||
$cache = $cacheManager->store((string) config('upstreams.snapshots.cache_store', 'upstream_shared'));
|
||
$failed = false;
|
||
|
||
foreach ($sourceKeys as $source) {
|
||
try {
|
||
$sources->definition($source);
|
||
|
||
if (! $sources->enabled($source) || $this->isFresh($source)) {
|
||
continue;
|
||
}
|
||
|
||
$lock = $cache->lock('upstream:schema-sync:' . $source, 600);
|
||
|
||
if (! $lock->get()) {
|
||
continue;
|
||
}
|
||
|
||
try {
|
||
$result = $synchronizer->sync($source);
|
||
$this->info(sprintf(
|
||
'%s：已同步 %d 个对象、%d 个字段。',
|
||
$source,
|
||
$result['objects'],
|
||
$result['columns']
|
||
));
|
||
} finally {
|
||
$lock->release();
|
||
}
|
||
} catch (Throwable $exception) {
|
||
$failed = true;
|
||
Log::warning('上游数据字典同步失败，已保留原目录。', [
|
||
'source' => $source,
|
||
'exception' => $exception::class,
|
||
]);
|
||
$this->error($source . '：同步失败，已保留原目录。');
|
||
}
|
||
}
|
||
|
||
return $failed ? self::FAILURE : self::SUCCESS;
|
||
}
|
||
|
||
private function isFresh(string $source): bool
|
||
{
|
||
if (! $this->option('due')) {
|
||
return false;
|
||
}
|
||
|
||
$latest = DB::table('upstream_schema_columns')
|
||
->where('source_key', $source)
|
||
->max('synchronized_at');
|
||
|
||
return $latest !== null && now()->diffInHours($latest) < 24;
|
||
}
|
||
}
|