<?php

declare(strict_types=1);

namespace PhpWA\Client;

use InvalidArgumentException;
use PhpWA\Exception\ProtocolException;
use PhpWA\Integration\EventType;
use PhpWA\Media\MediaDownloader;
use PhpWA\Media\MediaType;
use PhpWA\Protocol\Compression\BoundedZlibInflater;
use PhpWA\Protocol\Jid\Jid;
use PhpWA\Protocol\Proto\Conversation;
use PhpWA\Protocol\Proto\HistorySync;
use PhpWA\Protocol\Proto\HistorySyncMsg;
use PhpWA\Protocol\Proto\WebMessageInfo;
use PhpWA\Runtime\Deadline;
use PhpWA\Storage\AccountStoreInterface;
use PhpWA\Storage\AccountTransactionInterface;
use PhpWA\Storage\PrivateFileArea;
use PhpWA\Storage\StoredValue;

/** Resumable, bounded converter from WhatsApp history chunks to outbox events. */
final class HistorySyncWorker
{
    /** @var string */ private $accountId;
    /** @var AccountStoreInterface */ private $accounts;
    /** @var PrivateFileArea */ private $files;
    /** @var MediaDownloader|null */ private $downloader;
    /** @var BoundedZlibInflater */ private $inflater;
    /** @var int */ private $maximumCompressedBytes;
    /** @var MessageProjector */ private $projector;

    public function __construct(
        string $accountId,
        AccountStoreInterface $accounts,
        string $workingDirectory,
        ?MediaDownloader $downloader = null,
        ?BoundedZlibInflater $inflater = null,
        int $maximumCompressedBytes = 67108864,
        ?MessageProjector $projector = null
    ) {
        if ($accountId === '' || strlen($accountId) > 512 || preg_match('//u', $accountId) !== 1
            || preg_match('/[\x00-\x1f\x7f]/', $accountId) === 1
            || $maximumCompressedBytes < 1 || $maximumCompressedBytes > 268435456) {
            throw new InvalidArgumentException('History-sync worker configuration is invalid.');
        }
        $this->accountId = $accountId; $this->accounts = $accounts;
        $this->files = new PrivateFileArea($workingDirectory); $this->downloader = $downloader;
        $this->inflater = $inflater ?? new BoundedZlibInflater($maximumCompressedBytes, 134217728);
        $this->maximumCompressedBytes = $maximumCompressedBytes;
        $this->projector = $projector ?? new MessageProjector();
    }

    public function run(Deadline $deadline, int $maximumEvents = 50): HistorySyncRunResult
    {
        if ($maximumEvents < 1 || $maximumEvents > 100) {
            throw new InvalidArgumentException('History-sync event budget must be between 1 and 100.');
        }
        $snapshot = $this->accounts->transaction($this->accountId, static function (AccountTransactionInterface $tx): ?array {
            $jobs = $tx->list('history_job', 1);
            if ($jobs === []) { return null; }
            $key = (string) array_key_first($jobs);
            return ['key' => $key, 'stored' => $jobs[$key]];
        });
        if ($snapshot === null) { return HistorySyncRunResult::idle(); }
        /** @var StoredValue $stored */
        $stored = $snapshot['stored'];
        $job = HistorySyncJob::decode($stored->value());
        if (!hash_equals($snapshot['key'], $job->key())) {
            throw new ProtocolException('History-sync job is stored under an invalid key.');
        }

        $temporary = null;
        try {
            try {
                $history = $this->loadHistory($job, $deadline, $temporary);
            } catch (ProtocolException $error) {
                if (!$this->isRemoteHistoryUnavailable($error, $job)) {
                    throw $error;
                }
                // История после pairing полезна, но не должна блокировать
                // live-сессию навсегда, когда WhatsApp больше не отдаёт
                // конкретный архив. Удаляем только неизменённый job.
                $this->discardUnavailableJob($snapshot, $stored);
                return new HistorySyncRunResult(true, true, 0, 0);
            }
            if (is_string($temporary)) {
                // The parsed protobuf owns the data now. Require plaintext
                // cleanup before any cursor/outbox commit can report success.
                $this->files->delete($temporary);
                $temporary = null;
            }
            $mappings = $this->collectMappings($history);
            $mappings = $this->excludePinnedConflicts($mappings);
            [$events, $nextCursor, $complete] = $this->selectEvents($history, $job, $mappings, $maximumEvents, $deadline);
            $staged = $this->accounts->transaction(
                $this->accountId,
                function (AccountTransactionInterface $tx) use ($snapshot, $stored, $job, $mappings, $events, $nextCursor, $complete): int {
                    $current = $tx->get('history_job', $snapshot['key']);
                    if ($current === null || $current->revision() !== $stored->revision()
                        || !hash_equals($current->value(), $stored->value())) {
                        throw new ProtocolException('History-sync job changed while it was being processed.');
                    }
                    $this->pinMappings($tx, $mappings);
                    $staged = 0;
                    foreach ($events as $event) {
                        if ($tx->stageIncoming($event['dedupe_id'], $event['type'], $event['payload'], $job->receivedAt())) {
                            $staged++;
                        }
                    }
                    if ($complete) {
                        $tx->delete('history_job', $snapshot['key'], $stored->revision());
                    } else {
                        $tx->put('history_job', $snapshot['key'], $job->withCursor($nextCursor)->encode(), $stored->revision());
                    }
                    return $staged;
                }
            );
            return new HistorySyncRunResult(true, $complete, $staged, $nextCursor);
        } finally {
            if (is_string($temporary)) {
                try { $this->files->delete($temporary); } catch (\Throwable $ignored) {}
            }
        }
    }

