<?php
namespace EasyChat\Library\Messenger\Marketplace;

use EasyChat\Library\AppClock;
use EasyChat\Library\Messenger\InboundService;
use EasyChat\Library\Messenger\MessengerLogger;
use EasyChat\Library\Messenger\ProviderRegistry;
use EasyChat\Library\Messenger\WildberriesProvider;
use EasyChat\Library\Messenger\WildberriesTokenInspector;
use EasyChat\Library\Queue\QueueManager;
use EasyChat\Model\Marketplace as MarketplaceModel;
use EasyChat\Model\Messenger as MessengerModel;

 
class MarketplaceSyncService {
    private const BUYER_CHAT_PRIORITY = 8;
    private const FEEDBACK_PRIORITY = 4;
    private const ATTACHMENT_RETRY_PRIORITY = 3;

    private $registry;

    public function __construct($registry) {
        $this->registry = $registry;
    }

    public function enqueueDueStreams(int $limit = 20): array {
        $model = new MarketplaceModel($this->registry);
        $queue = new QueueManager($this->registry);
        $ids = [];
        foreach ($model->dueStreams($limit) as $stream) {
            $stream_id = (int)($stream['stream_id'] ?? 0);
            if ($stream_id <= 0) {
                continue;
            }
            $ids[] = $queue->pushUnique(
                'messenger_marketplace_sync',
                ['stream_id' => $stream_id],
                'marketplace_stream_' . $stream_id,
                0,
                (string)($stream['stream'] ?? '') === MarketplaceStream::BUYER_CHAT
                    ? self::BUYER_CHAT_PRIORITY
                    : self::FEEDBACK_PRIORITY
            );
        }
        return ['ran' => true, 'queued' => count($ids), 'job_ids' => $ids];
    }

