<?php
namespace EasyChat\Model;

use EasyChat\Library\Messenger\Marketplace\MarketplaceProviderCatalog;
use EasyChat\Library\Messenger\Marketplace\MarketplaceText;







class Marketplace extends Base {

    public function getStreamById(int $stream_id): array {
        if ($stream_id <= 0) {
            return [];
        }
        $query = $this->db->query(
            "SELECT s.*, c.`provider`, c.`status` AS `channel_status`,
                    c.`credentials_json`, c.`settings_json`, c.`capabilities_json`,
                    c.`external_account_id`, c.`title`
             FROM `{$this->prefix}messenger_channel_streams` s
             INNER JOIN `{$this->prefix}messenger_channels` c ON c.`channel_id` = s.`channel_id`
             WHERE s.`stream_id` = " . (int)$stream_id . " LIMIT 1"
        );
        return $query->row ?: [];
    }

    public function ensureStreams(int $channel_id, array $streams, bool $enabled = true): void {
        if ($channel_id <= 0) {
            return;
        }
        foreach ($streams as $stream) {
            $stream = $this->normalizeStream($stream);
            if ($stream === '') {
                continue;
            }
            $this->db->query(
                "INSERT INTO `{$this->prefix}messenger_channel_streams` SET
                    `channel_id` = {$channel_id},
                    `stream` = '" . $this->db->escape($stream) . "',
                    `status` = '" . ($enabled ? 'enabled' : 'disabled') . "',
                    `cursor_json` = '{}',
                    `next_run_at` = NOW(),
                    `date_added` = NOW(), `date_modified` = NOW()
                 ON DUPLICATE KEY UPDATE `date_modified` = `date_modified`"
            );
        }
    }