    private function loadHistory(HistorySyncJob $job, Deadline $deadline, ?string &$temporary): HistorySync
    {
        $notification = $job->notification();
        $inline = $notification->getInitialHistBootstrapInlinePayload();
        if ($inline !== '') {
            $raw = $this->inflater->inflateString($inline, $deadline);
        } else {
            if (!$this->downloader instanceof MediaDownloader) {
                throw new ProtocolException('A media downloader is required for a remote history-sync chunk.');
            }
            $length = $this->boundedInteger($notification->getFileLength(), 'history-sync file length', $this->maximumCompressedBytes);
            if (strlen($notification->getMediaKey()) !== 32 || strlen($notification->getFileSha256()) !== 32
                || strlen($notification->getFileEncSha256()) !== 32) {
                throw new ProtocolException('Remote history-sync media metadata is incomplete.');
            }
            $temporary = $this->files->accountPath('history', $this->accountId . ':history:' . $job->key(), '.zlib');
            $this->files->delete($temporary);
            $this->downloader->download(
                $notification->getDirectPath(), MediaType::HISTORY, $notification->getMediaKey(),
                $notification->getFileSha256(), $notification->getFileEncSha256(), $length,
                $temporary, $deadline
            );
            $raw = $this->inflater->inflateFile($temporary, $deadline);
        }
        $history = new HistorySync();
        try { $history->mergeFromString($raw); }
        catch (\Throwable $error) { throw new ProtocolException('Inflated history-sync protobuf is malformed.', 0, $error); }
        return $history;
    }

    /** Remote bootstrap archives are optional; malformed inline history is not. */
    private function isRemoteHistoryUnavailable(ProtocolException $error, HistorySyncJob $job): bool
    {
        return $job->notification()->getInitialHistBootstrapInlinePayload() === ''
            && hash_equals('All WhatsApp resumable media download hosts failed.', $error->getMessage());
    }

    /** @param array{key:string,stored:StoredValue} $snapshot */
    private function discardUnavailableJob(array $snapshot, StoredValue $stored): void
    {
        $this->accounts->transaction($this->accountId, static function (AccountTransactionInterface $tx) use ($snapshot, $stored): void {
            $current = $tx->get('history_job', $snapshot['key']);
            if ($current !== null && $current->revision() === $stored->revision()
                && hash_equals($current->value(), $stored->value())) {
                $tx->delete('history_job', $snapshot['key'], $stored->revision());
            }
        });
    }

    /** @return array<string,string> PN JID => LID JID */
    private function collectMappings(HistorySync $history): array
    {
        $byPhone = []; $byLid = []; $candidateCount = 0;
        foreach ($history->getPhoneNumberToLidMappings() as $mapping) {
            $this->addMappingCandidate($byPhone, $byLid, $candidateCount, $mapping->getPnJid(), $mapping->getLidJid());
        }
        foreach ($history->getConversations() as $conversation) {
            $id = $conversation->getId(); $pn = $conversation->getPnJid(); $lid = $conversation->getLidJid();
            $parsed = Jid::parse($id);
            if ($parsed instanceof Jid && ($parsed->isLid() || $parsed->isHostedLid()) && $pn !== '') {
                $this->addMappingCandidate($byPhone, $byLid, $candidateCount, $pn, $id);
            } elseif ($parsed instanceof Jid && ($parsed->isPhoneNumber() || $parsed->isHostedPhoneNumber()) && $lid !== '') {
                $this->addMappingCandidate($byPhone, $byLid, $candidateCount, $id, $lid);
            }
        }
        $mappings = [];
        foreach ($byPhone as $phone => $lids) {
            if (count($lids) !== 1) { continue; }
            $lid = (string) array_key_first($lids);
            // Не угадываем личность при спорных данных WhatsApp. Историю можно
            // принять, но неоднозначная связь PN/LID не покидает этот слой.
            if (count($byLid[$lid] ?? []) !== 1) { continue; }
            $mappings[$phone] = $lid;
        }
        return $mappings;
    }

