230 lines
11 KiB
PHP
230 lines
11 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\Log;
|
||
use InvalidArgumentException;
|
||
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
|
||
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
||
use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer;
|
||
use Modules\EcommerceFinance\Services\DingProcessDetailMirrorSynchronizer;
|
||
use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer;
|
||
use Throwable;
|
||
|
||
class SyncBusinessMirrors extends Command
|
||
{
|
||
protected $signature = 'business-mirror:sync-tables
|
||
{dataset? : 只同步指定数据集}
|
||
{--full : 执行或继续首次全量/校准同步}
|
||
{--batch= : 每批最多读取行数}
|
||
{--max-batches= : 每个数据集本轮最多批次数，0 表示直到完成}
|
||
{--max-datasets= : 本轮最多处理的数据集数量}';
|
||
|
||
protected $description = '将项目实际使用的上游表按稳定游标同步到共享业务镜像库';
|
||
|
||
private bool $hasSchedulingErrors = false;
|
||
|
||
public function handle(
|
||
BusinessMirrorDatasetRegistry $datasets,
|
||
BusinessMirrorSynchronizer $synchronizer,
|
||
GenericBusinessMirrorRepository $mirror,
|
||
?DingPayableWorkflowMirrorSynchronizer $dingPayableWorkflows = null,
|
||
?DingProcessDetailMirrorSynchronizer $dingProcessDetails = null,
|
||
): int
|
||
{
|
||
if (! (bool) config('upstream_mirrors.refresh_enabled', false) && ! $this->option('full')) {
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
$this->hasSchedulingErrors = false;
|
||
$requested = $this->argument('dataset');
|
||
$keys = $requested === null
|
||
? $this->scheduledKeys($datasets, $mirror)
|
||
: [['key' => (string) $requested, 'full' => (bool) $this->option('full'), 'catchup' => false]];
|
||
$maxDatasets = max((int) ($this->option('max-datasets')
|
||
?? config('upstream_pipeline.sync.max_datasets', 8)), 1);
|
||
$processed = 0;
|
||
$failed = $this->hasSchedulingErrors;
|
||
$deadline = microtime(true) + max((int) config('upstream_pipeline.sync.cycle_seconds', 30), 1);
|
||
|
||
foreach ($keys as $task) {
|
||
if ($processed >= $maxDatasets || ($processed > 0 && microtime(true) >= $deadline)) {
|
||
break;
|
||
}
|
||
$key = $task['key'];
|
||
|
||
try {
|
||
$result = $synchronizer->sync(
|
||
$key,
|
||
(bool) $task['full'],
|
||
$this->option('batch') === null ? null : (int) $this->option('batch'),
|
||
$this->option('max-batches') === null
|
||
? ($task['catchup'] ? max((int) config('upstream_pipeline.sync.catchup_batches', 6), 1) : null)
|
||
: (int) $this->option('max-batches'),
|
||
);
|
||
} catch (Throwable $exception) {
|
||
$failed = true;
|
||
Log::error('共享业务镜像表同步失败。', ['dataset' => $key, 'exception' => $exception::class]);
|
||
$this->error($key . '：同步失败，已保留上一版共享数据。');
|
||
$processed++;
|
||
continue;
|
||
}
|
||
|
||
if ($result['status'] === 'requires_full_sync') {
|
||
if ($requested !== null) {
|
||
$this->warn($key . '：尚未完成首次全量同步，请显式使用 --full。');
|
||
}
|
||
continue;
|
||
}
|
||
|
||
$processed++;
|
||
$this->info(sprintf(
|
||
'%s：%s，本轮 %d 批，累计处理 %d 行。',
|
||
$key,
|
||
$result['status'] === 'succeeded' ? '同步完成' : '同步进行中',
|
||
$result['batches'],
|
||
$result['rows'],
|
||
));
|
||
}
|
||
|
||
// 钉钉应付流程采用按流程聚合的专用镜像；接入同一入口后由现有
|
||
// 生产 cron 自动断点续传，避免额外维护一套调度脚本。
|
||
if ($requested === null && $dingPayableWorkflows !== null && ! app()->environment('testing')) {
|
||
try {
|
||
$result = $dingPayableWorkflows->sync(
|
||
(bool) $this->option('full'),
|
||
$this->option('batch') === null ? null : (int) $this->option('batch'),
|
||
$this->option('max-batches') === null ? null : (int) $this->option('max-batches'),
|
||
);
|
||
if ($result['status'] !== 'requires_full_sync') {
|
||
$this->info(sprintf(
|
||
'operations_replica.rpt.ding_payable_workflows：%s，本轮 %d 批，累计处理 %d 个流程。',
|
||
$result['status'] === 'succeeded' ? '同步完成' : '同步进行中',
|
||
$result['batches'],
|
||
$result['rows'],
|
||
));
|
||
}
|
||
} catch (Throwable $exception) {
|
||
$failed = true;
|
||
Log::error('钉钉应付流程镜像同步失败。', ['exception' => $exception::class]);
|
||
$this->error('operations_replica.rpt.ding_payable_workflows：同步失败，已保留上一版共享数据。');
|
||
}
|
||
|
||
// 线上代发业务报销流程复用同一套钉钉流程镜像同步链路，接入此处后
|
||
// 由现有生产 cron 自动断点续传，不额外维护调度脚本。
|
||
try {
|
||
$daifaResult = $dingPayableWorkflows
|
||
->forConfigKey('ding_online_daifa_price')
|
||
->sync(
|
||
(bool) $this->option('full'),
|
||
$this->option('batch') === null ? null : (int) $this->option('batch'),
|
||
$this->option('max-batches') === null ? null : (int) $this->option('max-batches'),
|
||
);
|
||
if ($daifaResult['status'] !== 'requires_full_sync') {
|
||
$this->info(sprintf(
|
||
'operations_replica.rpt.ding_online_daifa_price：%s，本轮 %d 批，累计处理 %d 条明细。',
|
||
$daifaResult['status'] === 'succeeded' ? '同步完成' : '同步进行中',
|
||
$daifaResult['batches'],
|
||
$daifaResult['rows'],
|
||
));
|
||
}
|
||
} catch (Throwable $exception) {
|
||
$failed = true;
|
||
Log::error('钉钉线上代发价格流程镜像同步失败。', ['exception' => $exception::class]);
|
||
$this->error('operations_replica.rpt.ding_online_daifa_price：同步失败，已保留上一版共享数据。');
|
||
}
|
||
|
||
// 财务流程明细按结构化明细/付款事实同步。首次全量通过专用命令执行，
|
||
// 日常只处理最近更新过的审批，避免每分钟重扫整批钉钉表单。
|
||
if ($dingProcessDetails !== null) {
|
||
try {
|
||
$detailResult = $dingProcessDetails->sync(false);
|
||
if ($detailResult['status'] === 'succeeded') {
|
||
$this->info(sprintf(
|
||
'operations_replica.rpt.ding_process_details：同步完成，本轮 %d 个审批。',
|
||
$detailResult['workflows'],
|
||
));
|
||
}
|
||
} catch (Throwable $exception) {
|
||
$failed = true;
|
||
Log::error('钉钉流程明细结构镜像同步失败。', ['exception' => $exception::class]);
|
||
$this->error('operations_replica.rpt.ding_process_details：同步失败，已保留上一版共享数据。');
|
||
}
|
||
}
|
||
}
|
||
|
||
return $failed ? self::FAILURE : self::SUCCESS;
|
||
}
|
||
|
||
/** @return list<array{key:string,full:bool,catchup:bool,last_attempted_at:string}> */
|
||
private function scheduledKeys(
|
||
BusinessMirrorDatasetRegistry $datasets,
|
||
GenericBusinessMirrorRepository $mirror,
|
||
): array {
|
||
$scheduled = [];
|
||
$dedicated = array_fill_keys(array_map(
|
||
'strval',
|
||
(array) config('upstream_mirrors.table_sync.dedicated_datasets', []),
|
||
), true);
|
||
|
||
foreach ($datasets->all() as $key => $dataset) {
|
||
if (isset($dedicated[$key])) {
|
||
continue;
|
||
}
|
||
|
||
try {
|
||
if (! $datasets->scheduled($key)) {
|
||
continue;
|
||
}
|
||
} catch (InvalidArgumentException $exception) {
|
||
// A stale source registration must not stop unrelated mirrors.
|
||
$this->hasSchedulingErrors = true;
|
||
Log::warning('共享业务镜像调度配置无效，已跳过该数据集。', [
|
||
'dataset' => $key,
|
||
'exception' => $exception::class,
|
||
]);
|
||
$this->warn($key . '：调度配置无效，已跳过，其他数据集继续同步。');
|
||
continue;
|
||
}
|
||
|
||
$state = $mirror->state($dataset);
|
||
$initial = ! is_array($state) || ($state['initial_completed_at'] ?? null) === null;
|
||
if ($initial && ! $datasets->automaticInitialSyncAllowed($key)) continue;
|
||
if (($state['status'] ?? '') === 'failed' && ! empty($state['last_attempted_at'])
|
||
&& now()->diffInSeconds(\Illuminate\Support\Carbon::parse($state['last_attempted_at'])) < 300) continue;
|
||
|
||
// Only explicitly opted-in small datasets receive periodic full reconciliation.
|
||
// Keep an interrupted full run in full mode so its cursor/generation can resume.
|
||
$hours = (int) $datasets->definitionValue($key, 'full_reconcile_hours', 0);
|
||
$lastFull = $state['last_full_succeeded_at'] ?? null;
|
||
$reconcile = $hours > 0 && (empty($lastFull)
|
||
|| \Illuminate\Support\Carbon::parse($lastFull)->addHours($hours)->lte(now()));
|
||
$resumeFull = $hours > 0 && ($state['run_mode'] ?? '') === 'full'
|
||
&& in_array($state['status'] ?? '', ['running', 'failed'], true);
|
||
|
||
$scheduled[] = [
|
||
'key' => $key,
|
||
'full' => $initial || $reconcile || $resumeFull || (bool) $this->option('full'),
|
||
'catchup' => $initial || ($state['status'] ?? '') === 'running',
|
||
'last_attempted_at' => (string) ($state['last_attempted_at'] ?? ''),
|
||
];
|
||
}
|
||
|
||
usort($scheduled, fn (array $left, array $right): int =>
|
||
[$left['last_attempted_at'], $left['key']] <=> [$right['last_attempted_at'], $right['key']]
|
||
);
|
||
|
||
// Alternate unfinished windows and routine refreshes: catch-up progresses
|
||
// every cycle without starving small tables or monopolizing a source.
|
||
$catchup = array_values(array_filter($scheduled, fn ($task) => $task['catchup']));
|
||
$routine = array_values(array_filter($scheduled, fn ($task) => ! $task['catchup']));
|
||
$ordered = [];
|
||
while ($catchup !== [] || $routine !== []) {
|
||
if ($routine !== []) $ordered[] = array_shift($routine);
|
||
if ($catchup !== []) $ordered[] = array_shift($catchup);
|
||
}
|
||
return $ordered;
|
||
}
|
||
}
|