    public function syncStreamId(int $stream_id): array {
        $model = new MarketplaceModel($this->registry);
        $row = $model->getStreamById($stream_id);
        if (!$row || (string)($row['status'] ?? '') !== 'enabled' || (string)($row['channel_status'] ?? '') !== 'active') {
            return ['success' => true, 'skipped' => 'stream_inactive'];
        }

        $lease = $model->acquireStreamLease($stream_id, 300);
        if ($lease === '') {
            return ['success' => true, 'skipped' => 'stream_leased'];
        }

        $provider_code = strtolower((string)($row['provider'] ?? ''));
        $logger = new MessengerLogger($this->registry, $provider_code);
        $provider_retry_after = 0;
        try {
            $provider = ProviderRegistry::get($provider_code, $this->registry);
            if (!$provider instanceof MarketplaceProviderInterface) {
                throw new \RuntimeException('marketplace_provider_unavailable');
            }
            $capabilities = $provider->marketplaceCapabilities();
            $channel_capabilities = json_decode((string)($row['capabilities_json'] ?? ''), true);
            $channel_capabilities = is_array($channel_capabilities) ? $channel_capabilities : [];
            $stored = json_decode((string)($row['credentials_json'] ?? ''), true);
            $stored = is_array($stored) ? $stored : [];
            $credentials = (new MarketplaceCredentialsVault($this->registry))->open(
                $stored,
                is_array($capabilities['secret_fields'] ?? null) ? $capabilities['secret_fields'] : []
            );
            $channel_capabilities = $this->refreshWildberriesTokenMetadata(
                $row,
                $credentials,
                $channel_capabilities
            );
            $rate_budget = new MarketplaceRateBudget($this->registry, $model);
            $reservation = $rate_budget->reserve($row, $credentials);
            if (empty($reservation['allowed'])) {
                $delay = max(5, min(21600, (int)($reservation['delay_seconds'] ?? 60)));
                $model->deferStreamWithoutAttempt($stream_id, $lease, $delay);
                return [
                    'success' => true,
                    'skipped' => 'marketplace_rate_budget',
                    'continuation_delay_seconds' => $delay,
                ];
            }
            $cursor = $this->decodedCursor($row);
             
             
             
            $provider_cursor = $cursor;
            $provider_cursor['notification_hints'] = $model->listNotificationHints($stream_id, 20);
            $result = $provider->syncMarketplaceStream((string)$row['stream'], $credentials, $provider_cursor, 50);
            if (empty($result['success'])) {
                $provider_retry_after = max(0, min(21600, (int)($result['retry_after_seconds'] ?? 0)));
                $rate_budget->deferFromProvider($row, $credentials, $provider_retry_after);
                throw new \RuntimeException((string)($result['error'] ?? 'marketplace_sync_failed'));
            }
            $this->renewLeaseOrFail($model, $stream_id, $lease);
            $last_lease_renewed_at = microtime(true);

            $events = is_array($result['events'] ?? null) ? $result['events'] : [];
            if (count($events) > 250) {
                throw new \RuntimeException('marketplace_event_batch_too_large');
            }
            $inbound = new InboundService($this->registry);
            $inbound_channel = $this->channelForInbound($row);
            $handled = 0;
            foreach ($events as $event) {
                if ((microtime(true) - $last_lease_renewed_at) >= 20.0) {
                    $this->renewLeaseOrFail($model, $stream_id, $lease);
                    $last_lease_renewed_at = microtime(true);
                }
                if (!is_array($event) || !$this->validEvent($event, (string)$row['stream'])) {
                    throw new \RuntimeException('marketplace_event_contract_invalid');
                }
                $event = $this->applyChannelPermissions(
                    $event,
                    $channel_capabilities,
                    (string)$row['stream']
                );
                $event_metadata = is_array($event['metadata'] ?? null) ? $event['metadata'] : [];
                $thread_snapshot = [
                    'provider_payload' => is_array($event['snapshot'] ?? null) ? $event['snapshot'] : [],
                    'item_context' => is_array($event['item_context'] ?? null) ? $event['item_context'] : [],
                    'workflow_actions' => is_array($event['workflow_actions'] ?? null) ? $event['workflow_actions'] : [],
                ];
                if (!empty($event_metadata['thread_only'])) {
                    $existing_thread = $model->getThread(
                        (int)$row['channel_id'],
                        (string)$event['conversation_kind'],
                        (string)$event['external_thread_id']
                    );
                    $existing_snapshot = json_decode((string)($existing_thread['snapshot_json'] ?? ''), true);
                    if (is_array($existing_snapshot)) {
                         
                         
                         
                        $previous_payload = is_array($existing_snapshot['provider_payload'] ?? null)
                            ? $existing_snapshot['provider_payload']
                            : [];
                        $incoming_payload = $thread_snapshot['provider_payload'];
                        $thread_snapshot = $existing_snapshot;
                        $thread_snapshot['provider_payload'] = array_replace_recursive(
                            $previous_payload,
                            $incoming_payload
                        );
                    }
                }
                $reply_secret = trim((string)($event['reply_secret'] ?? ''));
                $thread_data = [
                    'channel_id' => (int)$row['channel_id'],
                    'provider' => $provider_code,
                    'conversation_kind' => (string)$event['conversation_kind'],
                    'external_thread_id' => (string)$event['external_thread_id'],
                    'external_chat_id' => (string)($event['external_chat_id'] ?? $event['external_thread_id']),
                    'state' => (string)($event['thread_state'] ?? 'unknown'),
                    'reply_secret' => $reply_secret !== ''
                        ? (new MarketplaceCredentialsVault($this->registry))->sealThreadSecret($reply_secret)
                        : '',
                    'item_id' => (string)($event['item_context']['item_id'] ?? ''),
                    'order_id' => (string)($event['order_id'] ?? ''),
                    'capabilities' => is_array($event['capabilities'] ?? null) ? $event['capabilities'] : [],
                    'snapshot' => $thread_snapshot,
                    'last_external_at' => (new AppClock($this->registry))->fromProviderTimestampSql(
                        0,
                        (string)($event['published_at'] ?? '')
                    ),
                ];
                unset($event['reply_secret']);
                if (!empty($event_metadata['thread_only'])) {
                    $thread = $model->upsertThread($thread_data);
                    if (empty($thread['thread_id'])) {
                        throw new \RuntimeException('marketplace_thread_persistence_failed');
                    }
                     
                     
                     
                    $this->reconcileConversationStart($model, $row, $event, $thread, 0);
                    $handled++;
                    continue;
                }
                $accepted = $inbound->acceptMarketplaceEvent($inbound_channel, $event);
                if (empty($accepted['success'])) {
                    throw new \RuntimeException((string)($accepted['error'] ?? 'marketplace_inbound_failed'));
                }
                if (!$this->acceptedEventIsMaterialized($accepted)) {
                     
                     
                     
                     
                    throw new \RuntimeException('marketplace_event_not_materialized');
                }
                $thread = $model->upsertThread($thread_data);
                if (empty($thread['thread_id'])) {
                     
                     
                     
                    throw new \RuntimeException('marketplace_thread_persistence_failed');
                }
                if (!empty($accepted['attachment_pending']) && !empty($event['attachments'])) {
                    $this->enqueueAttachmentRetry((int)$row['channel_id'], $event);
                }
                if (!empty($thread['thread_id']) && !empty($accepted['chat_id'])) {
                    $model->linkThreadToChat((int)$thread['thread_id'], (int)$accepted['chat_id']);
                }
                $this->reconcileConversationStart(
                    $model,
                    $row,
                    $event,
                    $thread,
                    (int)($accepted['chat_id'] ?? 0)
                );
                $handled++;
            }

            $this->renewLeaseOrFail($model, $stream_id, $lease);
            $model->deleteNotificationHints((array)($result['consumed_hint_ids'] ?? []));
            $delay = max(5, (int)($result['delay_seconds'] ?? 30));
            $model->finishStreamSuccess(
                $stream_id,
                $lease,
                is_array($result['cursor'] ?? null) ? $result['cursor'] : $cursor,
                !empty($result['has_more']) ? min($delay, 5) : $delay
            );
            $logger->info('marketplace_sync_ok', [
                'stream_id' => $stream_id,
                'stream' => (string)$row['stream'],
                'event_count' => $handled,
                'has_more' => !empty($result['has_more']) ? 1 : 0,
            ]);
            return [
                'success' => true,
                'handled' => $handled,
                'has_more' => !empty($result['has_more']),
                'continuation_delay_seconds' => !empty($result['has_more']) ? min($delay, 5) : 0,
            ];
        } catch (\Throwable $e) {
            $error = preg_replace('/[^a-z0-9_:\-]/i', '_', MarketplaceText::truncate($e->getMessage(), 160));
            $failure_count = max(0, (int)($row['failure_count'] ?? 0));
            $delay = min(21600, 30 * (int)pow(2, min(9, $failure_count)));
            if ($provider_retry_after > 0) {
                 
                 
                 
                $delay = max($delay, min(21600, $provider_retry_after + 1 + ($stream_id % 5)));
            }
            $model->finishStreamFailure($stream_id, $lease, $error, $delay);
            $logger->error('marketplace_sync_failed', [
                'stream_id' => $stream_id,
                'stream' => (string)($row['stream'] ?? ''),
                'error' => $error,
                'retry_after_seconds' => $delay,
            ]);
             
             
             
            return ['success' => false, 'error' => $error, 'retry_after_seconds' => $delay];
        }
    }

    




