From 17e9e7b6b65af47ea1b3f2fe5c6c88b7745d3d5b Mon Sep 17 00:00:00 2001 From: long <452591453@qq.com> Date: Thu, 10 Sep 2026 11:09:26 +0800 Subject: [PATCH] feat: add non-destructive EJ medicine sync --- docs/ej-pharmacy-incremental-medicine-sync.md | 46 ++ .../app/command/EjPharmacyPushMedicines.php | 76 +++ .../EjMedicineIncrementalPushService.php | 590 ++++++++++++++++++ server/config/console.php | 1 + .../pharmacy/incremental_medicine_push.php | 342 ++++++++++ server/tests/pharmacy/route_contracts.php | 32 + 6 files changed, 1087 insertions(+) create mode 100644 docs/ej-pharmacy-incremental-medicine-sync.md create mode 100644 server/app/command/EjPharmacyPushMedicines.php create mode 100644 server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php create mode 100644 server/tests/pharmacy/incremental_medicine_push.php diff --git a/docs/ej-pharmacy-incremental-medicine-sync.md b/docs/ej-pharmacy-incremental-medicine-sync.md new file mode 100644 index 000000000..8a3ff0821 --- /dev/null +++ b/docs/ej-pharmacy-incremental-medicine-sync.md @@ -0,0 +1,46 @@ +# EJ 药材非破坏性增量同步 + +`ej-pharmacy:push-medicines` 将 ZYT 中启用且未删除的药材，通过现有 HMAC OpenAPI 增量推送到 EJ。 + +## 不变式 + +- EJ 只执行 `POST /api/openapi/v1/medicine-imports` 的新增/幂等确认，不删除或清空 EJ 药材。 +- EJ 中由其他来源或人工录入的药材保持不变。 +- ZYT 只增量写入或更新 EJ 返回的目录投影，不清空目录和映射。 +- ZYT 中已停用或软删除的人工映射不重新启用。 +- 命令默认 dry-run；只有同时提供 `--apply` 和确认令牌才执行远端导入。 + +## 先预检 + +```bash +cd /www/wwwroot/zyt/server +php think ej-pharmacy:push-medicines +``` + +输出示例： + +```text +dry-run source=654 candidates=654 batches=7 mapped=654 unmapped=0 preserved_inactive=0 remote_delete=0 local_delete=0 +``` + +## 执行增量同步 + +```bash +cd /www/wwwroot/zyt/server +php think ej-pharmacy:push-medicines \ + --apply \ + --confirm=INCREMENTAL_NO_DELETE \ + --batch-size=100 +``` + +可用 `--run-id=` 固定本次批次的幂等标识；ID 最长 25 位。若省略，命令使用稳定默认值 `zyt-incremental`，相同内容重试时复用同一 import ID。若旧批次已经完成、但需要修复 EJ 中后来被单独移除的药材，应提供新的维修 run ID，例如 `--run-id=repair-20260910`。 + +执行结果会分别报告 EJ 新增、EJ 已存在、本地目录新增/更新、本地映射新增/更新/不变，以及两侧删除数量；删除数量固定为零。 + +## 方向说明 + +- `ej-pharmacy:push-medicines`：ZYT → EJ，非破坏性增量新增或恢复远端缺项。 +- `ej-pharmacy:sync-catalog`：EJ → ZYT，拉取 EJ 目录变化。 +- `ej-pharmacy:bootstrap-medicines`：一次性初始化并替换本地投影，不用于已有业务数据的生产环境增量同步。 + +如果 EJ 已存在 source 映射，但对应药材被停用或软删除，EJ 会返回冲突；此时需要先在 EJ 药材管理中恢复该药材，再重新执行增量命令。 diff --git a/server/app/command/EjPharmacyPushMedicines.php b/server/app/command/EjPharmacyPushMedicines.php new file mode 100644 index 000000000..cb97a8ff2 --- /dev/null +++ b/server/app/command/EjPharmacyPushMedicines.php @@ -0,0 +1,76 @@ +setName('ej-pharmacy:push-medicines') + ->setDescription('非破坏性增量推送 ZYT 药材到恩济药房；保留双方已有药材') + ->addOption('apply', null, Option::VALUE_NONE, '执行远端增量导入；缺省仅做 dry-run') + ->addOption('confirm', null, Option::VALUE_OPTIONAL, '增量写入确认令牌：INCREMENTAL_NO_DELETE', '') + ->addOption('batch-size', null, Option::VALUE_OPTIONAL, '每批药材数量（1-500）', 100) + ->addOption('run-id', null, Option::VALUE_OPTIONAL, '幂等运行标识（缺省 zyt-incremental，最长 25 位）', ''); + } + + protected function execute(Input $input, Output $output) + { + try { + $apply = (bool) $input->getOption('apply'); + $confirm = (string) $input->getOption('confirm'); + $batchSize = (int) $input->getOption('batch-size'); + EjMedicineIncrementalPushService::assertCommandGate($apply, $confirm, $batchSize); + + if (!$apply) { + $plan = EjMedicineIncrementalPushService::plan($batchSize); + $output->writeln(sprintf( + 'dry-run source=%d candidates=%d batches=%d mapped=%d unmapped=%d ' + . 'preserved_inactive=%d remote_delete=0 local_delete=0', + $plan['source_count'], + $plan['candidate_count'], + $plan['batch_count'], + $plan['mapped_count'], + $plan['unmapped_count'], + $plan['preserved_inactive'] + )); + return 0; + } + + $result = EjMedicineIncrementalPushService::execute( + $batchSize, + (string) $input->getOption('run-id') + ); + $output->writeln(sprintf( + 'incremental 完成 run_id=%s source=%d candidates=%d batches=%d ' + . 'remote_created=%d remote_existing=%d preserved_inactive=%d ' + . 'catalog_created=%d catalog_updated=%d mapping_created=%d mapping_updated=%d ' + . 'mapping_unchanged=%d remote_delete=0 local_delete=0', + $result['run_id'], + $result['source_count'], + $result['candidate_count'], + $result['batch_count'], + $result['remote_created'], + $result['remote_existing'], + $result['preserved_inactive'], + $result['catalog_created'], + $result['catalog_updated'], + $result['mapping_created'], + $result['mapping_updated'], + $result['mapping_unchanged'] + )); + return 0; + } catch (\Throwable $exception) { + $output->error($exception->getMessage()); + return 1; + } + } +} diff --git a/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php b/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php new file mode 100644 index 000000000..836722d81 --- /dev/null +++ b/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php @@ -0,0 +1,590 @@ +> $sourceLoader + * @param null|callable(array):array> $mappingLoader + * @return array{source_count:int,candidate_count:int,batch_count:int,mapped_count:int,unmapped_count:int,preserved_inactive:int,remote_delete_count:int,local_delete_count:int} + */ + public static function plan( + int $batchSize = 100, + ?callable $sourceLoader = null, + ?callable $mappingLoader = null + ): array { + self::assertBatchSize($batchSize); + $sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader(); + $items = EjMedicineBootstrapItem::fromRows($sourceRows); + $localIds = array_map('intval', array_column($items, 'source_medicine_id')); + $mappingRows = $mappingLoader === null + ? self::loadMappingRows($localIds) + : $mappingLoader($localIds); + $selection = self::selectCandidates($items, $mappingRows); + + return [ + 'source_count' => count($items), + 'candidate_count' => count($selection['candidates']), + 'batch_count' => (int) ceil(count($selection['candidates']) / $batchSize), + 'mapped_count' => $selection['mapped_count'], + 'unmapped_count' => $selection['unmapped_count'], + 'preserved_inactive' => $selection['preserved_inactive'], + 'remote_delete_count' => 0, + 'local_delete_count' => 0, + ]; + } + + /** + * Incrementally imports every active local medicine through EJ's idempotent + * source identity and upserts only the returned ZYT projection rows. + * Existing EJ-only medicines and unrelated local projections are untouched. + * + * @param null|callable():array> $sourceLoader + * @param null|callable(array):array $importer + * @param null|callable(array>):array $projectionUpserter + * @param null|callable(array):array> $mappingLoader + * @param null|callable(string):bool $lockAcquirer + * @param null|callable(string):bool $lockRenewer + * @param null|callable(string):void $lockReleaser + * @return array + */ + public static function execute( + int $batchSize = 100, + string $runId = '', + ?callable $sourceLoader = null, + ?callable $importer = null, + ?callable $projectionUpserter = null, + ?callable $mappingLoader = null, + ?callable $lockAcquirer = null, + ?callable $lockRenewer = null, + ?callable $lockReleaser = null + ): array { + self::assertBatchSize($batchSize); + + $customLockCallbacks = count(array_filter( + [$lockAcquirer, $lockRenewer, $lockReleaser], + static fn (?callable $callback): bool => $callback !== null + )); + if ($customLockCallbacks !== 0 && $customLockCallbacks !== 3) { + throw new InvalidArgumentException('增量同步锁回调必须同时提供 acquire、renew 和 release'); + } + $lockAcquirer ??= static fn (string $token): bool => self::acquireLock($token); + $lockRenewer ??= static fn (string $token): bool => self::renewLock($token); + $lockReleaser ??= static function (string $token): void { + self::releaseLock($token); + }; + + $lockToken = bin2hex(random_bytes(16)); + if (!$lockAcquirer($lockToken)) { + throw new DomainException('恩济药材同步正在执行，请稍后重试'); + } + + try { + return self::executeLocked( + $batchSize, + $runId, + $sourceLoader, + $importer, + $projectionUpserter, + $mappingLoader, + $lockRenewer, + $lockToken + ); + } finally { + $lockReleaser($lockToken); + } + } + + /** + * @param null|callable():array> $sourceLoader + * @param null|callable(array):array $importer + * @param null|callable(array>):array $projectionUpserter + * @param null|callable(array):array> $mappingLoader + * @param callable(string):bool $lockRenewer + * @return array + */ + private static function executeLocked( + int $batchSize, + string $runId, + ?callable $sourceLoader, + ?callable $importer, + ?callable $projectionUpserter, + ?callable $mappingLoader, + callable $lockRenewer, + string $lockToken + ): array { + $sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader(); + $allItems = EjMedicineBootstrapItem::fromRows($sourceRows); + $localIds = array_map('intval', array_column($allItems, 'source_medicine_id')); + $mappingRows = $mappingLoader === null + ? self::loadMappingRows($localIds) + : $mappingLoader($localIds); + $selection = self::selectCandidates($allItems, $mappingRows); + $items = $selection['candidates']; + $runId = self::normalizeRunId($runId); + $batches = self::buildBatches($items, $batchSize, $runId); + + if ($items !== [] && $importer === null) { + if (!EjPharmacyClient::isConfigured()) { + throw new RuntimeException('恩济药房接口未启用或配置不完整'); + } + $client = new EjPharmacyClient(); + $importer = static fn (array $payload): array => $client->importMedicines($payload); + } + + $sourceById = []; + foreach ($items as $item) { + $sourceById[(string) $item['source_medicine_id']] = $item; + } + $seenCodes = []; + $seenVersions = []; + $projectionRows = []; + $remoteCreated = 0; + $remoteExisting = 0; + foreach ($batches as $payload) { + self::assertLockLease($lockRenewer, $lockToken); + $response = $importer($payload); + self::assertLockLease($lockRenewer, $lockToken); + $responseItems = self::validateImportResponse( + $response, + $payload, + $seenCodes, + $seenVersions + ); + foreach ($responseItems as $responseItem) { + $sourceId = (string) $responseItem['source_medicine_id']; + $source = $sourceById[$sourceId] ?? null; + if (!is_array($source)) { + throw new RuntimeException("恩济增量导入返回未知 source_medicine_id：{$sourceId}"); + } + $action = (string) $responseItem['action']; + $medicineCode = (string) $responseItem['medicine_code']; + $expectedCode = $selection['active_mapping_codes'][$sourceId] ?? null; + if ($expectedCode !== null && !hash_equals($expectedCode, $medicineCode)) { + throw new DomainException( + "本地药材 {$sourceId} 的启用映射编码 {$expectedCode} 与恩济返回 {$medicineCode} 不一致" + ); + } + $remoteCreated += $action === 'created' ? 1 : 0; + $remoteExisting += $action === 'existing' ? 1 : 0; + $projectionRows[] = [ + 'local_medicine_id' => $sourceId, + 'medicine_code' => $medicineCode, + 'name' => (string) $source['name'], + 'brand' => (string) ($source['brand'] ?? ''), + 'unit' => (string) $source['unit'], + 'settlement_price' => (string) $source['settlement_price'], + 'retail_price' => (string) $source['retail_price'], + 'status' => (int) $source['status'], + 'catalog_version' => (int) $responseItem['catalog_version'], + ]; + } + } + + $projectionStats = [ + 'catalog_created' => 0, + 'catalog_updated' => 0, + 'mapping_created' => 0, + 'mapping_updated' => 0, + 'mapping_unchanged' => 0, + ]; + if ($projectionRows !== []) { + self::assertLockLease($lockRenewer, $lockToken); + $projectionStats = $projectionUpserter === null + ? self::upsertProjection($projectionRows, $lockToken) + : $projectionUpserter($projectionRows); + } + + return array_merge([ + 'run_id' => $runId, + 'source_count' => count($allItems), + 'candidate_count' => count($items), + 'batch_count' => count($batches), + 'preserved_inactive' => $selection['preserved_inactive'], + 'remote_created' => $remoteCreated, + 'remote_existing' => $remoteExisting, + 'remote_delete_count' => 0, + 'local_delete_count' => 0, + ], $projectionStats); + } + + /** + * @param list> $items + * @return list>}> + */ + public static function buildBatches(array $items, int $batchSize, string $runId): array + { + self::assertBatchSize($batchSize); + $runId = self::normalizeRunId($runId); + usort($items, static fn (array $left, array $right): int => EjMedicineBootstrapItem::compareIds( + (string) ($left['source_medicine_id'] ?? ''), + (string) ($right['source_medicine_id'] ?? '') + )); + + $batches = []; + foreach (array_chunk($items, $batchSize) as $index => $batchItems) { + $contentJson = json_encode( + $batchItems, + JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES + ); + $batches[] = [ + 'source_system' => self::SOURCE_SYSTEM, + 'import_id' => sprintf( + '%s-%04d-%s', + $runId, + $index + 1, + substr(hash('sha256', $contentJson), 0, 32) + ), + 'items' => $batchItems, + ]; + } + + return $batches; + } + + private static function assertBatchSize(int $batchSize): void + { + if ($batchSize < 1 || $batchSize > 500) { + throw new InvalidArgumentException('--batch-size 必须在 1 到 500 之间'); + } + } + + private static function normalizeRunId(string $runId): string + { + $runId = trim($runId); + if ($runId === '') { + $runId = self::DEFAULT_RUN_ID; + } + if (strlen($runId) > 25 || preg_match('/^[A-Za-z0-9][A-Za-z0-9._-]*$/D', $runId) !== 1) { + throw new InvalidArgumentException('--run-id 必须为不超过 25 位的字母、数字、点、下划线或短横线'); + } + + return $runId; + } + + /** @return array> */ + private static function loadSourceRows(): array + { + return Db::name('doctor_medicine') + ->field('id,name,unit,settlement_price,retail_price,status') + ->where('status', 1) + ->whereNull('delete_time') + ->order('id', 'asc') + ->select() + ->toArray(); + } + + /** @param array $localIds @return array> */ + private static function loadMappingRows(array $localIds): array + { + if ($localIds === []) { + return []; + } + + return Db::name('ej_medicine_mapping') + ->whereIn('local_medicine_id', $localIds) + ->field('id,local_medicine_id,medicine_code,status,delete_time') + ->order('local_medicine_id', 'asc') + ->select() + ->toArray(); + } + + /** + * Active mappings are replayed so EJ can recover a missing remote medicine. + * A medicine with an inactive or soft-deleted mapping is intentionally + * excluded; an operator decision must never be undone by synchronization. + * + * @param list> $items + * @param array> $mappingRows + * @return array{candidates:list>,mapped_count:int,unmapped_count:int,preserved_inactive:int,active_mapping_codes:array} + */ + private static function selectCandidates(array $items, array $mappingRows): array + { + $byLocalId = []; + foreach ($mappingRows as $mapping) { + $localId = (int) ($mapping['local_medicine_id'] ?? 0); + if ($localId < 1 || isset($byLocalId[$localId])) { + throw new DomainException("本地药材 {$localId} 存在重复的恩济映射记录"); + } + $byLocalId[$localId] = $mapping; + } + + $candidates = []; + $mappedCount = 0; + $unmappedCount = 0; + $preservedInactive = 0; + $activeMappingCodes = []; + foreach ($items as $item) { + $localId = (int) $item['source_medicine_id']; + $mapping = $byLocalId[$localId] ?? null; + if ($mapping === null) { + ++$unmappedCount; + $candidates[] = $item; + continue; + } + if ((int) ($mapping['status'] ?? 0) === 1 && ($mapping['delete_time'] ?? null) === null) { + $medicineCode = trim((string) ($mapping['medicine_code'] ?? '')); + if ($medicineCode === '') { + throw new DomainException("本地药材 {$localId} 的启用恩济映射编码为空"); + } + ++$mappedCount; + $candidates[] = $item; + $activeMappingCodes[(string) $localId] = $medicineCode; + continue; + } + ++$preservedInactive; + } + + return [ + 'candidates' => $candidates, + 'mapped_count' => $mappedCount, + 'unmapped_count' => $unmappedCount, + 'preserved_inactive' => $preservedInactive, + 'active_mapping_codes' => $activeMappingCodes, + ]; + } + + /** + * @param array $response + * @param array $payload + * @param array $seenCodes + * @param array $seenVersions + * @return list + */ + private static function validateImportResponse( + array $response, + array $payload, + array &$seenCodes, + array &$seenVersions + ): array { + $nextSeenCodes = $seenCodes; + $nextSeenVersions = $seenVersions; + $items = EjMedicineBootstrapService::validateImportResponse( + $response, + $payload, + $nextSeenCodes, + $nextSeenVersions + ); + + $data = $response['body']['data'] ?? null; + if (!is_array($data) || !hash_equals(self::SOURCE_SYSTEM, (string) ($data['source_system'] ?? ''))) { + throw new RuntimeException('恩济药材导入响应 source_system 不匹配'); + } + $expectedPayloadHash = hash('sha256', json_encode( + self::canonicalize($payload), + JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES + )); + $actualPayloadHash = strtolower(trim((string) ($data['payload_hash'] ?? ''))); + if ( + preg_match('/^[a-f0-9]{64}$/D', $actualPayloadHash) !== 1 + || !hash_equals($expectedPayloadHash, $actualPayloadHash) + ) { + throw new RuntimeException('恩济药材导入响应 payload_hash 不匹配'); + } + + $seenCodes = $nextSeenCodes; + $seenVersions = $nextSeenVersions; + return $items; + } + + private static function canonicalize(mixed $value): mixed + { + if (!is_array($value)) { + return $value; + } + if (array_is_list($value)) { + return array_map([self::class, 'canonicalize'], $value); + } + ksort($value, SORT_STRING); + foreach ($value as $key => $child) { + $value[$key] = self::canonicalize($child); + } + return $value; + } + + /** @param callable(string):bool $lockRenewer */ + private static function assertLockLease(callable $lockRenewer, string $lockToken): void + { + if (!$lockRenewer($lockToken)) { + throw new DomainException('恩济药材同步锁已失效，请重试'); + } + } + + private static function acquireLock(string $token): bool + { + $now = time(); + return Db::name('ej_pharmacy_sync_state') + ->where('id', self::STATE_ID) + ->where(function ($query) use ($now): void { + $query->where('lock_token', '')->whereOr('lock_expires_at', '<', $now); + }) + ->update([ + 'lock_token' => $token, + 'lock_expires_at' => $now + self::LOCK_TTL, + 'update_time' => $now, + ]) === 1; + } + + private static function renewLock(string $token): bool + { + $now = time(); + $query = Db::name('ej_pharmacy_sync_state') + ->where('id', self::STATE_ID) + ->where('lock_token', $token) + ->where('lock_expires_at', '>=', $now); + $updated = $query->update([ + 'lock_expires_at' => $now + self::LOCK_TTL, + 'update_time' => $now, + ]); + if ($updated === 1) { + return true; + } + + $state = Db::name('ej_pharmacy_sync_state') + ->where('id', self::STATE_ID) + ->where('lock_token', $token) + ->find(); + return is_array($state) && (int) ($state['lock_expires_at'] ?? 0) >= $now; + } + + private static function releaseLock(string $token): void + { + Db::name('ej_pharmacy_sync_state') + ->where('id', self::STATE_ID) + ->where('lock_token', $token) + ->update([ + 'lock_token' => '', + 'lock_expires_at' => 0, + 'update_time' => time(), + ]); + } + + /** + * @param array> $rows + * @return array{catalog_created:int,catalog_updated:int,mapping_created:int,mapping_updated:int,mapping_unchanged:int} + */ + private static function upsertProjection(array $rows, string $lockToken): array + { + return Db::transaction(static function () use ($rows, $lockToken): array { + $stats = [ + 'catalog_created' => 0, + 'catalog_updated' => 0, + 'mapping_created' => 0, + 'mapping_updated' => 0, + 'mapping_unchanged' => 0, + ]; + $state = Db::name('ej_pharmacy_sync_state') + ->where('id', self::STATE_ID) + ->lock(true) + ->find(); + if ( + !is_array($state) + || !hash_equals($lockToken, (string) ($state['lock_token'] ?? '')) + || (int) ($state['lock_expires_at'] ?? 0) < time() + ) { + throw new DomainException('恩济药材同步锁已失效，请重试'); + } + $now = time(); + foreach ($rows as $row) { + $localId = (int) ($row['local_medicine_id'] ?? 0); + $medicineCode = trim((string) ($row['medicine_code'] ?? '')); + if ($localId < 1 || $medicineCode === '') { + throw new RuntimeException('恩济增量导入投影缺少本地药材 ID 或 medicine_code'); + } + + $conflict = Db::name('ej_medicine_mapping') + ->where('medicine_code', $medicineCode) + ->where('local_medicine_id', '<>', $localId) + ->lock(true) + ->find(); + if ($conflict) { + throw new DomainException( + "恩济药材编码 {$medicineCode} 已映射到本地药材 " . (int) $conflict['local_medicine_id'] + ); + } + + $catalogValues = [ + 'name' => (string) $row['name'], + 'brand' => (string) ($row['brand'] ?? ''), + 'unit' => (string) $row['unit'], + 'settlement_price' => (string) $row['settlement_price'], + 'retail_price' => (string) $row['retail_price'], + 'status' => (int) $row['status'], + 'catalog_version' => (int) $row['catalog_version'], + 'remote_deleted' => 0, + 'update_time' => $now, + ]; + $catalog = Db::name('ej_medicine_catalog') + ->where('medicine_code', $medicineCode) + ->lock(true) + ->find(); + if ($catalog) { + Db::name('ej_medicine_catalog')->where('id', (int) $catalog['id'])->update($catalogValues); + ++$stats['catalog_updated']; + } else { + Db::name('ej_medicine_catalog')->insert($catalogValues + [ + 'medicine_code' => $medicineCode, + 'create_time' => $now, + ]); + ++$stats['catalog_created']; + } + + $mapping = Db::name('ej_medicine_mapping') + ->where('local_medicine_id', $localId) + ->lock(true) + ->find(); + $mappingValues = [ + 'medicine_code' => $medicineCode, + 'status' => 1, + 'operator_id' => 0, + 'operator_name' => 'system-incremental', + 'delete_time' => null, + 'update_time' => $now, + ]; + if (!$mapping) { + Db::name('ej_medicine_mapping')->insert($mappingValues + [ + 'local_medicine_id' => $localId, + 'create_time' => $now, + ]); + ++$stats['mapping_created']; + } elseif ((int) $mapping['status'] !== 1 || ($mapping['delete_time'] ?? null) !== null) { + throw new DomainException("本地药材 {$localId} 的恩济映射已被停用，增量同步保持该状态不变"); + } elseif (hash_equals((string) $mapping['medicine_code'], $medicineCode)) { + ++$stats['mapping_unchanged']; + } else { + throw new DomainException( + "本地药材 {$localId} 的启用映射编码 " + . (string) $mapping['medicine_code'] + . " 与恩济返回 {$medicineCode} 不一致，增量同步未改写该映射" + ); + } + } + + return $stats; + }); + } +} diff --git a/server/config/console.php b/server/config/console.php index 6de151f83..dd547f59f 100755 --- a/server/config/console.php +++ b/server/config/console.php @@ -41,6 +41,7 @@ return [ 'gancao:sync-logistics' => 'app\\command\\GancaoSyncLogisticsRoute', 'ej-pharmacy:sync-catalog' => 'app\\command\\EjPharmacySyncCatalog', 'ej-pharmacy:bootstrap-medicines' => 'app\\command\\EjPharmacyBootstrapMedicines', + 'ej-pharmacy:push-medicines' => 'app\\command\\EjPharmacyPushMedicines', // 历史 internal_cost：批量甘草预报价回填（CTM_PREVIEW） 'tcm:backfill-internal-cost' => 'app\\command\\TcmBackfillPrescriptionOrderInternalCost', // 批量回填业务订单签收时间到物流库（导出读库即可，不再逐单 HTTP） diff --git a/server/tests/pharmacy/incremental_medicine_push.php b/server/tests/pharmacy/incremental_medicine_push.php new file mode 100644 index 000000000..baf1acbe0 --- /dev/null +++ b/server/tests/pharmacy/incremental_medicine_push.php @@ -0,0 +1,342 @@ + $child) { + $value[$key] = $canonicalize($child); + } + return $value; +}; +$canonicalPayloadHash = static fn (array $payload): string => hash( + 'sha256', + json_encode($canonicalize($payload), JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) +); + +$sourceRows = [ + [ + 'id' => 101, + 'name' => '增量药材甲', + 'unit' => '克', + 'settlement_price' => '1.100000', + 'retail_price' => '1.200000', + 'status' => 1, + ], + [ + 'id' => 202, + 'name' => '增量药材乙', + 'unit' => '克', + 'settlement_price' => '2.100000', + 'retail_price' => '2.200000', + 'status' => 1, + ], + [ + 'id' => 303, + 'name' => '人工停用映射药材', + 'unit' => '克', + 'settlement_price' => '3.100000', + 'retail_price' => '3.200000', + 'status' => 1, + ], +]; + +EjMedicineIncrementalPushService::assertCommandGate(false, '', 100); +$blocked = false; +try { + EjMedicineIncrementalPushService::assertCommandGate(true, '', 100); +} catch (InvalidArgumentException $exception) { + $blocked = str_contains($exception->getMessage(), 'INCREMENTAL_NO_DELETE'); +} +$assertTrue($blocked, 'apply mode must require the non-destructive confirmation token'); +foreach ([0, 501] as $invalidBatchSize) { + try { + EjMedicineIncrementalPushService::assertCommandGate(false, '', $invalidBatchSize); + throw new RuntimeException('invalid incremental batch size unexpectedly accepted'); + } catch (InvalidArgumentException $exception) { + $assertTrue(str_contains($exception->getMessage(), 'batch-size'), 'invalid batch size must return an actionable error'); + } +} + +$plan = EjMedicineIncrementalPushService::plan( + 100, + static fn (): array => $sourceRows, + static fn (array $localIds): array => [ + ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null], + ['id' => 2, 'local_medicine_id' => 303, 'medicine_code' => 'EJ-303', 'status' => 0, 'delete_time' => 123], + ] +); +$assertSame(3, $plan['source_count'], 'dry-run must count every active local medicine'); +$assertSame(2, $plan['candidate_count'], 'dry-run must replay active mappings and add never-mapped medicines'); +$assertSame(1, $plan['batch_count'], 'dry-run must report the remote batch count'); +$assertSame(1, $plan['mapped_count'], 'dry-run must count existing active mappings'); +$assertSame(1, $plan['unmapped_count'], 'dry-run must identify local medicines without an active mapping'); +$assertSame(1, $plan['preserved_inactive'], 'dry-run must preserve manually disabled or unlinked mappings'); +$assertSame(0, $plan['remote_delete_count'], 'incremental push must never plan remote deletes'); +$assertSame(0, $plan['local_delete_count'], 'incremental push must never plan local projection deletes'); + +$capturedPayloads = []; +$capturedProjection = []; +$heldLockToken = ''; +$lockAcquireCount = 0; +$lockRenewCount = 0; +$lockReleaseCount = 0; +$result = EjMedicineIncrementalPushService::execute( + 1, + 'zyt-inc-contract', + static fn (): array => $sourceRows, + static function (array $payload) use (&$capturedPayloads, $canonicalPayloadHash): array { + $capturedPayloads[] = $payload; + $sourceId = (string) $payload['items'][0]['source_medicine_id']; + $action = $sourceId === '101' ? 'existing' : 'created'; + return [ + 'http_status' => $action === 'created' ? 201 : 200, + 'body' => [ + 'code' => 0, + 'message' => 'ok', + 'data' => [ + 'source_system' => 'zyt', + 'import_id' => $payload['import_id'], + 'payload_hash' => $canonicalPayloadHash($payload), + 'idempotent' => false, + 'item_count' => 1, + 'created_count' => $action === 'created' ? 1 : 0, + 'existing_count' => $action === 'existing' ? 1 : 0, + 'items' => [[ + 'source_medicine_id' => $sourceId, + 'medicine_code' => 'EJ-' . $sourceId, + 'catalog_version' => $sourceId === '101' ? 10 : 11, + 'action' => $action, + ]], + ], + ], + 'request_id' => 'request-' . $sourceId, + ]; + }, + static function (array $rows) use (&$capturedProjection): array { + $capturedProjection = $rows; + return [ + 'catalog_created' => 1, + 'catalog_updated' => 1, + 'mapping_created' => 1, + 'mapping_updated' => 0, + 'mapping_unchanged' => 1, + ]; + }, + static fn (array $localIds): array => [ + ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null], + ['id' => 2, 'local_medicine_id' => 303, 'medicine_code' => 'EJ-303', 'status' => 0, 'delete_time' => 123], + ], + static function (string $token) use (&$heldLockToken, &$lockAcquireCount): bool { + ++$lockAcquireCount; + if ($heldLockToken !== '') { + return false; + } + $heldLockToken = $token; + return true; + }, + static function (string $token) use (&$heldLockToken, &$lockRenewCount): bool { + ++$lockRenewCount; + return $heldLockToken !== '' && hash_equals($heldLockToken, $token); + }, + static function (string $token) use (&$heldLockToken, &$lockReleaseCount): void { + ++$lockReleaseCount; + if ($heldLockToken !== '' && hash_equals($heldLockToken, $token)) { + $heldLockToken = ''; + } + } +); + +$assertSame(2, count($capturedPayloads), 'apply must send deterministic bounded batches'); +$assertTrue( + str_starts_with((string) $capturedPayloads[0]['import_id'], 'zyt-inc-contract-0001-'), + 'incremental import id must include the caller-provided run id and batch ordinal' +); +$assertSame(['101', '202'], array_column($capturedProjection, 'local_medicine_id'), 'projection upsert must retain source identity'); +$assertSame(['EJ-101', 'EJ-202'], array_column($capturedProjection, 'medicine_code'), 'projection upsert must use EJ-returned codes'); +$assertSame(1, $result['remote_created'], 'result must report remotely created medicines'); +$assertSame(1, $result['remote_existing'], 'result must report idempotently existing medicines'); +$assertSame(1, $result['preserved_inactive'], 'apply must not reactivate an operator-disabled mapping'); +$assertSame(0, $result['remote_delete_count'], 'apply result must prove no EJ medicine was deleted'); +$assertSame(0, $result['local_delete_count'], 'apply result must prove no ZYT projection was deleted'); +$assertSame(1, $lockAcquireCount, 'apply must acquire one shared EJ synchronization lease'); +$assertSame(5, $lockRenewCount, 'apply must renew its lease around every remote batch and before projection writes'); +$assertSame(1, $lockReleaseCount, 'apply must release its synchronization lease'); +$assertSame('', $heldLockToken, 'successful apply must not leave the synchronization lease held'); + +$candidateItems = array_merge($capturedPayloads[0]['items'], $capturedPayloads[1]['items']); +$defaultBatches = EjMedicineIncrementalPushService::buildBatches($candidateItems, 1, ''); +$retryBatches = EjMedicineIncrementalPushService::buildBatches(array_reverse($candidateItems), 1, ''); +$assertSame( + array_column($defaultBatches, 'import_id'), + array_column($retryBatches, 'import_id'), + 'default incremental import ids must be deterministic for identical content' +); +$assertTrue( + str_starts_with((string) $defaultBatches[0]['import_id'], 'zyt-incremental-0001-'), + 'default run id must be stable so an interrupted apply can safely retry' +); + +$responseFor = static function ( + array $payload, + string $medicineCode, + string $sourceSystem = 'zyt', + ?string $responsePayloadHash = null +) use ($canonicalPayloadHash): array { + $sourceId = (string) $payload['items'][0]['source_medicine_id']; + return [ + 'http_status' => 200, + 'body' => ['code' => 0, 'message' => 'ok', 'data' => [ + 'source_system' => $sourceSystem, + 'import_id' => $payload['import_id'], + 'payload_hash' => $responsePayloadHash ?? $canonicalPayloadHash($payload), + 'idempotent' => true, + 'item_count' => 1, + 'created_count' => 0, + 'existing_count' => 1, + 'items' => [[ + 'source_medicine_id' => $sourceId, + 'medicine_code' => $medicineCode, + 'catalog_version' => 99, + 'action' => 'existing', + ]], + ]], + 'request_id' => 'validation-' . $sourceId, + ]; +}; +$alwaysAcquire = static fn (string $token): bool => $token !== ''; +$alwaysRenew = static fn (string $token): bool => $token !== ''; +$releaseNoop = static function (string $token): void {}; + +foreach ([ + 'source_system' => static fn (array $payload): array => $responseFor($payload, 'EJ-202', 'other'), + 'payload_hash' => static fn (array $payload): array => $responseFor($payload, 'EJ-202', 'zyt', str_repeat('0', 64)), +] as $field => $invalidImporter) { + $projectionCalled = false; + try { + EjMedicineIncrementalPushService::execute( + 1, + 'response-validation', + static fn (): array => [$sourceRows[1]], + $invalidImporter, + static function () use (&$projectionCalled): array { + $projectionCalled = true; + return []; + }, + static fn (array $localIds): array => [], + $alwaysAcquire, + $alwaysRenew, + $releaseNoop + ); + throw new RuntimeException("invalid {$field} unexpectedly accepted"); + } catch (RuntimeException $exception) { + $assertTrue(str_contains($exception->getMessage(), $field), "incremental response must reject invalid {$field}"); + } + $assertSame(false, $projectionCalled, "invalid {$field} must block local projection writes"); +} + +$projectionCalled = false; +try { + EjMedicineIncrementalPushService::execute( + 1, + 'mapping-mismatch', + static fn (): array => [$sourceRows[0]], + static fn (array $payload): array => $responseFor($payload, 'EJ-DIFFERENT'), + static function () use (&$projectionCalled): array { + $projectionCalled = true; + return []; + }, + static fn (array $localIds): array => [ + ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null], + ], + $alwaysAcquire, + $alwaysRenew, + $releaseNoop + ); + throw new RuntimeException('active mapping code mismatch unexpectedly accepted'); +} catch (DomainException $exception) { + $assertTrue(str_contains($exception->getMessage(), '不一致'), 'active mapping replay must reject a different EJ code'); +} +$assertSame(false, $projectionCalled, 'active mapping mismatch must not overwrite its local projection'); + +$sourceLoaded = false; +$failedAcquireReleased = false; +try { + EjMedicineIncrementalPushService::execute( + 1, + 'concurrent-run', + static function () use (&$sourceLoaded): array { + $sourceLoaded = true; + return []; + }, + null, + null, + null, + static fn (string $token): bool => false, + $alwaysRenew, + static function (string $token) use (&$failedAcquireReleased): void { $failedAcquireReleased = true; } + ); + throw new RuntimeException('concurrent incremental run unexpectedly acquired the lease'); +} catch (DomainException $exception) { + $assertTrue(str_contains($exception->getMessage(), '正在执行'), 'concurrent apply must fail with a business error'); +} +$assertSame(false, $sourceLoaded, 'concurrent apply must stop before reading or pushing source medicines'); +$assertSame(false, $failedAcquireReleased, 'a failed lock acquisition must not release another run lease'); + +$leaseRenewals = 0; +$leaseFailureReleased = false; +$projectionCalled = false; +try { + EjMedicineIncrementalPushService::execute( + 1, + 'lease-loss', + static fn (): array => [$sourceRows[1]], + static fn (array $payload): array => $responseFor($payload, 'EJ-202'), + static function () use (&$projectionCalled): array { + $projectionCalled = true; + return []; + }, + static fn (array $localIds): array => [], + $alwaysAcquire, + static function () use (&$leaseRenewals): bool { + ++$leaseRenewals; + return $leaseRenewals === 1; + }, + static function (string $token) use (&$leaseFailureReleased): void { $leaseFailureReleased = true; } + ); + throw new RuntimeException('expired synchronization lease unexpectedly allowed projection'); +} catch (DomainException $exception) { + $assertTrue(str_contains($exception->getMessage(), '锁已失效'), 'lease loss must return an actionable retry error'); +} +$assertSame(false, $projectionCalled, 'lease loss after a remote response must block local projection writes'); +$assertSame(true, $leaseFailureReleased, 'lease loss must still release the caller-owned lock token'); + +$serviceSource = (string) file_get_contents(dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineIncrementalPushService.php'); +$assertTrue(!str_contains($serviceSource, '->delete('), 'incremental service must not call model delete'); +$assertTrue(!str_contains($serviceSource, '->delete()'), 'incremental service must not call model delete without arguments'); + +echo "incremental EJ medicine push tests passed: {$passed}\n"; diff --git a/server/tests/pharmacy/route_contracts.php b/server/tests/pharmacy/route_contracts.php index 2b8aaa14c..613322a96 100644 --- a/server/tests/pharmacy/route_contracts.php +++ b/server/tests/pharmacy/route_contracts.php @@ -39,9 +39,13 @@ $assertTrue( $bootstrapItemPath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineBootstrapItem.php'; $bootstrapServicePath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineBootstrapService.php'; $bootstrapCommandPath = dirname(__DIR__, 2) . '/app/command/EjPharmacyBootstrapMedicines.php'; +$incrementalServicePath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineIncrementalPushService.php'; +$incrementalCommandPath = dirname(__DIR__, 2) . '/app/command/EjPharmacyPushMedicines.php'; $assertTrue(is_file($bootstrapItemPath), 'ZYT must provide a dedicated EJ medicine bootstrap item normalizer'); $assertTrue(is_file($bootstrapServicePath), 'ZYT must provide an atomic EJ medicine bootstrap service'); $assertTrue(is_file($bootstrapCommandPath), 'ZYT must provide the ej-pharmacy:bootstrap-medicines command'); +$assertTrue(is_file($incrementalServicePath), 'ZYT must provide a non-destructive EJ medicine incremental push service'); +$assertTrue(is_file($incrementalCommandPath), 'ZYT must provide the ej-pharmacy:push-medicines command'); $clientSource = (string) file_get_contents(dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjPharmacyClient.php'); $configSource = (string) file_get_contents(dirname(__DIR__, 2) . '/config/ej_pharmacy.php'); @@ -62,6 +66,34 @@ $assertTrue( && str_contains((string) file_get_contents($bootstrapCommandPath), 'RESET_TEST_CATALOG'), 'bootstrap command must be registered and guard destructive replacement with the exact confirmation token' ); +$incrementalServiceSource = (string) file_get_contents($incrementalServicePath); +$incrementalCommandSource = (string) file_get_contents($incrementalCommandPath); +$assertTrue( + str_contains($consoleSource, "'ej-pharmacy:push-medicines'") + && str_contains($incrementalCommandSource, 'INCREMENTAL_NO_DELETE') + && str_contains($incrementalCommandSource, "addOption('apply'") + && str_contains($incrementalCommandSource, 'dry-run'), + 'incremental push command must be registered, dry-run by default, and require the non-delete confirmation token' +); +$assertTrue( + !str_contains($incrementalServiceSource, '->delete(') + && str_contains($incrementalServiceSource, "'remote_delete_count' => 0") + && str_contains($incrementalServiceSource, "'local_delete_count' => 0"), + 'incremental medicine push must preserve EJ-only medicines and unrelated local projections' +); +$assertTrue( + str_contains($incrementalServiceSource, "private const DEFAULT_RUN_ID = 'zyt-incremental'") + && str_contains($incrementalServiceSource, 'private static function acquireLock(') + && str_contains($incrementalServiceSource, "->where('lock_token', '')") + && str_contains($incrementalServiceSource, "'lock_expires_at'"), + 'incremental medicine apply must use deterministic retry ids and share the EJ synchronization lease' +); +$assertTrue( + str_contains($incrementalServiceSource, "['source_system']") + && str_contains($incrementalServiceSource, "['payload_hash']") + && str_contains($incrementalServiceSource, '增量同步未改写该映射'), + 'incremental medicine apply must validate response identity and never rewrite an active mapping to a different code' +); $assertTrue( str_contains($configSource, "'catalog_sync_enabled'") && str_contains($configSource, "env('EJ_PHARMACY_CATALOG_SYNC_ENABLED', false)"),