    /** @param array<string,array<string,true>> $byPhone @param array<string,array<string,true>> $byLid */
    private function addMappingCandidate(array &$byPhone, array &$byLid, int &$candidateCount, string $phoneText, string $lidText): void
    {
        $phone = Jid::parse($phoneText); $lid = Jid::parse($lidText);
        if (!$phone instanceof Jid || (!$phone->isPhoneNumber() && !$phone->isHostedPhoneNumber())
            || !$lid instanceof Jid || (!$lid->isLid() && !$lid->isHostedLid())
            || $phone->device() !== null || $lid->device() !== null) {
            throw new ProtocolException('History-sync chunk contains an invalid phone-to-LID mapping.');
        }
        $pn = (string) $phone; $li = (string) $lid;
        $candidateCount++;
        if ($candidateCount > 10000) { throw new ProtocolException('History-sync chunk contains too many LID mappings.'); }
        $byPhone[$pn][$li] = true;
        $byLid[$li][$pn] = true;
    }

    /** @param array<string,string> $mappings */
    private function pinMappings(AccountTransactionInterface $tx, array $mappings): void
    {
        foreach ($mappings as $phone => $lid) {
            $this->pin($tx, 'pn_mapping', $phone, $lid);
            $this->pin($tx, 'lid_mapping', $lid, $phone);
        }
    }

    /** @param array<string,string> $mappings @return array<string,string> */
    private function excludePinnedConflicts(array $mappings): array
    {
        return $this->accounts->transaction($this->accountId, static function (AccountTransactionInterface $tx) use ($mappings): array {
            $trusted = [];
            foreach ($mappings as $phone => $lid) {
                $knownLid = $tx->get('pn_mapping', $phone);
                $knownPhone = $tx->get('lid_mapping', $lid);
                if (($knownLid !== null && !hash_equals($knownLid->value(), $lid))
                    || ($knownPhone !== null && !hash_equals($knownPhone->value(), $phone))) {
                    continue;
                }
                $trusted[$phone] = $lid;
            }
            return $trusted;
        });
    }

    private function pin(AccountTransactionInterface $tx, string $collection, string $key, string $value): void
    {
        $known = $tx->get($collection, $key);
        if ($known === null) { $tx->put($collection, $key, $value, 0); return; }
        if (!hash_equals($known->value(), $value)) {
            throw new ProtocolException('History-sync LID mapping conflicts with pinned account state.');
        }
    }

    /** @return array{0:list<array{dedupe_id:string,type:string,payload:string}>,1:int,2:bool} */
    private function selectEvents(HistorySync $history, HistorySyncJob $job, array $mappings, int $limit, Deadline $deadline): array
    {
        $events = []; $index = 0; $hasMore = false;
        foreach ($this->events($history, $job, $mappings, $deadline) as $event) {
            if ($index < $job->cursor()) { $index++; continue; }
            if (count($events) >= $limit) { $hasMore = true; break; }
            $event['dedupe_id'] = 'history:' . hash('sha256', $job->sourceMessageId() . "\0" . $index . "\0" . $event['type']);
            $events[] = $event; $index++;
        }
        if (!$hasMore && $job->cursor() > $index) {
            throw new ProtocolException('Stored history-sync cursor exceeds the event stream.');
        }
        return [$events, $job->cursor() + count($events), !$hasMore];
    }