    private function channelForInbound(array $stream_row): array {
        $channel = $stream_row;
        $channel['status'] = (string)($stream_row['channel_status'] ?? '');
        return $channel;
    }

    private function validEvent(array $event, string $expected_stream): bool {
        return MarketplaceStream::normalize((string)($event['conversation_kind'] ?? '')) === MarketplaceStream::normalize($expected_stream)
            && trim((string)($event['external_thread_id'] ?? '')) !== ''
            && trim((string)($event['external_user_id'] ?? '')) !== ''
            && trim((string)($event['dedupe_key'] ?? $event['external_message_id'] ?? '')) !== ''
            && in_array((string)($event['direction'] ?? ''), ['in', 'out'], true);
    }

    private function acceptedEventIsMaterialized(array $accepted): bool {
        if ((int)($accepted['chat_id'] ?? 0) > 0 && (int)($accepted['message_id'] ?? 0) > 0) {
            return true;
        }
         
         
         
        return in_array((string)($accepted['ignored'] ?? ''), ['chat_blocked', 'empty_platform_message'], true);
    }

    private function renewLeaseOrFail(MarketplaceModel $model, int $stream_id, string $lease): void {
        if (!$model->renewStreamLease($stream_id, $lease, 300)) {
            throw new \RuntimeException('marketplace_stream_lease_lost');
        }
    }

