<?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;
        });
    }
}