diff --git a/app/Console/Commands/SyncDingProcessDetailMirror.php b/app/Console/Commands/SyncDingProcessDetailMirror.php index 97af40a6..1d8addc0 100644 --- a/app/Console/Commands/SyncDingProcessDetailMirror.php +++ b/app/Console/Commands/SyncDingProcessDetailMirror.php @@ -10,17 +10,33 @@ class SyncDingProcessDetailMirror extends Command { protected $signature = 'business-mirror:sync-ding-process-details {--full : 从配置起始日期完整重建明细和付款事实} + {--business-id= : 只同步指定审批编号，多个用逗号分隔（最多 50 个）} {--dry-run : 只读取上游并校验，不写入镜像}'; - protected $description = '从钉钉结构化审批表单同步流程明细和付款事实'; + protected $description = '从结构化审批镜像同步流程明细，并用钉钉官方接口补取缺失付款'; public function handle(DingProcessDetailMirrorSynchronizer $synchronizer): int { + $businessIds = array_values(array_filter(array_map( + 'trim', + explode(',', (string) $this->option('business-id')), + ))); + try { - $result = $synchronizer->sync( - (bool) $this->option('full'), - (bool) $this->option('dry-run'), - ); + if ($businessIds !== []) { + if ((bool) $this->option('full')) { + throw new \InvalidArgumentException('--full 不能和 --business-id 同时使用。'); + } + $result = $synchronizer->syncBusinessIds( + $businessIds, + (bool) $this->option('dry-run'), + ); + } else { + $result = $synchronizer->sync( + (bool) $this->option('full'), + (bool) $this->option('dry-run'), + ); + } } catch (Throwable $exception) { $this->error('同步失败，已保留上一版共享数据：' . $exception->getMessage()); diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index 595fd496..6b1fbc45 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -37,6 +37,8 @@ 流程明细另设独立定时任务，避免与每分钟科目映射消费争用同一上游目录时被静默跳过。正式发布后由工作站执行 `scripts/install-ding-process-sync-cron.ps1` 安装 `/etc/cron.d/caiwu-ding-process-details`：每 5 分钟尝试运行 `business-mirror:sync-ding-process-details`，使用独立调度锁并在共用上游锁上最多等待 15 分钟。增量同步除跟进审批主表更新时间外，还以 `ding_WorkflowScheme` 的数字 `id` 高水位发现后补表单行，避免付款明细等新组件写入后主审批 `updateAt` 不变而被漏掉；原地改写同一组件仍由每周全量校准兜底。页面按审批单费用合计和独立付款事实展示“付款明细待同步 / 未付款 / 部分付款 / 已付款 / 不适用”：只有源表单明确存在付款表或付款合计字段但金额为零时才标“未付款”，付款表单缺失时必须显示“付款明细待同步”，不能把上游尚未返回的数据误报为未付款。审批状态不能代替付款状态。 +`operations_replica` 缺失付款组件时，同步器复用现有钉钉应用凭据调用 `https://api.dingtalk.com/v1.0/workflow/processInstances` 补取官方最新表单。付款来源标记为 `api`，付款事实仍进入 `ding_process_payment_mirror`；不会用官方接口覆盖副本已有的明细，只补缺失或空的付款字段。自动增量每 5 分钟按 `process_instance_id` 游标轮转最多 50 个缺失流程，避免单个失败记录阻塞队列或瞬时打满接口；人工核查可执行 `business-mirror:sync-ding-process-details --business-id=审批编号1,审批编号2 --dry-run`，确认后再去掉 `--dry-run`。单次最多 50 个审批编号。 + 抖音月中校准可在上述导入命令追加 `--through=YYYY-MM-DD`，文件必须包含该月 1 日至截止日（含当天）的全部店铺和流水。系统仅替换这一日期范围，截止日之后继续读取原始账单镜像；省略参数仍表示完整月份。截止日期不得早于同月已校准范围。每天更新时重新导出月初至昨日的完整结果并核对笔数、收支总额，以覆盖迟到及修改记录，不能把单日增量当作月初至截止日的完整结果导入。 ### 成本价镜像观测 diff --git a/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php b/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php index 05f7a6f2..c8f4553e 100644 --- a/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/DingProcessDetailMirrorRepository.php @@ -307,7 +307,7 @@ class DingProcessDetailMirrorRepository < COALESCE(detail_totals.approval_amount, 0) THEN '部分付款' WHEN COALESCE(payment_totals.payment_amount, 0) > 0.005 THEN '已付款' WHEN {$sourceState} = 'not_applicable' THEN '不适用' - WHEN {$sourceState} = 'present' THEN '未付款' + WHEN {$sourceState} IN ('present', 'api') THEN '未付款' ELSE '付款明细待同步' END"; } diff --git a/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php b/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php index 4591593a..1d66108a 100644 --- a/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/DingProcessDetailMirrorSynchronizer.php @@ -5,6 +5,8 @@ namespace Modules\EcommerceFinance\Services; use Illuminate\Database\ConnectionInterface; use Illuminate\Database\DatabaseManager; use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\Log; +use Illuminate\Support\Facades\Schema; use Illuminate\Support\Str; use RuntimeException; use Throwable; @@ -21,6 +23,8 @@ class DingProcessDetailMirrorSynchronizer private const SOURCE = 'operations_replica'; + private const APPROVAL_API_FALLBACK_LIMIT = 50; + private const STATUS_MAP = [ 'COMPLETED' => '已完成', 'RUNNING' => '审批中', @@ -32,9 +36,14 @@ class DingProcessDetailMirrorSynchronizer private readonly ConnectionInterface $connection; + private int $approvalApiAttempts = 0; + + private bool $approvalApiAvailable = true; + public function __construct( DatabaseManager $database, private readonly UpstreamReadOnlyQueryFactory $queries, + private readonly DingTalkProcessInstanceClient $approvalClient, ) { $this->connection = $database->connection( (string) config('upstream_mirrors.connection', 'business_mirror'), @@ -54,6 +63,8 @@ class DingProcessDetailMirrorSynchronizer */ public function sync(bool $forceFull = false, bool $dryRun = false): array { + $this->approvalApiAttempts = 0; + $this->approvalApiAvailable = true; $scope = $this->scope(); $state = $this->state(); $structuredState = $this->structuredState($state); @@ -132,6 +143,108 @@ class DingProcessDetailMirrorSynchronizer } } + /** + * Rebuild selected approvals from the source replica and fill missing + * payment components from DingTalk's current process-instance API. + * + * @param list $businessIds + * @return array{ + * status:string, + * run_mode:string, + * workflows:int, + * details:int, + * payments:int, + * amount_total:float, + * bank_amount_total:float + * } + */ + public function syncBusinessIds(array $businessIds, bool $dryRun = false): array + { + $businessIds = array_values(array_unique(array_filter(array_map( + static fn (mixed $value): string => trim((string) $value), + $businessIds, + )))); + if ($businessIds === []) { + throw new RuntimeException('至少需要一个审批编号。'); + } + if (count($businessIds) > self::APPROVAL_API_FALLBACK_LIMIT) { + throw new RuntimeException('单次最多按审批编号补取 50 个流程。'); + } + + $this->approvalApiAttempts = 0; + $this->approvalApiAvailable = true; + $scope = $this->scope(); + $state = $this->state(); + if (! $this->structuredState($state)) { + return [ + 'status' => 'requires_full_sync', + 'run_mode' => 'full', + 'workflows' => 0, + 'details' => 0, + 'payments' => 0, + 'amount_total' => 0.0, + 'bank_amount_total' => 0.0, + ]; + } + + $now = now(); + if (! $dryRun) { + $this->markAttempted($state, false, $now); + } + + try { + $placeholders = implode(',', array_fill(0, count($businessIds), '?')); + $extracted = $this->extractWorkflows( + "w.businessId IN ({$placeholders})", + $businessIds, + 'w.businessId', + 'businessId', + $now, + ); + $cursor = $this->stateCursor($state); + $extracted['metrics']['approval_api_cursor'] = (string) ( + $cursor['approval_api_cursor'] ?? '' + ); + + if ($dryRun) { + return [ + 'status' => 'validated', + 'run_mode' => 'business_ids', + 'workflows' => $extracted['metrics']['workflows'], + 'details' => $extracted['metrics']['details'], + 'payments' => $extracted['metrics']['payments'], + 'amount_total' => $extracted['metrics']['amount_total'], + 'bank_amount_total' => $extracted['metrics']['bank_amount_total'], + ]; + } + + $this->publish( + $extracted['details'], + $extracted['payments'], + $extracted['process_instance_ids'], + false, + $scope, + $extracted['metrics'], + $now, + ); + + return [ + 'status' => 'succeeded', + 'run_mode' => 'business_ids', + 'workflows' => $extracted['metrics']['workflows'], + 'details' => $extracted['metrics']['details'], + 'payments' => $extracted['metrics']['payments'], + 'amount_total' => $extracted['metrics']['amount_total'], + 'bank_amount_total' => $extracted['metrics']['bank_amount_total'], + ]; + } catch (Throwable $exception) { + if (! $dryRun) { + $this->markFailed($state, $exception, $now); + } + throw $exception; + } + } + /** * @return array{ * details:list>, @@ -283,11 +396,101 @@ class DingProcessDetailMirrorSynchronizer } } + $apiCursor = (string) ($cursor['approval_api_cursor'] ?? ''); + $fallbackIds = $this->approvalApiFallbackCandidates( + $apiCursor, + $result['process_instance_ids'], + ); + if ($fallbackIds !== []) { + foreach (array_chunk($fallbackIds, 25) as $chunk) { + $placeholders = implode(',', array_fill(0, count($chunk), '?')); + $fallbackResult = $this->extractWorkflows( + "w.processInstanceId IN ({$placeholders})", + $chunk, + 'w.processInstanceId', + 'processInstanceId', + $now, + ); + $result = $this->mergeExtracted($result, $fallbackResult); + } + $apiCursor = (string) $fallbackIds[array_key_last($fallbackIds)]; + } + $result['metrics']['scheme_high_watermark'] = $schemeHighWatermark; + $result['metrics']['approval_api_cursor'] = $apiCursor; return $result; } + /** + * Rotate through approvals whose payment component is absent from the + * replica. The cursor prevents one repeatedly failing approval from + * blocking the rest of the backlog. + * + * @param list $excludedProcessInstanceIds + * @return list + */ + private function approvalApiFallbackCandidates( + string $afterProcessInstanceId, + array $excludedProcessInstanceIds, + ): array { + if (! $this->approvalClient->isConfigured() + || ! Schema::connection($this->connection->getName())->hasColumn( + 'ding_process_detail_mirror', + 'payment_source_state', + )) { + return []; + } + $remainingAttempts = self::APPROVAL_API_FALLBACK_LIMIT - $this->approvalApiAttempts; + if (! $this->approvalApiAvailable || $remainingAttempts <= 0) { + return []; + } + + $base = $this->connection->table('ding_process_detail_mirror') + ->where('source_scope', $this->scope()) + ->whereIn('status', ['审批中', '已完成']) + ->where('payment_source_state', 'missing') + ->whereNotNull('process_instance_id') + ->where('process_instance_id', '!=', ''); + $excluded = array_fill_keys($excludedProcessInstanceIds, true); + + $ids = (clone $base) + ->when( + $afterProcessInstanceId !== '', + fn ($query) => $query->where('process_instance_id', '>', $afterProcessInstanceId), + ) + ->distinct() + ->orderBy('process_instance_id') + ->limit($remainingAttempts * 2) + ->pluck('process_instance_id') + ->map(static fn (mixed $value): string => trim((string) $value)) + ->filter() + ->values() + ->all(); + $ids = array_values(array_filter( + $ids, + static fn (string $id): bool => ! isset($excluded[$id]), + )); + + if ($ids === [] && $afterProcessInstanceId !== '') { + $ids = (clone $base) + ->distinct() + ->orderBy('process_instance_id') + ->limit($remainingAttempts * 2) + ->pluck('process_instance_id') + ->map(static fn (mixed $value): string => trim((string) $value)) + ->filter() + ->values() + ->all(); + $ids = array_values(array_filter( + $ids, + static fn (string $id): bool => ! isset($excluded[$id]), + )); + } + + return array_slice($ids, 0, $remainingAttempts); + } + /** * New scheme rows can be appended after the workflow itself stopped * changing. Track their numeric ID so late payment/details are discovered @@ -643,6 +846,20 @@ class DingProcessDetailMirrorSynchronizer $status = $this->status((string) ($workflow['status'] ?? '')); $businessMonth = $this->businessMonth($businessId, (string) ($workflow['joinAt'] ?? '')); $paymentSourceState = $this->paymentSourceState($process, $fields); + if ($this->needsApprovalApiFallback($process, $fields, $paymentSourceState)) { + $apiFields = $this->approvalApiFields($workflow); + if ($apiFields !== []) { + foreach ($apiFields as $fieldName => $fieldValue) { + if (! array_key_exists($fieldName, $fields) + || trim((string) ($fields[$fieldName] ?? '')) === '') { + $fields[$fieldName] = $fieldValue; + } + } + $paymentSourceState = $this->hasApprovalPaymentSource($process, $apiFields) + ? 'api' + : $this->paymentSourceState($process, $fields); + } + } [$workflowDetails, $workflowAmountCents, $detailTableName] = $this->detailRows( $workflow, @@ -921,6 +1138,93 @@ class DingProcessDetailMirrorSynchronizer return 'missing'; } + /** + * @param array $process + * @param array $fields + */ + private function needsApprovalApiFallback( + array $process, + array $fields, + string $paymentSourceState, + ): bool { + if ($paymentSourceState === 'missing') { + return true; + } + if ($paymentSourceState !== 'present') { + return false; + } + + foreach ($this->paymentSourceFieldNames($process) as $fieldName) { + if (trim((string) ($fields[$fieldName] ?? '')) === '') { + return true; + } + } + + return false; + } + + /** + * @param array $workflow + * @return array + */ + private function approvalApiFields(array $workflow): array + { + if (! $this->approvalClient->isConfigured() + || ! $this->approvalApiAvailable + || $this->approvalApiAttempts >= self::APPROVAL_API_FALLBACK_LIMIT) { + return []; + } + + $processInstanceId = trim((string) ($workflow['processInstanceId'] ?? '')); + if ($processInstanceId === '') { + return []; + } + + $this->approvalApiAttempts++; + try { + return $this->approvalClient->formValues($processInstanceId); + } catch (Throwable $exception) { + $this->approvalApiAvailable = false; + Log::warning('钉钉官方审批接口补取流程表单失败。', [ + 'process_instance_id' => $processInstanceId, + 'exception' => $exception::class, + ]); + + return []; + } + } + + /** + * @param array $process + * @param array $apiFields + */ + private function hasApprovalPaymentSource(array $process, array $apiFields): bool + { + foreach ($this->paymentSourceFieldNames($process) as $fieldName) { + if (array_key_exists($fieldName, $apiFields) + && trim((string) ($apiFields[$fieldName] ?? '')) !== '') { + return true; + } + } + + return false; + } + + /** + * @param array $process + * @return list + */ + private function paymentSourceFieldNames(array $process): array + { + return array_values(array_unique(array_filter([ + trim((string) ($process['payment_table'] ?? '')), + ...array_map( + static fn (mixed $value): string => trim((string) $value), + (array) ($process['payment_total_fields'] ?? []), + ), + ]))); + } + /** * @param string $businessId * @param array $process @@ -1343,7 +1647,9 @@ class DingProcessDetailMirrorSynchronizer $values = [ 'source_key' => self::SOURCE, - 'source_object' => 'ding_Workflow + ding_WorkflowScheme structured details', + 'source_object' => $this->approvalApiAttempts > 0 + ? 'ding_Workflow + ding_WorkflowScheme + DingTalk processInstance API fallback' + : 'ding_Workflow + ding_WorkflowScheme structured details', 'target_table' => 'ding_process_detail_mirror', 'status' => 'succeeded', 'run_mode' => $full ? 'full' : 'incremental', @@ -1358,6 +1664,8 @@ class DingProcessDetailMirrorSynchronizer 'amount_total' => round($amountTotal, 2), 'bank_amount_total' => round($bankAmountTotal, 2), 'scheme_high_watermark' => (int) ($metrics['scheme_high_watermark'] ?? 0), + 'approval_api_cursor' => (string) ($metrics['approval_api_cursor'] ?? ''), + 'approval_api_attempts' => $this->approvalApiAttempts, ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), 'high_watermark' => null, 'window_started_at' => null, diff --git a/modules/EcommerceFinance/Services/DingTalkProcessInstanceClient.php b/modules/EcommerceFinance/Services/DingTalkProcessInstanceClient.php new file mode 100644 index 00000000..2f48fb54 --- /dev/null +++ b/modules/EcommerceFinance/Services/DingTalkProcessInstanceClient.php @@ -0,0 +1,119 @@ + + */ + public function formValues(string $processInstanceId): array + { + $processInstanceId = trim($processInstanceId); + if ($processInstanceId === '') { + throw new RuntimeException('DingTalk process instance ID is required.'); + } + + try { + $response = Http::acceptJson() + ->timeout($this->timeout()) + ->withHeaders([ + 'x-acs-dingtalk-access-token' => $this->appAccessToken(), + ]) + ->get(self::PROCESS_INSTANCE_URL, [ + 'processInstanceId' => $processInstanceId, + ]); + } catch (RequestException $exception) { + throw new RuntimeException( + 'DingTalk process instance lookup failed.', + 0, + $exception, + ); + } + + if (! $response->successful()) { + throw new RuntimeException( + 'DingTalk process instance lookup returned HTTP ' . $response->status() . '.', + ); + } + + $result = $response->json('result'); + if (! is_array($result)) { + throw new RuntimeException('DingTalk process instance response was incomplete.'); + } + + $values = []; + foreach ((array) ($result['formComponentValues'] ?? []) as $component) { + if (! is_array($component)) { + continue; + } + $name = trim((string) ($component['name'] ?? '')); + if ($name !== '') { + $values[$name] = $component['value'] ?? null; + } + } + + return $values; + } + + private function appAccessToken(): string + { + if ($this->accessToken !== null && time() < $this->accessTokenExpiresAt) { + return $this->accessToken; + } + if (! $this->isConfigured()) { + throw new RuntimeException('DingTalk process API credentials are not configured.'); + } + + try { + $response = Http::acceptJson() + ->timeout($this->timeout()) + ->get((string) config('dingtalk.app_token_url'), [ + 'appkey' => (string) config('dingtalk.client_id'), + 'appsecret' => (string) config('dingtalk.client_secret'), + ]) + ->throw(); + } catch (RequestException $exception) { + throw new RuntimeException('DingTalk application token lookup failed.', 0, $exception); + } + + $token = trim((string) $response->json('access_token')); + if ((int) $response->json('errcode', -1) !== 0 || $token === '') { + throw new RuntimeException('DingTalk application token was unavailable.'); + } + + $expiresIn = max(300, (int) $response->json('expires_in', 7200)); + $this->accessToken = $token; + $this->accessTokenExpiresAt = time() + max( + 120, + $expiresIn - self::TOKEN_SAFETY_SECONDS, + ); + + return $token; + } + + private function timeout(): int + { + return max(3, min(30, (int) config('dingtalk.timeout', 10))); + } +} diff --git a/tests/Feature/Console/SyncDingProcessDetailMirrorTest.php b/tests/Feature/Console/SyncDingProcessDetailMirrorTest.php new file mode 100644 index 00000000..eecdf796 --- /dev/null +++ b/tests/Feature/Console/SyncDingProcessDetailMirrorTest.php @@ -0,0 +1,51 @@ +shouldReceive('syncBusinessIds') + ->once() + ->with( + ['202610151743000310225', '202610151743000310226'], + true, + ) + ->andReturn([ + 'status' => 'validated', + 'run_mode' => 'business_ids', + 'workflows' => 1, + 'details' => 2, + 'payments' => 1, + 'amount_total' => 15535.37, + 'bank_amount_total' => 15535.37, + ]); + $this->app->instance(DingProcessDetailMirrorSynchronizer::class, $synchronizer); + + $this->artisan('business-mirror:sync-ding-process-details', [ + '--business-id' => '202610151743000310225,202610151743000310226', + '--dry-run' => true, + ]) + ->expectsOutputToContain('校验通过：1 个审批，2 条明细，1 笔付款') + ->assertSuccessful(); + } + + public function test_full_and_business_id_cannot_be_combined(): void + { + $synchronizer = Mockery::mock(DingProcessDetailMirrorSynchronizer::class); + $synchronizer->shouldNotReceive('sync'); + $synchronizer->shouldNotReceive('syncBusinessIds'); + $this->app->instance(DingProcessDetailMirrorSynchronizer::class, $synchronizer); + + $this->artisan('business-mirror:sync-ding-process-details', [ + '--full' => true, + '--business-id' => '202610151743000310225', + ])->assertFailed(); + } +} diff --git a/tests/Unit/EcommerceFinance/DingProcessDetailMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorRepositoryTest.php index 8db5281a..884fd3cb 100644 --- a/tests/Unit/EcommerceFinance/DingProcessDetailMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorRepositoryTest.php @@ -164,6 +164,23 @@ class DingProcessDetailMirrorRepositoryTest extends TestCase ); } + public function test_official_api_payment_source_uses_normal_payment_states(): void + { + DB::connection('business_mirror') + ->table('ding_process_payment_mirror') + ->where('business_id', '202602031200000000002') + ->delete(); + DB::connection('business_mirror') + ->table('ding_process_detail_mirror') + ->where('business_id', '202602031200000000002') + ->update(['payment_source_state' => 'api']); + + $result = $this->repository->paginate(['payment_state' => '未付款'], 1, 50); + + $this->assertSame(2, $result['total']); + $this->assertSame('api', (string) $result['data'][0]->payment_source_state); + } + public function test_export_rows_returns_every_filtered_row(): void { $rows = iterator_to_array($this->repository->exportRows(['status' => '已完成'])); diff --git a/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php index 8a2a7d19..d02c9aaa 100644 --- a/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/DingProcessDetailMirrorSynchronizerTest.php @@ -7,6 +7,7 @@ use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Repositories\DingProcessDetailMirrorRepository; use Modules\EcommerceFinance\Services\DingProcessDetailMirrorSynchronizer; +use Modules\EcommerceFinance\Services\DingTalkProcessInstanceClient; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory; use RuntimeException; @@ -16,6 +17,8 @@ class DingProcessDetailMirrorSynchronizerTest extends TestCase { private FakeDingProcessDetailQuery $query; + private FakeDingTalkProcessInstanceClient $approvalClient; + private DingProcessDetailMirrorSynchronizer $synchronizer; protected function setUp(): void @@ -64,6 +67,8 @@ class DingProcessDetailMirrorSynchronizerTest extends TestCase } }; $this->app->instance(UpstreamReadOnlyQueryFactory::class, $factory); + $this->approvalClient = new FakeDingTalkProcessInstanceClient(); + $this->app->instance(DingTalkProcessInstanceClient::class, $this->approvalClient); $this->synchronizer = app(DingProcessDetailMirrorSynchronizer::class); } @@ -242,6 +247,155 @@ class DingProcessDetailMirrorSynchronizerTest extends TestCase $this->assertSame(101, (int) json_decode((string) $state->cursor, true)['scheme_high_watermark']); } + public function test_official_api_fills_payment_when_replica_omits_payment_table(): void + { + $workflow = [ + 'processInstanceId' => 'process-api-payment', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202610151743000310225', + 'title' => '李斯提交的线上代发业务报销', + 'status' => 'RUNNING', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-10-15 17:44:06', + 'updateAt' => '2026-10-15 17:44:06', + ]; + $this->query->workflows = [$workflow]; + $this->query->scheme = [ + '收款费用明细' => json_encode([[ + 'rowValue' => [ + ['label' => '代发订单金额', 'value' => '15535.37'], + ['label' => '科目编码', 'value' => '5601.05.03'], + ['label' => '科目名称', 'value' => '销售费用_营销费用_软件服务费'], + ], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '收款合计' => 15535.37, + ]; + $this->approvalClient->valuesByInstance['process-api-payment'] = [ + '付款明细' => json_encode([[ + 'rowValue' => [ + ['label' => '对应序号', 'value' => '1'], + ['label' => '付款方式', 'value' => '银行卡'], + ['label' => '付款金额', 'value' => '15535.37'], + ['label' => '科目编码', 'value' => '1002.01.544'], + ['label' => '科目名称', 'value' => '华夏银行-福尼达'], + ['label' => '打款时间', 'value' => '2026-10-18 00:00'], + ], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '付款金额合计' => 15535.37, + ]; + + $result = $this->synchronizer->sync(true); + + $this->assertSame(1, $result['payments']); + $row = DB::connection('business_mirror')->table('ding_process_detail_mirror')->first(); + $this->assertSame('api', $row->payment_source_state); + $this->assertSame('银行卡', $row->payment_method); + $this->assertSame('2026-10-18 00:00', $row->paid_at); + $this->assertSame(['process-api-payment'], $this->approvalClient->calls); + } + + public function test_incremental_sync_rotates_missing_payments_through_official_api(): void + { + $workflow = [ + 'processInstanceId' => 'process-api-queue', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202610151743000310226', + 'title' => '李斯提交的线上代发业务报销', + 'status' => 'RUNNING', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-10-15 17:44:06', + 'updateAt' => '2026-10-15 17:44:06', + ]; + $this->query->workflows = [$workflow]; + $this->query->scheme = [ + '收款费用明细' => json_encode([[ + 'rowValue' => [['label' => '代发订单金额', 'value' => '100.00']], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '收款合计' => 100, + ]; + $this->synchronizer->sync(true); + $this->assertSame('missing', DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->value('payment_source_state')); + + DB::connection('business_mirror') + ->table('business_mirror_table_states') + ->where('dataset_key', 'ding.process_details.personal_permission') + ->update(['last_succeeded_at' => now()->subMinutes(10)]); + $this->query->workflows = []; + $this->query->schemeChangedWorkflows = [$workflow]; + $this->approvalClient->valuesByInstance['process-api-queue'] = [ + '付款明细' => json_encode([[ + 'rowValue' => [ + ['label' => '对应序号', 'value' => '1'], + ['label' => '付款方式', 'value' => '银行卡'], + ['label' => '付款金额', 'value' => '100.00'], + ['label' => '科目编码', 'value' => '1002.01.544'], + ['label' => '打款时间', 'value' => '2026-10-18 00:00'], + ], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '付款金额合计' => 100, + ]; + + $result = $this->synchronizer->sync(false); + + $this->assertSame(1, $result['payments']); + $this->assertSame('api', DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->value('payment_source_state')); + $state = DB::connection('business_mirror') + ->table('business_mirror_table_states') + ->first(); + $this->assertSame( + 'process-api-queue', + json_decode((string) $state->cursor, true)['approval_api_cursor'], + ); + } + + public function test_business_id_dry_run_uses_official_api_without_publishing(): void + { + $workflow = [ + 'processInstanceId' => 'process-target-api', + 'processCode' => 'PROC-ACF09A14-B094-46B1-A9E2-8BB1F4E1FDCF', + 'businessId' => '202610151743000310227', + 'title' => '李斯提交的线上代发业务报销', + 'status' => 'RUNNING', + 'originatorUserId' => 'user-1', + 'joinAt' => '2026-10-15 17:44:06', + 'updateAt' => '2026-10-15 17:44:06', + ]; + $this->query->workflows = [$workflow]; + $this->query->scheme = [ + '收款费用明细' => json_encode([[ + 'rowValue' => [['label' => '代发订单金额', 'value' => '100.00']], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '收款合计' => 100, + ]; + $this->synchronizer->sync(true); + + $this->approvalClient->valuesByInstance['process-target-api'] = [ + '付款明细' => json_encode([[ + 'rowValue' => [ + ['label' => '对应序号', 'value' => '1'], + ['label' => '付款金额', 'value' => '100.00'], + ['label' => '付款方式', 'value' => '银行卡'], + ['label' => '打款时间', 'value' => '2026-10-18 00:00'], + ], + ]], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), + '付款金额合计' => 100, + ]; + + $result = $this->synchronizer->syncBusinessIds( + ['202610151743000310227'], + true, + ); + + $this->assertSame('validated', $result['status']); + $this->assertSame(1, $result['payments']); + $this->assertSame('missing', DB::connection('business_mirror') + ->table('ding_process_detail_mirror')->value('payment_source_state')); + $this->assertSame(0, DB::connection('business_mirror') + ->table('ding_process_payment_mirror')->count()); + } + /** @return array */ private function daifaScheme(): array { @@ -430,3 +584,26 @@ class FakeDingProcessDetailQuery extends UpstreamReadOnlyQuery return []; } } + +class FakeDingTalkProcessInstanceClient extends DingTalkProcessInstanceClient +{ + public bool $configured = true; + + /** @var array> */ + public array $valuesByInstance = []; + + /** @var list */ + public array $calls = []; + + public function isConfigured(): bool + { + return $this->configured; + } + + public function formValues(string $processInstanceId): array + { + $this->calls[] = $processInstanceId; + + return $this->valuesByInstance[$processInstanceId] ?? []; + } +}