    private function decodedCursor(array $row): array {
        $cursor = json_decode((string)($row['cursor_json'] ?? ''), true);
        return is_array($cursor) ? $cursor : [];
    }

    





    private function refreshWildberriesTokenMetadata(
        array $row,
        array $credentials,
        array $capabilities
    ): array {
        if ((string)($row['provider'] ?? '') !== WildberriesProvider::PROVIDER) {
            return $capabilities;
        }
        $inspection = WildberriesTokenInspector::inspect((string)($credentials['api_token'] ?? ''));
        if (empty($inspection['success'])) {
            return $capabilities;
        }
        $public = WildberriesTokenInspector::publicMetadata($inspection);
        $mapping = [
            'token_type' => 'token_type',
            'token_type_label' => 'token_type_label',
            'expires_at' => 'token_expires_at',
            'read_only' => 'token_read_only',
            'permissions' => 'token_permissions',
        ];
        $changed = false;
        foreach ($mapping as $public_key => $capability_key) {
            $value = $public[$public_key] ?? null;
            if (($capabilities[$capability_key] ?? null) !== $value) {
                $capabilities[$capability_key] = $value;
                $changed = true;
            }
        }
        if ($changed) {
            $json = json_encode($capabilities, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
            if (is_string($json)) {
                (new MessengerModel($this->registry))->updateChannel(
                    (int)($row['channel_id'] ?? 0),
                    ['capabilities_json' => $json]
                );
            }
        }
        return $capabilities;
    }

    




    private function reconcileConversationStart(
        MarketplaceModel $model,
        array $channel,
        array $event,
        array $thread,
        int $accepted_chat_id
    ): void {
        $provider = (string)($channel['provider'] ?? '');
        if (!in_array($provider, ['ozon', 'yandex_market'], true)
            || (string)($event['conversation_kind'] ?? '') !== MarketplaceStream::BUYER_CHAT) {
            return;
        }
        $resource_id = trim((string)($event['order_id'] ?? $thread['order_id'] ?? ''));
        if ($resource_id === '') {
            return;
        }
        $metadata = is_array($event['metadata'] ?? null) ? $event['metadata'] : [];
        if ($provider === 'yandex_market') {
            $context_type = strtolower((string)($metadata['context_type'] ?? ''));
            if (!in_array($context_type, ['order', 'return'], true)) return;
            $resource = $provider . ':' . $context_type . ':' . $resource_id;
        } else {
            $resource = $provider . ':posting:' . $resource_id;
        }
        $chat_id = $accepted_chat_id > 0 ? $accepted_chat_id : (int)($thread['chat_id'] ?? 0);
        $external_chat_id = (string)($thread['external_chat_id'] ?? $event['external_chat_id'] ?? '');
        $provider_title = $provider === 'yandex_market' ? 'Яндекс Маркет' : 'Ozon';
        $model->reconcileConversationStart((int)($channel['channel_id'] ?? 0), $resource, [
            'success' => true,
            'chat_id' => $chat_id,
            'external_chat_id' => $external_chat_id,
            'external_message_id' => (string)($event['external_message_id'] ?? ''),
            'sync_pending' => $chat_id <= 0,
            'message' => $chat_id > 0
                ? 'Диалог ' . $provider_title . ' найден. Перед отправкой проверьте его историю.'
                : $provider_title . ' подтвердил диалог. EasyChat покажет его после ближайшей сверки.',
        ]);
    }

    



    private function applyChannelPermissions(array $event, array $channel_capabilities, string $stream): array {
        $stream = MarketplaceStream::normalize($stream);
        $event_capabilities = is_array($event['capabilities'] ?? null) ? $event['capabilities'] : [];
        $stream_access = is_array($channel_capabilities['stream_access'][$stream] ?? null)
            ? $channel_capabilities['stream_access'][$stream]
            : [];
        $action_access = is_array($stream_access['action_access'] ?? null)
            ? $stream_access['action_access']
            : [];
        $read_only = !empty($channel_capabilities['token_read_only']);

        $allowed_actions = [];
        foreach ((array)($event_capabilities['actions'] ?? []) as $action) {
            $action = preg_replace('/[^a-z0-9_\-]/i', '', (string)$action);
            if ($action === '' || $read_only) continue;
            if (array_key_exists($action, $action_access) && $action_access[$action] === false) continue;
            $allowed_actions[] = $action;
        }
        $event_capabilities['actions'] = array_values(array_unique($allowed_actions));

        if ($stream === MarketplaceStream::BUYER_CHAT) {
            $reply_allowed = !$read_only && !empty($event_capabilities['can_reply']);
            if (array_key_exists('buyer_chat_reply', $action_access) && $action_access['buyer_chat_reply'] === false) {
                $reply_allowed = false;
            }
            $attachments_allowed = $reply_allowed;
            if (array_key_exists('buyer_chat_attachment', $action_access)
                && $action_access['buyer_chat_attachment'] === false) {
                $attachments_allowed = false;
            }
            $event_capabilities['can_reply'] = $reply_allowed;
            $event_capabilities['read_only'] = !$reply_allowed;
            $event_capabilities['attachments_out'] = $attachments_allowed;
        }

        if (!in_array('return_action', $event_capabilities['actions'], true)) {
            $event_capabilities['workflow_actions'] = [];
            $event_capabilities['allowed_claim_actions'] = [];
            $event_capabilities['return_rejection_reasons'] = [];
            $event['workflow_actions'] = [];
        }
        $event['capabilities'] = $event_capabilities;
        return $event;
    }

    private function enqueueAttachmentRetry(int $channel_id, array $event): void {
        $dedupe = trim((string)($event['dedupe_key'] ?? $event['external_message_id'] ?? ''));
        if ($channel_id <= 0 || $dedupe === '') {
            return;
        }
         
         
        $event = array_intersect_key($event, array_flip([
            'update_type', 'text', 'external_user_id', 'external_chat_id',
            'external_message_id', 'dedupe_key', 'direction', 'display_name',
            'attachments', 'conversation_kind', 'external_thread_id',
            'published_at', 'metadata', 'item_context',
        ]));
        $encoded = json_encode($event, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        if (!is_string($encoded) || strlen($encoded) > 48 * 1024) {
            (new MessengerLogger($this->registry, 'marketplace'))->warning('marketplace_attachment_retry_payload_too_large', [
                'channel_id' => $channel_id,
                'message_hash' => MessengerLogger::hashExternalId($dedupe),
            ]);
            return;
        }
        (new QueueManager($this->registry))->pushUnique(
            'messenger_marketplace_attachment_retry',
            ['channel_id' => $channel_id, 'event' => $event, 'retry_count' => 0],
            'marketplace_attachment_' . substr(hash('sha256', $channel_id . ':' . $dedupe), 0, 48),
            15,
             
             
             
            self::ATTACHMENT_RETRY_PRIORITY
        );
    }
}
