docs: record EJ incremental sync verification
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,590 @@
|
|||||||
|
<?php
|
||||||
|
|
||||||
|
declare(strict_types=1);
|
||||||
|
|
||||||
|
namespace app\common\service\pharmacy;
|
||||||
|
|
||||||
|
use DomainException;
|
||||||
|
use InvalidArgumentException;
|
||||||
|
use RuntimeException;
|
||||||
|
use think\facade\Db;
|
||||||
|
|
||||||
|
final class EjMedicineIncrementalPushService
|
||||||
|
{
|
||||||
|
private const SOURCE_SYSTEM = 'zyt';
|
||||||
|
private const APPLY_CONFIRMATION = 'INCREMENTAL_NO_DELETE';
|
||||||
|
private const DEFAULT_RUN_ID = 'zyt-incremental';
|
||||||
|
private const STATE_ID = 1;
|
||||||
|
private const LOCK_TTL = 120;
|
||||||
|
|
||||||
|
public static function assertCommandGate(bool $apply, string $confirm, int $batchSize): void
|
||||||
|
{
|
||||||
|
self::assertBatchSize($batchSize);
|
||||||
|
if ($apply && !hash_equals(self::APPLY_CONFIRMATION, $confirm)) {
|
||||||
|
throw new InvalidArgumentException(
|
||||||
|
'执行增量写入必须提供 --confirm=' . self::APPLY_CONFIRMATION
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param null|callable():array<int,array<string,mixed>> $sourceLoader
|
||||||
|
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $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<int,array<string,mixed>> $sourceLoader
|
||||||
|
* @param null|callable(array<string,mixed>):array<string,mixed> $importer
|
||||||
|
* @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
|
||||||
|
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
|
||||||
|
* @param null|callable(string):bool $lockAcquirer
|
||||||
|
* @param null|callable(string):bool $lockRenewer
|
||||||
|
* @param null|callable(string):void $lockReleaser
|
||||||
|
* @return array<string,int|string>
|
||||||
|
*/
|
||||||
|
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<int,array<string,mixed>> $sourceLoader
|
||||||
|
* @param null|callable(array<string,mixed>):array<string,mixed> $importer
|
||||||
|
* @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
|
||||||
|
* @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
|
||||||
|
* @param callable(string):bool $lockRenewer
|
||||||
|
* @return array<string,int|string>
|
||||||
|
*/
|
||||||
|
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<array<string,mixed>> $items
|
||||||
|
* @return list<array{source_system:string,import_id:string,items:array<int,array<string,mixed>>}>
|
||||||
|
*/
|
||||||
|
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<int,array<string,mixed>> */
|
||||||
|
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<int,int> $localIds @return array<int,array<string,mixed>> */
|
||||||
|
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<array<string,mixed>> $items
|
||||||
|
* @param array<int,array<string,mixed>> $mappingRows
|
||||||
|
* @return array{candidates:list<array<string,mixed>>,mapped_count:int,unmapped_count:int,preserved_inactive:int,active_mapping_codes:array<string,string>}
|
||||||
|
*/
|
||||||
|
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<string,mixed> $response
|
||||||
|
* @param array<string,mixed> $payload
|
||||||
|
* @param array<string,bool> $seenCodes
|
||||||
|
* @param array<int,bool> $seenVersions
|
||||||
|
* @return list<array{source_medicine_id:string,medicine_code:string,catalog_version:int,action:string}>
|
||||||
|
*/
|
||||||
|
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<int,array<string,mixed>> $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;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
+22
@@ -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.'
|
||||||
@@ -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
|
||||||
Reference in New Issue
Block a user