    public function listStreams(int $channel_id): array {
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}messenger_channel_streams`
             WHERE `channel_id` = " . (int)$channel_id . "
             ORDER BY FIELD(`stream`, 'buyer_chat', 'question', 'review', 'return_claim'), `stream` ASC"
        );
        return $query->rows ?: [];
    }

     
    public function listAccountStreams(): array {
        $providers = $this->marketplaceProviderSql();
        $query = $this->db->query(
            "SELECT s.*
             FROM `{$this->prefix}messenger_channel_streams` s
             INNER JOIN `{$this->prefix}messenger_channels` c ON c.`channel_id` = s.`channel_id`
             WHERE c.`provider` IN ({$providers})
             ORDER BY c.`provider` ASC, s.`channel_id` ASC,
                      FIELD(s.`stream`, 'buyer_chat', 'question', 'review', 'return_claim'), s.`stream` ASC"
        );
        return $query->rows ?: [];
    }

    public function setStreamStatus(int $channel_id, string $stream, bool $enabled): void {
        $stream = $this->normalizeStream($stream);
        if ($channel_id <= 0 || $stream === '') {
            return;
        }
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `status` = '" . ($enabled ? 'enabled' : 'disabled') . "',
                 `next_run_at` = " . ($enabled ? 'NOW()' : 'NULL') . ",
                 `lease_token` = '', `lease_until` = NULL, `date_modified` = NOW()
             WHERE `channel_id` = {$channel_id}
               AND `stream` = '" . $this->db->escape($stream) . "'"
        );
    }

    public function dueStreams(int $limit = 20): array {
        $limit = max(1, min(100, $limit));
        $providers = $this->marketplaceProviderSql();
        $query = $this->db->query(
            "SELECT s.*, c.`provider`, c.`status` AS `channel_status`
             FROM `{$this->prefix}messenger_channel_streams` s
             INNER JOIN `{$this->prefix}messenger_channels` c ON c.`channel_id` = s.`channel_id`
             WHERE s.`status` = 'enabled'
               AND c.`status` = 'active'
               AND c.`provider` IN ({$providers})
               AND (s.`next_run_at` IS NULL OR s.`next_run_at` <= NOW())
               AND (s.`lease_until` IS NULL OR s.`lease_until` < NOW())
             ORDER BY COALESCE(s.`next_run_at`, '1970-01-01 00:00:00') ASC, s.`stream_id` ASC
             LIMIT {$limit}"
        );
        return $query->rows ?: [];
    }

    





    public function reserveRateBudget(
        string $provider,
        string $scope_hash,
        string $credential_class,
        string $api_group,
        int $interval_seconds,
        int $channel_id,
        string $current_stream
    ): array {
        $provider = $this->normalizeProvider($provider);
        $scope_hash = preg_match('/^[a-f0-9]{64}$/', $scope_hash) ? $scope_hash : '';
        $credential_class = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', $credential_class));
        $api_group = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', $api_group));
        $current_stream = $this->normalizeStream($current_stream);
        $interval_seconds = max(1, min(21600, $interval_seconds));
        if ($provider === '' || $scope_hash === '' || $credential_class === '' || $api_group === '' || $current_stream === '') {
            return ['allowed' => false, 'delay_seconds' => 60];
        }

        $provider_sql = $this->db->escape($provider);
        $scope_sql = $this->db->escape($scope_hash);
        $class_sql = $this->db->escape($credential_class);
        $group_sql = $this->db->escape($api_group);
        $this->db->query(
            "INSERT IGNORE INTO `{$this->prefix}marketplace_rate_budgets` SET
                `provider` = '{$provider_sql}',
                `account_scope_hash` = '{$scope_sql}',
                `credential_class` = '{$class_sql}',
                `api_group` = '{$group_sql}',
                `last_stream` = '',
                `blocked_until` = '1970-01-01 00:00:01',
                `date_added` = NOW(), `date_modified` = NOW()"
        );
        if ($api_group === 'feedbacks' && $channel_id > 0) {
            $budget = $this->db->query(
                "SELECT `last_stream`,
                        (`blocked_until` IS NULL OR `blocked_until` <= NOW()) AS `available_now`
                 FROM `{$this->prefix}marketplace_rate_budgets`
                 WHERE `provider` = '{$provider_sql}'
                   AND `account_scope_hash` = '{$scope_sql}'
                   AND `credential_class` = '{$class_sql}'
                   AND `api_group` = '{$group_sql}' LIMIT 1"
            );
            $last_stream = (string)($budget->row['last_stream'] ?? '');
            if ($last_stream === $current_stream
                && !empty($budget->row['available_now'])
                && $this->hasDueFeedbackSibling($channel_id, $current_stream)) {
                 
                 
                 
                return ['allowed' => false, 'delay_seconds' => 5, 'fairness_yield' => true];
            }
        }
        $stream_sql = $this->db->escape($current_stream);
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_rate_budgets`
             SET `blocked_until` = DATE_ADD(NOW(), INTERVAL {$interval_seconds} SECOND),
                 `last_stream` = '{$stream_sql}',
                 `reservation_count` = LEAST(`reservation_count` + 1, 4294967295),
                 `date_modified` = NOW()
             WHERE `provider` = '{$provider_sql}'
               AND `account_scope_hash` = '{$scope_sql}'
               AND `credential_class` = '{$class_sql}'
               AND `api_group` = '{$group_sql}'
               AND (`blocked_until` IS NULL OR `blocked_until` <= NOW())"
        );
        if ((int)$this->db->countAffected() > 0) {
            return ['allowed' => true, 'delay_seconds' => 0];
        }
        $query = $this->db->query(
            "SELECT GREATEST(1, TIMESTAMPDIFF(SECOND, NOW(), `blocked_until`)) AS `delay_seconds`
             FROM `{$this->prefix}marketplace_rate_budgets`
             WHERE `provider` = '{$provider_sql}'
               AND `account_scope_hash` = '{$scope_sql}'
               AND `credential_class` = '{$class_sql}'
               AND `api_group` = '{$group_sql}' LIMIT 1"
        );
        return [
            'allowed' => false,
            'delay_seconds' => max(1, min(21600, (int)($query->row['delay_seconds'] ?? $interval_seconds))),
        ];
    }

    private function hasDueFeedbackSibling(int $channel_id, string $current_stream): bool {
        $siblings = ['question', 'review'];
        $sibling = $current_stream === 'question' ? 'review' : 'question';
        if (!in_array($current_stream, $siblings, true)) return false;
        $query = $this->db->query(
            "SELECT `stream_id`
             FROM `{$this->prefix}messenger_channel_streams`
             WHERE `channel_id` = " . (int)$channel_id . "
               AND `stream` = '" . $this->db->escape($sibling) . "'
               AND `status` = 'enabled'
               AND (`next_run_at` IS NULL OR `next_run_at` <= NOW())
               AND (`lease_until` IS NULL OR `lease_until` < NOW())
             LIMIT 1"
        );
        return !empty($query->num_rows);
    }

    public function deferRateBudget(
        string $provider,
        string $scope_hash,
        string $credential_class,
        string $api_group,
        int $delay_seconds
    ): void {
        $provider = $this->normalizeProvider($provider);
        if ($provider === '' || preg_match('/^[a-f0-9]{64}$/', $scope_hash) !== 1) return;
        $credential_class = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', $credential_class));
        $api_group = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', $api_group));
        $delay_seconds = max(1, min(21600, $delay_seconds));
        if ($credential_class === '' || $api_group === '') return;
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_rate_budgets`
             SET `blocked_until` = GREATEST(
                    COALESCE(`blocked_until`, NOW()),
                    DATE_ADD(NOW(), INTERVAL {$delay_seconds} SECOND)
                 ),
                 `throttle_count` = LEAST(`throttle_count` + 1, 4294967295),
                 `date_modified` = NOW()
             WHERE `provider` = '" . $this->db->escape($provider) . "'
               AND `account_scope_hash` = '" . $this->db->escape($scope_hash) . "'
               AND `credential_class` = '" . $this->db->escape($credential_class) . "'
               AND `api_group` = '" . $this->db->escape($api_group) . "'"
        );
    }

    public function wakeStream(int $channel_id, string $stream): void {
        $stream = $this->normalizeStream($stream);
        if ($channel_id <= 0 || $stream === '') {
            return;
        }
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `next_run_at` = NOW(), `date_modified` = NOW()
             WHERE `channel_id` = {$channel_id}
               AND `stream` = '" . $this->db->escape($stream) . "'
               AND `status` = 'enabled'"
        );
    }

    



    public function rememberNotificationHint(
        int $channel_id,
        string $stream,
        string $entity_type,
        string $entity_id,
        array $payload = []
    ): bool {
        $stream = $this->normalizeStream($stream);
        $entity_type = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', $entity_type));
        $entity_id = MarketplaceText::truncate(trim($entity_id), 191);
        $allowed_types = [
            'chat',
            'chat_message',
            'question',
            'question_answer',
            'question_comment',
            'feedback',
            'feedback_comment',
        ];
        if ($channel_id <= 0 || $stream === '' || $entity_id === ''
            || !in_array($entity_type, $allowed_types, true)) {
            return false;
        }
         
         
        $entity_hash = hash('sha256', $entity_id);

        $this->db->query(
            "INSERT INTO `{$this->prefix}marketplace_sync_hints` SET
                `channel_id` = {$channel_id},
                `stream` = '" . $this->db->escape($stream) . "',
                `entity_type` = '" . $this->db->escape($entity_type) . "',
                `entity_id` = '" . $this->db->escape($entity_id) . "',
                `entity_hash` = '" . $this->db->escape($entity_hash) . "',
                `payload_json` = '" . $this->db->escape($this->safeJson($payload)) . "',
                `date_added` = NOW(), `date_modified` = NOW()
             ON DUPLICATE KEY UPDATE
                `entity_id` = VALUES(`entity_id`),
                `payload_json` = VALUES(`payload_json`),
                `date_modified` = NOW()"
        );
        return true;
    }

     
    public function listNotificationHints(int $stream_id, int $limit = 20): array {
        if ($stream_id <= 0) {
            return [];
        }
        $limit = max(1, min(100, $limit));
        $query = $this->db->query(
            "SELECT h.*
             FROM `{$this->prefix}marketplace_sync_hints` h
             INNER JOIN `{$this->prefix}messenger_channel_streams` s
                ON s.`channel_id` = h.`channel_id` AND s.`stream` = h.`stream`
             WHERE s.`stream_id` = {$stream_id}
             ORDER BY h.`hint_id` ASC
             LIMIT {$limit}"
        );
        $rows = [];
        foreach (($query->rows ?: []) as $row) {
            $payload = json_decode((string)($row['payload_json'] ?? ''), true);
            $row['payload'] = is_array($payload) ? $payload : [];
            unset($row['payload_json']);
            $rows[] = $row;
        }
        return $rows;
    }

     
    public function deleteNotificationHints(array $hint_ids): void {
        $ids = [];
        foreach ($hint_ids as $hint_id) {
            $hint_id = (int)$hint_id;
            if ($hint_id > 0) {
                $ids[$hint_id] = $hint_id;
            }
        }
        if (!$ids) {
            return;
        }
        $this->db->query(
            "DELETE FROM `{$this->prefix}marketplace_sync_hints`
             WHERE `hint_id` IN (" . implode(',', array_values($ids)) . ")"
        );
    }

    public function listAccounts(string $provider = ''): array {
        $provider = $this->normalizeProvider($provider);
        $where = "c.`provider` IN (" . $this->marketplaceProviderSql() . ")";
        if ($provider !== '') {
            $where = "c.`provider` = '" . $this->db->escape($provider) . "'";
        }
        $query = $this->db->query(
            "SELECT c.*,
                    COUNT(s.`stream_id`) AS `stream_count`,
                    SUM(s.`status` = 'enabled') AS `enabled_stream_count`,
                    MAX(s.`last_success_at`) AS `last_sync_at`,
                    MAX(s.`last_error_at`) AS `last_stream_error_at`,
                    SUM(s.`failure_count`) AS `stream_failure_count`
             FROM `{$this->prefix}messenger_channels` c
             LEFT JOIN `{$this->prefix}messenger_channel_streams` s ON s.`channel_id` = c.`channel_id`
             WHERE {$where}
             GROUP BY c.`channel_id`
             ORDER BY c.`provider` ASC, c.`channel_id` ASC"
        );
        return $query->rows ?: [];
    }

    public function acquireStreamLease(int $stream_id, int $ttl_seconds = 45): string {
        if ($stream_id <= 0) {
            return '';
        }
        $ttl_seconds = max(30, min(300, $ttl_seconds));
        $token = bin2hex(random_bytes(32));
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `lease_token` = '" . $this->db->escape($token) . "',
                 `lease_until` = DATE_ADD(NOW(), INTERVAL {$ttl_seconds} SECOND),
                 `last_started_at` = NOW(), `date_modified` = NOW()
             WHERE `stream_id` = {$stream_id}
               AND `status` = 'enabled'
               AND (`lease_until` IS NULL OR `lease_until` < NOW())"
        );
        return (int)$this->db->countAffected() > 0 ? $token : '';
    }

     
    public function renewStreamLease(int $stream_id, string $lease_token, int $ttl_seconds = 300): bool {
        if (!$this->validLeaseInput($stream_id, $lease_token)) {
            return false;
        }
        $ttl_seconds = max(30, min(300, $ttl_seconds));
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `lease_until` = GREATEST(
                     DATE_ADD(NOW(), INTERVAL {$ttl_seconds} SECOND),
                     DATE_ADD(`lease_until`, INTERVAL 1 SECOND)
                 ),
                 `date_modified` = NOW()
             WHERE `stream_id` = {$stream_id}
               AND `status` = 'enabled'
               AND `lease_token` = '" . $this->db->escape($lease_token) . "'
               AND `lease_until` >= NOW()"
        );
        return (int)$this->db->countAffected() > 0;
    }

    public function finishStreamSuccess(int $stream_id, string $lease_token, array $cursor, int $delay_seconds): void {
        if (!$this->validLeaseInput($stream_id, $lease_token)) {
            return;
        }
        $delay_seconds = max(5, min(21600, $delay_seconds));
        $cursor_json = json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        if (!is_string($cursor_json)) {
            $cursor_json = '{}';
        }
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `cursor_json` = '" . $this->db->escape($cursor_json) . "',
                 `lease_token` = '', `lease_until` = NULL,
                 `next_run_at` = DATE_ADD(NOW(), INTERVAL {$delay_seconds} SECOND),
                 `last_success_at` = NOW(), `last_error` = '', `failure_count` = 0,
                 `date_modified` = NOW()
             WHERE `stream_id` = {$stream_id}
               AND `lease_token` = '" . $this->db->escape($lease_token) . "'"
        );
    }

    






    public function deferStreamWithoutAttempt(int $stream_id, string $lease_token, int $delay_seconds): void {
        if (!$this->validLeaseInput($stream_id, $lease_token)) {
            return;
        }
        $delay_seconds = max(5, min(21600, $delay_seconds));
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `lease_token` = '', `lease_until` = NULL,
                 `next_run_at` = DATE_ADD(NOW(), INTERVAL {$delay_seconds} SECOND),
                 `date_modified` = NOW()
             WHERE `stream_id` = {$stream_id}
               AND `lease_token` = '" . $this->db->escape($lease_token) . "'"
        );
    }

    public function finishStreamFailure(int $stream_id, string $lease_token, string $error, int $delay_seconds): void {
        if (!$this->validLeaseInput($stream_id, $lease_token)) {
            return;
        }
        $delay_seconds = max(15, min(21600, $delay_seconds));
        $error = MarketplaceText::truncate(trim($error), 512);
        $this->db->query(
            "UPDATE `{$this->prefix}messenger_channel_streams`
             SET `lease_token` = '', `lease_until` = NULL,
                 `next_run_at` = DATE_ADD(NOW(), INTERVAL {$delay_seconds} SECOND),
                 `last_error_at` = NOW(),
                 `last_error` = '" . $this->db->escape($error) . "',
                 `failure_count` = LEAST(`failure_count` + 1, 1000000),
                 `date_modified` = NOW()
             WHERE `stream_id` = {$stream_id}
               AND `lease_token` = '" . $this->db->escape($lease_token) . "'"
        );
    }

    public function upsertThread(array $data): array {
        $channel_id = (int)($data['channel_id'] ?? 0);
        $provider = $this->normalizeProvider($data['provider'] ?? '');
        $kind = $this->normalizeStream($data['conversation_kind'] ?? '');
        $external_thread_id = MarketplaceText::truncate(trim((string)($data['external_thread_id'] ?? '')), 191);
        if ($channel_id <= 0 || $provider === '' || $kind === '' || $external_thread_id === '') {
            return [];
        }

        $snapshot = $this->safeJson($data['snapshot'] ?? []);
        $capabilities = $this->safeJson($data['capabilities'] ?? []);
        $external_chat_id = MarketplaceText::truncate(trim((string)($data['external_chat_id'] ?? $external_thread_id)), 191);
        $external_thread_hash = hash('sha256', $external_thread_id);
        $state = MarketplaceText::truncate(preg_replace('/[^a-z0-9_\-]/i', '', (string)($data['state'] ?? 'open')), 32);
        $item_id = MarketplaceText::truncate(trim((string)($data['item_id'] ?? '')), 128);
        $order_id = MarketplaceText::truncate(trim((string)($data['order_id'] ?? '')), 128);
        $reply_secret = (string)($data['reply_secret'] ?? '');
        $last_external_at = trim((string)($data['last_external_at'] ?? ''));

        $this->db->query(
            "INSERT INTO `{$this->prefix}marketplace_threads` SET
                `channel_id` = {$channel_id},
                `provider` = '" . $this->db->escape($provider) . "',
                `conversation_kind` = '" . $this->db->escape($kind) . "',
                `external_thread_id` = '" . $this->db->escape($external_thread_id) . "',
                `external_thread_hash` = '" . $this->db->escape($external_thread_hash) . "',
                `external_chat_id` = '" . $this->db->escape($external_chat_id) . "',
                `state` = '" . $this->db->escape($state !== '' ? $state : 'open') . "',
                `reply_secret` = '" . $this->db->escape($reply_secret) . "',
                `item_id` = '" . $this->db->escape($item_id) . "',
                `order_id` = '" . $this->db->escape($order_id) . "',
                `capabilities_json` = '" . $this->db->escape($capabilities) . "',
                `snapshot_json` = '" . $this->db->escape($snapshot) . "',
                `sync_revision` = 1,
                `last_external_at` = " . ($last_external_at !== '' ? "'" . $this->db->escape($last_external_at) . "'" : 'NULL') . ",
                `date_added` = NOW(), `date_modified` = NOW()
             ON DUPLICATE KEY UPDATE
                `external_chat_id` = VALUES(`external_chat_id`),
                `state` = VALUES(`state`),
                `reply_secret` = IF(VALUES(`reply_secret`) = '', `reply_secret`, VALUES(`reply_secret`)),
                `item_id` = IF(VALUES(`item_id`) = '', `item_id`, VALUES(`item_id`)),
                `order_id` = IF(VALUES(`order_id`) = '', `order_id`, VALUES(`order_id`)),
                `capabilities_json` = VALUES(`capabilities_json`),
                `snapshot_json` = VALUES(`snapshot_json`),
                `sync_revision` = `sync_revision` + 1,
                `last_external_at` = COALESCE(VALUES(`last_external_at`), `last_external_at`),
                `date_modified` = NOW()"
        );
        return $this->getThread($channel_id, $kind, $external_thread_id);
    }

    public function getThread(int $channel_id, string $kind, string $external_thread_id): array {
        $kind = $this->normalizeStream($kind);
        if ($channel_id <= 0 || $kind === '' || trim($external_thread_id) === '') {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_threads`
             WHERE `channel_id` = {$channel_id}
               AND `conversation_kind` = '" . $this->db->escape($kind) . "'
               AND `external_thread_hash` = '" . $this->db->escape(hash('sha256', trim($external_thread_id))) . "'
               AND `external_thread_id` = '" . $this->db->escape(trim($external_thread_id)) . "'
             LIMIT 1"
        );
        return $query->row;
    }

     
    public function getThreadByOrder(int $channel_id, string $kind, string $order_id): array {
        $kind = $this->normalizeStream($kind);
        $order_id = MarketplaceText::truncate(trim($order_id), 128);
        if ($channel_id <= 0 || $kind === '' || $order_id === '') {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_threads`
             WHERE `channel_id` = {$channel_id}
               AND `conversation_kind` = '" . $this->db->escape($kind) . "'
               AND `order_id` = '" . $this->db->escape($order_id) . "'
             ORDER BY `thread_id` DESC LIMIT 1"
        );
        return $query->row ?: [];
    }

    




    public function getYandexThreadByContext(int $channel_id, string $context_type, string $context_id): array {
        $context_type = strtoupper(trim($context_type));
        $context_id = MarketplaceText::truncate(trim($context_id), 128);
        if ($channel_id <= 0 || !in_array($context_type, ['ORDER', 'RETURN'], true) || $context_id === '') {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_threads`
             WHERE `channel_id` = {$channel_id}
               AND `provider` = 'yandex_market'
               AND `conversation_kind` = 'buyer_chat'
               AND `order_id` = '" . $this->db->escape($context_id) . "'
             ORDER BY `thread_id` DESC LIMIT 10"
        );
        foreach (($query->rows ?: []) as $row) {
            $snapshot = json_decode((string)($row['snapshot_json'] ?? ''), true);
            $provider_payload = is_array($snapshot['provider_payload'] ?? null)
                ? $snapshot['provider_payload']
                : [];
            $chat = is_array($provider_payload['chat'] ?? null) ? $provider_payload['chat'] : [];
            $context = is_array($chat['context'] ?? null) ? $chat['context'] : [];
            if (strtoupper((string)($context['type'] ?? '')) === $context_type) {
                return $row;
            }
        }
        return [];
    }

    public function getThreadByChat(int $chat_id): array {
        if ($chat_id <= 0) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_threads` WHERE `chat_id` = {$chat_id} LIMIT 1"
        );
        return $query->row;
    }

    public function getThreadById(int $thread_id): array {
        if ($thread_id <= 0) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_threads` WHERE `thread_id` = {$thread_id} LIMIT 1"
        );
        return $query->row ?: [];
    }

     
    public function confirmReadReceipt(int $thread_id, int $message_id): void {
        if ($thread_id <= 0 || $message_id <= 0) {
            return;
        }
        $thread = $this->getThreadById($thread_id);
        if (!$thread) {
            return;
        }
        $capabilities = json_decode((string)($thread['capabilities_json'] ?? ''), true);
        $capabilities = is_array($capabilities) ? $capabilities : [];
        if ((int)($capabilities['read_ack_message_id'] ?? 0) !== $message_id) {
            return;
        }
        $capabilities['read_ack_message_id'] = 0;
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_threads`
             SET `capabilities_json` = '" . $this->db->escape($this->safeJson($capabilities)) . "',
                 `date_modified` = NOW()
             WHERE `thread_id` = {$thread_id}"
        );
    }

    public function linkThreadToChat(int $thread_id, int $chat_id): void {
        if ($thread_id <= 0 || $chat_id <= 0) {
            return;
        }
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_threads`
             SET `chat_id` = {$chat_id}, `date_modified` = NOW()
             WHERE `thread_id` = {$thread_id}"
        );
    }

    





    public function beginAction(array $data): array {
        $channel_id = (int)($data['channel_id'] ?? 0);
        $thread_id = (int)($data['thread_id'] ?? 0);
        $provider = $this->normalizeProvider($data['provider'] ?? '');
        $kind = $this->normalizeStream($data['conversation_kind'] ?? '');
        $action = MarketplaceText::truncate(preg_replace('/[^a-z0-9_\-]/i', '', (string)($data['action'] ?? '')), 64);
        $key = strtolower(trim((string)($data['idempotency_key'] ?? '')));
        $resource_key = strtolower(trim((string)($data['resource_key'] ?? '')));
        if ($channel_id <= 0 || $provider === '' || $kind === '' || $action === ''
            || preg_match('/^[a-f0-9]{64}$/', $key) !== 1
            || ($resource_key !== '' && preg_match('/^[a-f0-9]{64}$/', $resource_key) !== 1)) {
            return [];
        }

        $request = $this->safeJson($data['request'] ?? []);
        $manager_id = max(0, (int)($data['manager_id'] ?? 0));
        $thread_revision = max(0, (int)($data['thread_revision'] ?? 0));
        $this->db->query(
            "INSERT IGNORE INTO `{$this->prefix}marketplace_actions` SET
                `channel_id` = {$channel_id},
                `thread_id` = " . ($thread_id > 0 ? $thread_id : 'NULL') . ",
                `provider` = '" . $this->db->escape($provider) . "',
                `conversation_kind` = '" . $this->db->escape($kind) . "',
                `action` = '" . $this->db->escape($action) . "',
                `resource_key` = '" . $this->db->escape($resource_key) . "',
                `idempotency_key` = '" . $this->db->escape($key) . "',
                `thread_revision` = {$thread_revision},
                `request_json` = '" . $this->db->escape($request) . "',
                `response_json` = '{}', `status` = 'pending',
                `manager_id` = {$manager_id}, `date_added` = NOW(), `date_modified` = NOW()"
        );
        $created = (int)$this->db->countAffected() > 0;
        $action = $created
            ? $this->getAction((int)$this->db->getLastId())
            : $this->getActionByIdempotencyKey($channel_id, $key);
        if ($action) {
             
             
            $action['created'] = $created;
        }
        return $action;
    }

    public function finishAction(int $action_id, string $status, array $response = [], string $error = ''): void {
        $status = in_array($status, ['succeeded', 'failed', 'unknown'], true) ? $status : 'unknown';
        if ($action_id <= 0) {
            return;
        }
        $revision_sql = $status === 'unknown'
            ? "`thread_revision` = GREATEST(
                    `thread_revision`,
                    COALESCE((
                        SELECT t.`sync_revision`
                        FROM `{$this->prefix}marketplace_threads` t
                        WHERE t.`thread_id` = `{$this->prefix}marketplace_actions`.`thread_id`
                        LIMIT 1
                    ), 0)
                ),"
            : '';
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_actions` SET
                `status` = '" . $this->db->escape($status) . "',
                {$revision_sql}
                `response_json` = '" . $this->db->escape($this->safeJson($response)) . "',
                `error` = '" . $this->db->escape(MarketplaceText::truncate(trim($error), 512)) . "',
                `date_modified` = NOW()
             WHERE `action_id` = " . (int)$action_id
        );
    }

    public function getAction(int $action_id): array {
        if ($action_id <= 0) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_actions` WHERE `action_id` = " . (int)$action_id . " LIMIT 1"
        );
        return $query->row ?: [];
    }

    





    public function conversationStartResourceKey(string $resource): string {
        $resource = MarketplaceText::truncate(trim($resource), 128);
        return $resource === '' ? '' : hash('sha256', 'ozon:start:' . $resource);
    }

    public function getConversationStartAction(int $channel_id, string $resource_key): array {
        $resource_key = strtolower(trim($resource_key));
        if ($channel_id <= 0 || preg_match('/^[a-f0-9]{64}$/', $resource_key) !== 1) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_actions`
             WHERE `channel_id` = {$channel_id}
               AND `action` = 'start_conversation'
               AND `resource_key` = '" . $this->db->escape($resource_key) . "'
               AND `status` IN ('pending', 'unknown', 'succeeded')
             ORDER BY `action_id` DESC LIMIT 1"
        );
        return $query->row ?: [];
    }

    



    public function reconcileConversationStart(int $channel_id, string $resource, array $response): void {
        $resource_key = $this->conversationStartResourceKey($resource);
        if ($channel_id <= 0 || $resource_key === '') {
            return;
        }
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_actions` SET
                `status` = 'succeeded',
                `response_json` = '" . $this->db->escape($this->safeJson($response)) . "',
                `error` = '', `date_modified` = NOW()
             WHERE `channel_id` = {$channel_id}
               AND `action` = 'start_conversation'
               AND `resource_key` = '" . $this->db->escape($resource_key) . "'
               AND `status` IN ('pending', 'unknown')"
        );
    }

    public function getActionByIdempotencyKey(int $channel_id, string $key): array {
        $key = strtolower(trim($key));
        if ($channel_id <= 0 || preg_match('/^[a-f0-9]{64}$/', $key) !== 1) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_actions`
             WHERE `channel_id` = " . (int)$channel_id . "
               AND `idempotency_key` = '" . $this->db->escape($key) . "' LIMIT 1"
        );
        return $query->row ?: [];
    }

    






    public function getUnreconciledAction(int $thread_id, int $thread_revision): array {
        if ($thread_id <= 0 || $thread_revision < 0) {
            return [];
        }
        $query = $this->db->query(
            "SELECT * FROM `{$this->prefix}marketplace_actions`
             WHERE `thread_id` = {$thread_id}
               AND (
                    `status` = 'pending'
                    OR (`status` = 'unknown' AND `thread_revision` >= {$thread_revision})
               )
             ORDER BY `action_id` DESC LIMIT 1"
        );
        return $query->row ?: [];
    }

    




    public function acquireActionLock(int $thread_id): bool {
        if ($thread_id <= 0) {
            return false;
        }
        $lock_name = $this->actionLockName($thread_id);
        $query = $this->db->query(
            "SELECT GET_LOCK('" . $this->db->escape($lock_name) . "', 0) AS `locked`"
        );
        return (int)($query->row['locked'] ?? 0) === 1;
    }

    public function releaseActionLock(int $thread_id): void {
        if ($thread_id <= 0) {
            return;
        }
        $lock_name = $this->actionLockName($thread_id);
        $this->db->query("SELECT RELEASE_LOCK('" . $this->db->escape($lock_name) . "')");
    }

    public function acquireConversationStartLock(int $channel_id, string $resource_key): bool {
        $lock_name = $this->conversationStartLockName($channel_id, $resource_key);
        if ($lock_name === '') {
            return false;
        }
        $query = $this->db->query(
            "SELECT GET_LOCK('" . $this->db->escape($lock_name) . "', 0) AS `locked`"
        );
        return (int)($query->row['locked'] ?? 0) === 1;
    }

    public function releaseConversationStartLock(int $channel_id, string $resource_key): void {
        $lock_name = $this->conversationStartLockName($channel_id, $resource_key);
        if ($lock_name !== '') {
            $this->db->query("SELECT RELEASE_LOCK('" . $this->db->escape($lock_name) . "')");
        }
    }

     
    public function applySuccessfulAction(int $thread_id, string $action): void {
        $action = MarketplaceText::truncate(preg_replace('/[^a-z0-9_\-]/i', '', $action), 64);
        if ($thread_id <= 0 || $action === '') {
            return;
        }
        $query = $this->db->query(
            "SELECT `provider`, `capabilities_json` FROM `{$this->prefix}marketplace_threads`
             WHERE `thread_id` = {$thread_id} LIMIT 1"
        );
        if (!$query->row) {
            return;
        }
        $capabilities = json_decode((string)($query->row['capabilities_json'] ?? ''), true);
        $capabilities = is_array($capabilities) ? $capabilities : [];
        $actions = array_values(array_unique(array_filter(array_map('strval', (array)($capabilities['actions'] ?? [])))));
        $state = 'processed';

        if ($action === 'question_mark_viewed') {
            $actions = array_values(array_diff($actions, [$action]));
            $state = 'viewed';
        } elseif ($action === 'review_edit') {
            $actions = ['review_edit'];
            $state = 'answered';
        } elseif ($action === 'review_answer' && (string)$query->row['provider'] === 'wildberries') {
            $actions = ['review_edit'];
            $state = 'answered';
        } else {
            $actions = [];
        }
        $capabilities['actions'] = $actions;
        if ($action === 'return_action') {
            $capabilities['workflow_actions'] = [];
            $capabilities['allowed_claim_actions'] = [];
            $capabilities['return_rejection_reasons'] = [];
            $state = 'decision_sent';
        }
        $this->db->query(
            "UPDATE `{$this->prefix}marketplace_threads` SET
                `state` = '" . $this->db->escape($state) . "',
                `capabilities_json` = '" . $this->db->escape($this->safeJson($capabilities)) . "',
                `date_modified` = NOW()
             WHERE `thread_id` = {$thread_id}"
        );
    }

    private function validLeaseInput(int $stream_id, string $lease_token): bool {
        return $stream_id > 0 && preg_match('/^[a-f0-9]{64}$/', $lease_token) === 1;
    }

    private function normalizeStream($stream): string {
        $stream = strtolower(preg_replace('/[^a-z0-9_\-]/i', '', (string)$stream));
        return in_array($stream, ['buyer_chat', 'question', 'review', 'return_claim'], true) ? $stream : '';
    }

    private function normalizeProvider($provider): string {
        return MarketplaceProviderCatalog::normalize((string)$provider);
    }

    private function marketplaceProviderSql(): string {
        $values = [];
        foreach (MarketplaceProviderCatalog::codes() as $provider) {
            $values[] = "'" . $this->db->escape($provider) . "'";
        }
        return implode(',', $values);
    }

    private function actionLockName(int $thread_id): string {
        return 'easychat_action_' . substr(hash('sha256', $this->prefix . ':' . $thread_id), 0, 40);
    }

    private function conversationStartLockName(int $channel_id, string $resource_key): string {
        $resource_key = strtolower(trim($resource_key));
        if ($channel_id <= 0 || preg_match('/^[a-f0-9]{64}$/', $resource_key) !== 1) {
            return '';
        }
        return 'easychat_marketplace_start_' . substr(hash('sha256', $this->prefix . ':' . $channel_id . ':' . $resource_key), 0, 34);
    }

    private function safeJson($value): string {
        $json = json_encode(is_array($value) ? $value : [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        return is_string($json) ? $json : '{}';
    }
}