    /** @return \Generator<int,array{type:string,payload:string}> */
    private function events(HistorySync $history, HistorySyncJob $job, array $mappings, Deadline $deadline): \Generator
    {
        $common = [
            'schema_version' => 2, 'source_message_id' => $job->sourceMessageId(),
            'sync_type' => $history->getSyncType(), 'chunk_order' => $history->getChunkOrder(),
            'progress' => $history->getProgress(),
        ];
        foreach ($mappings as $phone => $lid) {
            $deadline->remaining('history sync event conversion');
            yield ['type' => EventType::HISTORY_MAPPING, 'payload' => $this->json($common + [
                'pn_jid' => $phone, 'lid_jid' => $lid,
            ])];
        }
        foreach ($history->getConversations() as $chatIndex => $conversation) {
            $deadline->remaining('history sync event conversion');
            $copy = new Conversation();
            try { $copy->mergeFromString($conversation->serializeToString()); $copy->setMessages([]); }
            catch (\Throwable $error) { throw new ProtocolException('Unable to normalize a history-sync conversation.', 0, $error); }
            [$phone, $lid] = $this->trustedConversationMapping($conversation, $mappings);
            yield ['type' => EventType::HISTORY_CHAT, 'payload' => $this->json($common + [
                'chat_index' => $chatIndex, 'jid' => $conversation->getId(),
                'name' => $conversation->getDisplayName() !== '' ? $conversation->getDisplayName() : $conversation->getName(),
                'username' => $conversation->getUsername(), 'pn_jid' => $phone,
                'lid_jid' => $lid, 'conversation_proto_base64' => base64_encode($copy->serializeToString()),
            ])];
            foreach ($conversation->getMessages() as $messageIndex => $item) {
                if (!$item instanceof HistorySyncMsg || !$item->hasMessage() || $item->getMessage() === null) { continue; }
                $deadline->remaining('history sync event conversion');
                yield ['type' => EventType::HISTORY_MESSAGE, 'payload' => $this->messagePayload(
                    $common + ['chat_index' => $chatIndex, 'message_index' => $messageIndex, 'chat_jid' => $conversation->getId()],
                    $item->getMessage(), (string) $item->getMsgOrderId()
                )];
            }
        }
        foreach ($history->getStatusV3Messages() as $statusIndex => $message) {
            $deadline->remaining('history sync event conversion');
            yield ['type' => EventType::HISTORY_STATUS, 'payload' => $this->messagePayload(
                $common + ['status_index' => $statusIndex], $message, null
            )];
        }
        foreach ($history->getPushnames() as $pushnameIndex => $pushname) {
            $deadline->remaining('history sync event conversion');
            yield ['type' => EventType::HISTORY_PUSHNAME, 'payload' => $this->json($common + [
                'pushname_index' => $pushnameIndex, 'jid' => $pushname->getId(), 'push_name' => $pushname->getPushname(),
            ])];
        }
        yield ['type' => EventType::HISTORY_SYNC, 'payload' => $this->json($common + ['complete' => true])];
    }

    /** @param array<string,string> $mappings @return array{0:string,1:string} */
    private function trustedConversationMapping(Conversation $conversation, array $mappings): array
    {
        $id = Jid::parse($conversation->getId());
        $phone = Jid::parse($conversation->getPnJid());
        $lid = Jid::parse($conversation->getLidJid());
        if ($id instanceof Jid && ($id->isPhoneNumber() || $id->isHostedPhoneNumber())) { $phone = $id; }
        if ($id instanceof Jid && ($id->isLid() || $id->isHostedLid())) { $lid = $id; }
        if (!$phone instanceof Jid || !$lid instanceof Jid) { return ['', '']; }
        $pn = (string) $phone; $li = (string) $lid;
        return isset($mappings[$pn]) && hash_equals($mappings[$pn], $li) ? [$pn, $li] : ['', ''];
    }

    /** @param array<string,mixed> $common */
    private function messagePayload(array $common, WebMessageInfo $message, ?string $orderId): string
    {
        $key = $message->getKey();
        $protocolMessage = $message->getMessage();
        $projection = $protocolMessage instanceof \PhpWA\Protocol\Proto\Message
            ? $this->projector->project($protocolMessage)
            : ['kind' => 'unknown', 'text' => null, 'media' => null, 'details' => null, 'wrappers' => []];
        return $this->json($common + [
            'order_id' => $orderId,
            'id' => $key === null ? '' : $key->getId(),
            'chat_jid' => $key === null ? ($common['chat_jid'] ?? '') : $key->getRemoteJid(),
            'participant_jid' => $key === null ? '' : $key->getParticipant(),
            'from_me' => $key !== null && $key->getFromMe(),
            'timestamp' => (string) $message->getMessageTimestamp(), 'push_name' => $message->getPushName(),
            'kind' => $projection['kind'], 'text' => $projection['text'], 'media' => $projection['media'],
            'details' => $projection['details'], 'wrappers' => $projection['wrappers'],
            'message_proto_base64' => base64_encode($message->serializeToString()),
        ]);
    }

    /** @param array<string,mixed> $payload */
    private function json(array $payload): string
    {
        try { $json = json_encode($payload, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR); }
        catch (\Throwable $error) { throw new ProtocolException('Unable to serialize a history-sync event.', 0, $error); }
        if (!is_string($json) || $json === '' || strlen($json) > 4194304) {
            throw new ProtocolException('History-sync event exceeds the durable outbox limit.');
        }
        return $json;
    }

    /** @param int|string $value */
    private function boundedInteger($value, string $name, int $maximum): int
    {
        if (is_int($value)) { $integer = $value; }
        elseif (is_string($value) && preg_match('/\A(?:0|[1-9][0-9]*)\z/D', $value) === 1
            && strlen($value) <= strlen((string) $maximum) && strcmp(str_pad($value, strlen((string) $maximum), '0', STR_PAD_LEFT), (string) $maximum) <= 0) {
            $integer = (int) $value;
        } else { throw new ProtocolException(ucfirst($name) . ' is invalid.'); }
        if ($integer < 0 || $integer > $maximum) { throw new ProtocolException(ucfirst($name) . ' exceeds its limit.'); }
        return $integer;
    }
}
