278 lines
13 KiB
PHP
278 lines
13 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services;
|
||
|
||
use Illuminate\Database\ConnectionInterface;
|
||
use Illuminate\Database\Schema\Blueprint;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use Illuminate\Support\Str;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 将钉钉应付申请流程的固定流程编码同步为按流程聚合的共享镜像。
|
||
*
|
||
* 钉钉表单字段存放在 JSON 中，且历史版本字段不完全一致；镜像保留原始
|
||
* 主表、表单和审批任务 payload，页面只在镜像库中解析和筛选。
|
||
*/
|
||
class DingPayableWorkflowMirrorSynchronizer
|
||
{
|
||
private readonly ConnectionInterface $connection;
|
||
|
||
public function __construct(private readonly UpstreamReadOnlyQueryFactory $queries)
|
||
{
|
||
$this->connection = app('db')->connection(
|
||
(string) config('upstream_mirrors.connection', 'business_mirror')
|
||
);
|
||
}
|
||
|
||
/** @return array{status:string,mode:string,batches:int,rows:int,dataset:string} */
|
||
public function sync(bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array
|
||
{
|
||
$config = (array) config('upstream_mirrors.ding_payable_workflows', []);
|
||
$dataset = (string) ($config['dataset_key'] ?? 'operations_replica.rpt.ding_payable_workflows');
|
||
$target = (string) ($config['target_table'] ?? 'ding_payable_workflow_mirror');
|
||
$source = (string) ($config['source_key'] ?? 'operations_replica');
|
||
$processCode = (string) ($config['process_code'] ?? '');
|
||
|
||
if ($processCode === '') {
|
||
throw new RuntimeException('ding_payable_workflow_mirror_not_initialized');
|
||
}
|
||
|
||
$this->ensureMirrorTable($target);
|
||
|
||
UpstreamOdbcConnection::reset($source);
|
||
$state = $this->state($dataset);
|
||
|
||
if (! $full && ($state['initial_completed_at'] ?? null) === null) {
|
||
return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'batches' => 0, 'rows' => 0, 'dataset' => $dataset];
|
||
}
|
||
|
||
$mode = $full ? 'full' : 'incremental';
|
||
$resume = is_array($state)
|
||
&& ($state['run_mode'] ?? null) === $mode
|
||
&& in_array($state['status'] ?? null, ['running', 'failed'], true)
|
||
&& is_string($state['generation'] ?? null)
|
||
&& is_array($state['cursor'] ?? null);
|
||
$generation = $resume ? (string) $state['generation'] : (string) Str::uuid();
|
||
$cursor = $resume
|
||
? (string) ($state['cursor']['processInstanceId'] ?? '')
|
||
: (string) ($this->decodeJson($state['high_watermark'] ?? null)['processInstanceId'] ?? '');
|
||
$rows = $resume ? (int) ($state['last_upserted_rows'] ?? 0) : 0;
|
||
$batchSize = min(max($batchSize ?? (int) ($config['batch_size'] ?? 100), 1), 500);
|
||
$maxBatches = max($maxBatches ?? (int) ($config['max_batches_per_run'] ?? 2), 0);
|
||
|
||
$this->start($dataset, $target, $source, $mode, $generation, $cursor, $resume);
|
||
$batches = 0;
|
||
|
||
try {
|
||
while (true) {
|
||
$workflows = $this->readWorkflows($source, $processCode, $cursor, $batchSize);
|
||
if ($workflows === []) {
|
||
$this->complete($dataset, $target, $mode, $generation, $rows, $cursor);
|
||
|
||
return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
|
||
}
|
||
|
||
$ids = array_values(array_filter(array_map(
|
||
static fn (array $row): string => (string) ($row['processInstanceId'] ?? ''),
|
||
$workflows,
|
||
)));
|
||
$schemes = $this->readRelated($source, 'ding_WorkflowScheme', ['id', 'processInstanceId', 'componentType', 'name', 'value', 'extValue'], $ids, 'id');
|
||
$tasks = $this->readRelated($source, 'ding_WorkflowTask', ['processInstanceId', 'result', 'activityId', 'finishTime', 'createTime', 'userId', 'taskId', 'status', 'id'], $ids, 'id');
|
||
|
||
$schemeById = $this->groupBy($schemes, 'processInstanceId');
|
||
$taskById = $this->groupBy($tasks, 'processInstanceId');
|
||
$now = now();
|
||
$payloads = [];
|
||
|
||
foreach ($workflows as $workflow) {
|
||
$id = (string) ($workflow['processInstanceId'] ?? '');
|
||
if ($id === '') {
|
||
continue;
|
||
}
|
||
$payloads[] = [
|
||
'process_instance_id' => $id,
|
||
'workflow_payload' => json_encode($workflow, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
|
||
'scheme_payload' => json_encode($schemeById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
|
||
'task_payload' => json_encode($taskById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
|
||
'workflow_created_at' => isset($workflow['createAt']) ? (string) $workflow['createAt'] : null,
|
||
'sync_generation' => $generation,
|
||
'synced_at' => $now,
|
||
];
|
||
}
|
||
|
||
$cursor = (string) ($workflows[array_key_last($workflows)]['processInstanceId'] ?? $cursor);
|
||
$rows += count($payloads);
|
||
$batches++;
|
||
$this->connection->transaction(function () use ($dataset, $target, $payloads, $cursor, $generation, $rows, $now): void {
|
||
if ($payloads !== []) {
|
||
$this->connection->table($target)->upsert(
|
||
$payloads,
|
||
['process_instance_id'],
|
||
['workflow_payload', 'scheme_payload', 'task_payload', 'workflow_created_at', 'sync_generation', 'synced_at'],
|
||
);
|
||
}
|
||
$this->connection->table('business_mirror_table_states')
|
||
->where('dataset_key', $dataset)
|
||
->update([
|
||
'cursor' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE),
|
||
'last_upserted_rows' => $rows,
|
||
'updated_at' => $now,
|
||
]);
|
||
});
|
||
|
||
if (count($workflows) < $batchSize) {
|
||
$this->complete($dataset, $target, $mode, $generation, $rows, $cursor);
|
||
|
||
return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
|
||
}
|
||
|
||
if ($maxBatches > 0 && $batches >= $maxBatches) {
|
||
return ['status' => 'running', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
|
||
}
|
||
}
|
||
} catch (Throwable $exception) {
|
||
$this->connection->table('business_mirror_table_states')
|
||
->where('dataset_key', $dataset)
|
||
->update(['status' => 'failed', 'last_error_code' => mb_substr($exception::class, 0, 120), 'updated_at' => now()]);
|
||
throw $exception;
|
||
}
|
||
}
|
||
|
||
private function ensureMirrorTable(string $target): void
|
||
{
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if ($schema->hasTable($target)) {
|
||
return;
|
||
}
|
||
$schema->create($target, function (Blueprint $table): void {
|
||
$table->string('process_instance_id', 100)->primary();
|
||
$table->json('workflow_payload');
|
||
$table->json('scheme_payload');
|
||
$table->json('task_payload')->nullable();
|
||
$table->string('workflow_created_at', 32)->nullable()->index();
|
||
$table->uuid('sync_generation')->nullable()->index();
|
||
$table->dateTime('synced_at', 6);
|
||
});
|
||
}
|
||
|
||
/** @return array<string,mixed>|null */
|
||
private function state(string $dataset): ?array
|
||
{
|
||
$row = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->first();
|
||
if ($row === null) {
|
||
return null;
|
||
}
|
||
$state = (array) $row;
|
||
$state['cursor'] = is_string($state['cursor'] ?? null) ? (json_decode($state['cursor'], true) ?: []) : ($state['cursor'] ?? []);
|
||
|
||
return $state;
|
||
}
|
||
|
||
private function start(string $dataset, string $target, string $source, string $mode, string $generation, string $cursor, bool $resume): void
|
||
{
|
||
$now = now();
|
||
$values = [
|
||
'source_key' => $source,
|
||
'source_object' => 'rpt.ding_Workflow + rpt.ding_WorkflowScheme + rpt.ding_WorkflowTask',
|
||
'target_table' => $target,
|
||
'status' => 'running',
|
||
'run_mode' => $mode,
|
||
'generation' => $generation,
|
||
'last_attempted_at' => $now,
|
||
'last_error_code' => null,
|
||
'updated_at' => $now,
|
||
];
|
||
if (! $resume) {
|
||
$values += ['cursor' => null, 'window_started_at' => null, 'window_ends_at' => null, 'last_upserted_rows' => 0];
|
||
}
|
||
$query = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset);
|
||
$query->exists()
|
||
? $query->update($values)
|
||
: $this->connection->table('business_mirror_table_states')->insert(['dataset_key' => $dataset, 'created_at' => $now] + $values);
|
||
}
|
||
|
||
private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, string $cursor): void
|
||
{
|
||
$now = now();
|
||
$this->connection->transaction(function () use ($dataset, $target, $mode, $generation, $rows, $cursor, $now): void {
|
||
if ($mode === 'full') {
|
||
$this->connection->table($target)->where(function ($query) use ($generation): void {
|
||
$query->whereNull('sync_generation')->orWhere('sync_generation', '!=', $generation);
|
||
})->delete();
|
||
}
|
||
$values = [
|
||
'status' => 'succeeded',
|
||
'cursor' => null,
|
||
'high_watermark' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE),
|
||
'last_succeeded_at' => $now,
|
||
'last_upserted_rows' => $rows,
|
||
'mirror_row_count' => $this->connection->table($target)->count(),
|
||
'last_error_code' => null,
|
||
'updated_at' => $now,
|
||
];
|
||
if ($mode === 'full') {
|
||
$values['initial_completed_at'] = $now;
|
||
$values['last_full_succeeded_at'] = $now;
|
||
}
|
||
$this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->update($values);
|
||
});
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function decodeJson(mixed $value): array
|
||
{
|
||
if (is_array($value)) {
|
||
return $value;
|
||
}
|
||
if (! is_string($value) || $value === '') {
|
||
return [];
|
||
}
|
||
$decoded = json_decode($value, true);
|
||
return is_array($decoded) ? $decoded : [];
|
||
}
|
||
|
||
/** @return list<array<string,mixed>> */
|
||
private function readWorkflows(string $source, string $processCode, string $cursor, int $limit): array
|
||
{
|
||
$where = '[processCode] = ?';
|
||
$bindings = [$processCode];
|
||
if ($cursor !== '') {
|
||
$where .= ' AND [processInstanceId] > ?';
|
||
$bindings[] = $cursor;
|
||
}
|
||
return $this->queries->forSource($source)->select(
|
||
'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId],[joinAt],[updateAt],[createAt],[processCode],[flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY [processInstanceId]',
|
||
$bindings,
|
||
);
|
||
}
|
||
|
||
/** @param list<string> $ids @param list<string> $columns @return list<array<string,mixed>> */
|
||
private function readRelated(string $source, string $table, array $columns, array $ids, string $order): array
|
||
{
|
||
if ($ids === []) {
|
||
return [];
|
||
}
|
||
$placeholders = implode(',', array_fill(0, count($ids), '?'));
|
||
return $this->queries->forSource($source)->select(
|
||
'SELECT ' . implode(', ', array_map(static fn (string $column): string => '[' . $column . ']', $columns))
|
||
. ' FROM [rpt].[' . $table . '] WHERE [processInstanceId] IN (' . $placeholders . ') ORDER BY [' . $order . ']',
|
||
$ids,
|
||
);
|
||
}
|
||
|
||
/** @param list<array<string,mixed>> $rows @return array<string,list<array<string,mixed>>> */
|
||
private function groupBy(array $rows, string $column): array
|
||
{
|
||
$grouped = [];
|
||
foreach ($rows as $row) {
|
||
$key = (string) ($row[$column] ?? '');
|
||
if ($key !== '') {
|
||
$grouped[$key][] = $row;
|
||
}
|
||
}
|
||
return $grouped;
|
||
}
|
||
}
|