From 2a0e49dbc9a01b131a34fe574cda36bdde5a1f7a Mon Sep 17 00:00:00 2001 From: long <452591453@qq.com> Date: Thu, 10 Sep 2026 11:11:06 +0800 Subject: [PATCH] docs: record EJ incremental sync verification --- .../ej-medicine-incremental-sync/DIFF_FILE | 1230 +++++++++++++++++ .../MODIFIED_FILE | 590 ++++++++ .../ej-medicine-incremental-sync/ROLLBACK.sh | 22 + .../VERIFICATION.txt | 98 ++ 4 files changed, 1940 insertions(+) create mode 100644 artifacts/ej-medicine-incremental-sync/DIFF_FILE create mode 100644 artifacts/ej-medicine-incremental-sync/MODIFIED_FILE create mode 100755 artifacts/ej-medicine-incremental-sync/ROLLBACK.sh create mode 100644 artifacts/ej-medicine-incremental-sync/VERIFICATION.txt diff --git a/artifacts/ej-medicine-incremental-sync/DIFF_FILE b/artifacts/ej-medicine-incremental-sync/DIFF_FILE new file mode 100644 index 000000000..14ac381bc --- /dev/null +++ b/artifacts/ej-medicine-incremental-sync/DIFF_FILE @@ -0,0 +1,1230 @@ +### ZYT /Users/long/Work/zyt commit 17e9e7b6b +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)"), +### EJ /Users/long/Work/ej commit 2637ffa +diff --git a/server/app/common/service/pharmacy/MedicineImportService.php b/server/app/common/service/pharmacy/MedicineImportService.php +index ab0fbc3..c7aeafa 100644 +--- a/server/app/common/service/pharmacy/MedicineImportService.php ++++ b/server/app/common/service/pharmacy/MedicineImportService.php +@@ -65,8 +65,6 @@ final class MedicineImportService + return self::batchReplay($input, $sourceRows, $medicineRows); + } + +- self::assertCatalogBelongsToBootstrap($medicineRows, $sourceRows); +- + $createdCount = 0; + $existingCount = 0; + $results = []; +@@ -149,20 +147,6 @@ final class MedicineImportService + return self::response($input, $results, $createdCount, $existingCount, false); + } + +- /** @param array> $medicineRows @param array> $sourceRows */ +- private static function assertCatalogBelongsToBootstrap(array $medicineRows, array $sourceRows): void +- { +- $medicineIds = array_map('intval', array_keys($medicineRows)); +- $mappedMedicineIds = array_map( +- static fn (array $row): int => (int) $row['medicine_id'], +- array_values($sourceRows) +- ); +- $unmappedMedicineIds = array_diff($medicineIds, $mappedMedicineIds); +- if ($unmappedMedicineIds !== []) { +- throw new DomainException('Medicine import preflight found unrelated existing catalog rows'); +- } +- } +- + /** + * @param array $input + * @param array> $sourceRows +diff --git a/server/tests/pharmacy/medicine_import_mysql_integration.php b/server/tests/pharmacy/medicine_import_mysql_integration.php +index 34b6b7a..4b0b065 100644 +--- a/server/tests/pharmacy/medicine_import_mysql_integration.php ++++ b/server/tests/pharmacy/medicine_import_mysql_integration.php +@@ -360,7 +360,7 @@ try { + $assertSame(0, (int) $medicineTwo['status'], 'continued import disabled status must persist'); + ++$passed; + +- Db::name('pharmacy_medicine')->insert([ ++ $unrelatedMedicineId = Db::name('pharmacy_medicine')->insertGetId([ + 'medicine_code' => 'IT_UNRELATED', + 'name' => 'Unrelated old medicine', + 'brand' => '', +@@ -375,14 +375,13 @@ try { + $unrelatedAttempt = $continuation; + $unrelatedAttempt['import_id'] = 'integration-batch-0005'; + $unrelatedAttempt['items'][0]['source_medicine_id'] = '103'; +- $rejected = false; +- try { +- MedicineImportService::import(9001, $unrelatedAttempt); +- } catch (DomainException $exception) { +- $rejected = str_contains($exception->getMessage(), 'unrelated existing catalog'); +- } +- $assertTrue($rejected, 'unmapped old medicine must fail bootstrap preflight'); +- $assertSame(2, $count('pharmacy_medicine_source'), 'preflight rejection must not add a source mapping'); ++ $unrelatedBefore = Db::name('pharmacy_medicine')->where('id', $unrelatedMedicineId)->find(); ++ $incrementalResult = MedicineImportService::import(9001, $unrelatedAttempt); ++ $unrelatedAfter = Db::name('pharmacy_medicine')->where('id', $unrelatedMedicineId)->find(); ++ $assertSame('created', $incrementalResult['items'][0]['action'], 'unrelated EJ medicines must not block incremental ZYT additions'); ++ $assertSame(3, $count('pharmacy_medicine_source'), 'incremental import must add only the new ZYT source mapping'); ++ $assertSame($unrelatedBefore, $unrelatedAfter, 'incremental import must preserve unrelated EJ medicines byte for byte'); ++ $assertSame(4, $count('pharmacy_medicine'), 'incremental import must append without replacing the EJ catalog'); + ++$passed; + + config([ +diff --git a/server/tests/pharmacy/route_contracts.php b/server/tests/pharmacy/route_contracts.php +index 39deace..7ce525c 100644 +--- a/server/tests/pharmacy/route_contracts.php ++++ b/server/tests/pharmacy/route_contracts.php +@@ -94,8 +94,9 @@ foreach ([ + } + $assertTrue(!str_contains($medicineImportService, 'StockLedger'), 'zero-stock bootstrap must not append a stock ledger movement'); + $assertTrue( +- str_contains($medicineImportService, 'array_diff($medicineIds, $mappedMedicineIds)'), +- 'bootstrap preflight must reject only medicines without a source mapping and allow deterministic continuation' ++ !str_contains($medicineImportService, 'assertCatalogBelongsToBootstrap') ++ && !str_contains($medicineImportService, 'unrelated existing catalog rows'), ++ 'incremental medicine import must preserve EJ medicines that were created by other sources' + ); + + $catalogLockPath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/MedicineCatalogLock.php'; diff --git a/artifacts/ej-medicine-incremental-sync/MODIFIED_FILE b/artifacts/ej-medicine-incremental-sync/MODIFIED_FILE new file mode 100644 index 000000000..836722d81 --- /dev/null +++ b/artifacts/ej-medicine-incremental-sync/MODIFIED_FILE @@ -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/artifacts/ej-medicine-incremental-sync/ROLLBACK.sh b/artifacts/ej-medicine-incremental-sync/ROLLBACK.sh new file mode 100755 index 000000000..ec761238f --- /dev/null +++ b/artifacts/ej-medicine-incremental-sync/ROLLBACK.sh @@ -0,0 +1,22 @@ +#!/bin/sh +set -eu +ZYT_ROOT=${ZYT_ROOT:-/Users/long/Work/zyt} +EJ_ROOT=${EJ_ROOT:-/Users/long/Work/ej} +ZYT_BASE=27fbef9321c67f962e4d73f04a52272887c04f95 +EJ_BASE=a7439cc2e8c37dd6c32c926f0cc39c1a9c9b4b67 + +git -C "$ZYT_ROOT" checkout "$ZYT_BASE" -- \ + server/config/console.php \ + server/tests/pharmacy/route_contracts.php +rm -f \ + "$ZYT_ROOT/docs/ej-pharmacy-incremental-medicine-sync.md" \ + "$ZYT_ROOT/server/app/command/EjPharmacyPushMedicines.php" \ + "$ZYT_ROOT/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php" \ + "$ZYT_ROOT/server/tests/pharmacy/incremental_medicine_push.php" + +git -C "$EJ_ROOT" checkout "$EJ_BASE" -- \ + server/app/common/service/pharmacy/MedicineImportService.php \ + server/tests/pharmacy/route_contracts.php \ + server/tests/pharmacy/medicine_import_mysql_integration.php + +echo 'ROLLBACK_OK: ZYT incremental push removed; EJ unrelated-catalog preflight restored.' diff --git a/artifacts/ej-medicine-incremental-sync/VERIFICATION.txt b/artifacts/ej-medicine-incremental-sync/VERIFICATION.txt new file mode 100644 index 000000000..a311e6542 --- /dev/null +++ b/artifacts/ej-medicine-incremental-sync/VERIFICATION.txt @@ -0,0 +1,98 @@ +OBJECT=ZYT_TO_EJ_INCREMENTAL_MEDICINE_SYNC +RESULT=NON_DESTRUCTIVE_INCREMENTAL_PUSH_IMPLEMENTED_AND_DRY_RUN_VERIFIED +NEXT=DEPLOY_BOTH_BRANCHES_THEN_RUN_APPLY_ONLY_AFTER_EXPLICIT_AUTHORIZATION +BRANCH_ZYT=codex/ej-medicine-incremental-sync +BRANCH_EJ=codex/ej-additive-medicine-import +ZYT_COMMIT=17e9e7b6b +ZYT_PUSH=origin/codex/ej-medicine-incremental-sync +EJ_COMMIT=2637ffa +EJ_PUSH=origin/codex/ej-additive-medicine-import +CHANGED_BRANCH_FIELD=ZYT ej-pharmacy:push-medicines add-only command + EJ medicine-imports unrelated-catalog preservation + +ARTIFACTS: +MODIFIED_FILE=/Users/long/Work/zyt/artifacts/ej-medicine-incremental-sync/MODIFIED_FILE +DIFF_FILE=/Users/long/Work/zyt/artifacts/ej-medicine-incremental-sync/DIFF_FILE +VERIFICATION=/Users/long/Work/zyt/artifacts/ej-medicine-incremental-sync/VERIFICATION.txt +ROLLBACK=/Users/long/Work/zyt/artifacts/ej-medicine-incremental-sync/ROLLBACK.sh + +ORIGINAL: +ZYT_BASE_COMMIT=27fbef9321c67f962e4d73f04a52272887c04f95 +ZYT_CONSOLE_SHA256=c722a3445f5027edcec9bb5be0981252bb6b3c28ba6f27e87364cc6829e7be6c +ZYT_ROUTE_CONTRACTS_SHA256=d17a7d25b1903f5f4bb52740f068b493a6b7842b57adc4f503bf995e43d26d7e +EJ_BASE_COMMIT=a7439cc2e8c37dd6c32c926f0cc39c1a9c9b4b67 +EJ_IMPORT_SERVICE_SHA256=70f2444db461be0773348814849eeec1742c622e8a807d4b50b2c1190f664b29 +EJ_IMPORT_TEST_SHA256=f91eca5aeb6b493a345dfb79e30614356a83695cc0cd37fe2b8f3d58b0c80e6f + +BASELINE_1: +COMMAND=cd /Users/long/Work/zyt/server && php tests/pharmacy/callback_auth_integration.php +INPUT=baseline checkout 27fbef9321c67f962e4d73f04a52272887c04f95 +LITERAL_OUTPUT=zyt pharmacy callback/auth integration tests passed: 57 +EXIT_STATUS=0 + +BASELINE_2: +COMMAND=cd /Users/long/Work/zyt/server && php tests/pharmacy/run.php +INPUT=baseline checkout 27fbef9321c67f962e4d73f04a52272887c04f95 +LITERAL_OUTPUT=Fatal error: Uncaught RuntimeException: tracking correction must append a strict operation log containing old and new logistics values +EXIT_STATUS=255 +BASELINE_STATUS=pre-existing unrelated tracking-log contract failure; identical after this change + +BASELINE_3: +COMMAND=cd /Users/long/Work/ej/server && php tests/pharmacy/run.php +INPUT=baseline checkout a7439cc2e8c37dd6c32c926f0cc39c1a9c9b4b67 +LITERAL_OUTPUT=pharmacy contract tests passed: 67; pharmacy domain contract tests passed: 41; admin pharmacy contracts passed; workflow template contracts passed: 14 +EXIT_STATUS=0 + +MODIFIED_1: +COMMAND=php /Users/long/Work/zyt/server/tests/pharmacy/incremental_medicine_push.php +INPUT=callback fixtures covering dry-run, stable import IDs, HMAC response identity, locks, created/existing responses, inactive mapping preservation, and zero deletes +LITERAL_OUTPUT=incremental EJ medicine push tests passed: 40 +EXIT_STATUS=0 + +MODIFIED_2: +COMMAND=php heredoc harness requiring /Users/long/Work/zyt/server/tests/pharmacy/route_contracts.php +INPUT=modified command/service registration contracts +LITERAL_OUTPUT=route contracts passed: 26 +EXIT_STATUS=0 + +MODIFIED_3: +COMMAND=cd /Users/long/Work/zyt/server && php think ej-pharmacy:push-medicines +INPUT=default dry-run; no --apply +LITERAL_OUTPUT=dry-run source=654 candidates=654 batches=7 mapped=654 unmapped=0 preserved_inactive=0 remote_delete=0 local_delete=0 +EXIT_STATUS=0 +MODIFIED_RESULT=no EJ HTTP write; no ZYT projection write; no delete + +MODIFIED_4: +COMMAND=cd /Users/long/Work/zyt/server && php think ej-pharmacy:push-medicines --apply +INPUT=apply requested without confirmation token +LITERAL_OUTPUT=执行增量写入必须提供 --confirm=INCREMENTAL_NO_DELETE +EXIT_STATUS=1 +MODIFIED_RESULT=write gate stopped before lock, HTTP, or projection mutation + +MODIFIED_5: +COMMAND=cd /Users/long/Work/ej/server && php tests/pharmacy/run.php +INPUT=EJ additive medicine-import service with unrelated catalog rows preserved +LITERAL_OUTPUT=pharmacy contract tests passed: 67; pharmacy domain contract tests passed: 41; admin pharmacy contracts passed; workflow template contracts passed: 14 +EXIT_STATUS=0 + +MODIFIED_6: +COMMAND=cd /Users/long/Work/ej/server && php tests/pharmacy/medicine_import_mysql_integration.php +INPUT=isolated temporary database containing an unrelated EJ medicine plus a new ZYT source medicine +LITERAL_OUTPUT=medicine import MySQL integration passed: 11 (temporary_database) +EXIT_STATUS=0 +MODIFIED_RESULT=unrelated EJ medicine preserved byte-for-byte; one new medicine/source mapping appended; temporary database cleaned + +MODIFIED_7: +COMMAND=php -l on ZYT service, ZYT command, ZYT console config, and EJ MedicineImportService +INPUT=all changed PHP runtime files +LITERAL_OUTPUT=No syntax errors detected +EXIT_STATUS=0 + +ROLLBACK: +COMMAND=ZYT_ROOT=/tmp/ej-sync-rollback.HWTG7D/zyt EJ_ROOT=/tmp/ej-sync-rollback.HWTG7D/ej /Users/long/Work/zyt/artifacts/ej-medicine-incremental-sync/ROLLBACK.sh +INPUT=detached copies at ZYT 27fbef932 and EJ a7439cc with both diffs applied; BEFORE_ZYT=6; BEFORE_EJ=3 +LITERAL_OUTPUT=ROLLBACK_OK: ZYT incremental push removed; EJ unrelated-catalog preflight restored. +EXIT_STATUS=0 +ROLLBACK_RESULT=AFTER_ZYT=0; AFTER_EJ=0; both copies restored to clean base behavior/status + +CURRENT_STATUS=both feature commits are pushed; ZYT artifact commit pending; EJ retains only pre-existing unrelated deletions; production apply was not run +RESTORED_BEHAVIOR=ROLLBACK removes the ZYT push command/service/docs/tests and restores EJ bootstrap-only unrelated-catalog rejection