<?php

declare(strict_types=1);

namespace app\common\service\qywx;

use RuntimeException;
use think\facade\Db;

/** 获客客户归因、会话统计与回调幂等落库。 */
class QywxCustomerAcquisitionCustomerService
{
    private QywxCustomerAcquisitionApiService $api;

    public function __construct(?QywxCustomerAcquisitionApiService $api = null)
    {
        $this->api = $api ?? new QywxCustomerAcquisitionApiService();
    }

    /**
     * 同步一个远端获客链接的全部客户，远端列表字段采用覆盖语义。
     * recv_msg_cnt 不在列表接口中返回，因此同步时保留本地值。
     *
     * @return array{scanned:int,created:int,updated:int,pages:int,truncated:bool}
     */
    public function syncLink(string $remoteLinkId, int $maxCustomers = 20000): array
    {
        $remoteLinkId = trim($remoteLinkId);
        if ($remoteLinkId === '') {
            throw new RuntimeException('获客链接 ID 不能为空');
        }
        $cursor = '';
        $scanned = 0;
        $created = 0;
        $updated = 0;
        $pages = 0;
        do {
            $page = $this->api->listCustomers($remoteLinkId, $cursor, 1000);
            $pages++;
            foreach ($page['customer_list'] as $customer) {
                if ($scanned >= $maxCustomers) {
                    break 2;
                }
                $scanned++;
                $result = self::upsertCustomer($remoteLinkId, $customer, false);
                $result === 'created' ? $created++ : $updated++;
            }
            $cursor = (string) ($page['next_cursor'] ?? '');
        } while ($cursor !== '');

        return compact('scanned', 'created', 'updated', 'pages') + ['truncated' => $cursor !== ''];
    }

    /**
     * 处理 customer_acquisition 回调。相同事件只成功处理一次；失败事件保留审计并允许企微重试。
     *
     * @return array{duplicate:bool,status:string}
     */
    public function handleCallback(array $message): array
    {
        $changeType = trim((string) ($message['ChangeType'] ?? $message['change_type'] ?? ''));
        if (!in_array($changeType, ['customer_start_chat', 'message_from_customer'], true)) {
            return ['duplicate' => false, 'status' => 'ignored'];
        }
        $chatKey = trim((string) ($message['ChatKey'] ?? $message['Chatkey'] ?? $message['chat_key'] ?? ''));
        $eventTime = (int) ($message['CreateTime'] ?? $message['create_time'] ?? 0);
        $eventKey = self::eventKey($message, $changeType, $chatKey, $eventTime);
        $event = self::beginEvent($eventKey, $changeType, $chatKey, $eventTime, $message);
        if (($event['duplicate'] ?? false) === true) {
            return ['duplicate' => true, 'status' => 'success'];
        }

        $eventId = (int) ($event['id'] ?? 0);
        try {
            // customer_start_chat 仅能确认“客户已发起会话”，企业微信不保证该事件携带 ChatKey。
            // 此时先落归因与聊天状态，精确消息数等待 message_from_customer 回调补齐。
            if ($changeType === 'customer_start_chat' && $chatKey === '') {
                $remoteLinkId = trim((string) (
                    $message['LinkID'] ?? $message['LinkId'] ?? $message['link_id'] ?? ''
                ));
                $externalUserId = trim((string) (
                    $message['ExternalUserID'] ?? $message['ExternalUserId'] ?? $message['external_userid'] ?? ''
                ));
                $userId = trim((string) ($message['UserID'] ?? $message['UserId'] ?? $message['userid'] ?? ''));
                if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
                    throw new RuntimeException('customer_start_chat 回调缺少 link_id / external_userid / userid');
                }
                self::upsertCustomer($remoteLinkId, [
                    'external_userid' => $externalUserId,
                    'userid' => $userId,
                    'chat_status' => 1,
                    'state' => (string) ($message['State'] ?? $message['state'] ?? ''),
                    'event_time' => $eventTime,
                    'snapshot' => $message,
                ], false);
                self::recordPromotionAssignment(
                    (string) ($message['State'] ?? $message['state'] ?? ''),
                    $remoteLinkId,
                    $userId,
                    $externalUserId,
                    $eventTime
                );
                self::finishEvent($eventId, 1, '', $remoteLinkId, $externalUserId, $userId);

                return ['duplicate' => false, 'status' => 'success'];
            }
            if ($chatKey === '') {
                self::finishEvent($eventId, 3, 'failed_invalid: message_from_customer 回调缺少 ChatKey');
                throw new RuntimeException('message_from_customer 回调缺少 ChatKey');
            }
            $now = time();
            if ($eventTime > 0 && ($now - $eventTime) >= 1800) {
                throw new RuntimeException('获客回调 ChatKey 已超过 30 分钟有效期');
            }
            $chat = $this->api->getChatInfo($chatKey);
            $chatInfo = is_array($chat['chat_info'] ?? null) ? $chat['chat_info'] : [];
            $remoteLinkId = trim((string) (
                $chatInfo['link_id'] ?? $message['LinkID'] ?? $message['LinkId'] ?? $message['link_id'] ?? ''
            ));
            $externalUserId = trim((string) (
                $chat['external_userid'] ?? $message['ExternalUserID'] ?? $message['ExternalUserId'] ?? ''
            ));
            $userId = trim((string) ($chat['userid'] ?? $message['UserID'] ?? $message['UserId'] ?? ''));
            if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
                throw new RuntimeException('get_chat_info 未返回完整的 link_id / external_userid / userid');
            }
            self::upsertCustomer($remoteLinkId, [
                'external_userid' => $externalUserId,
                'userid' => $userId,
                'chat_status' => max(1, (int) ($message['ChatStatus'] ?? 1)),
                'recv_msg_cnt' => max(0, (int) ($chatInfo['recv_msg_cnt'] ?? 0)),
                'state' => (string) ($chatInfo['state'] ?? $message['State'] ?? ''),
                'event_time' => $eventTime,
                'snapshot' => $chat,
            ], true);
            self::recordPromotionAssignment(
                (string) ($chatInfo['state'] ?? $message['State'] ?? ''),
                $remoteLinkId,
                $userId,
                $externalUserId,
                $eventTime
            );
            self::finishEvent($eventId, 1, '', $remoteLinkId, $externalUserId, $userId);

            return ['duplicate' => false, 'status' => 'success'];
        } catch (\Throwable $e) {
            if ($eventId > 0 && str_contains($e->getMessage(), 'message_from_customer 回调缺少 ChatKey')) {
                throw $e;
            }
            self::scheduleRetryOrExpire($eventId, $eventTime, $e->getMessage());
            throw $e;
        }
    }

    /**
     * 重试仍在 ChatKey 30 分钟有效期内的失败回调，并把到期记录明确标记 failed_expired。
     *
     * @return array{selected:int,success:int,failed:int,expired:int}
     */
    public function retryPending(int $limit = 100): array
    {
        $now = time();
        // 进程在 beginEvent 后异常退出时，处理中事件会卡在 status=0；一分钟后自动回收再试。
        Db::name('qywx_customer_acquisition_event')
            ->where('status', 0)
            ->where('update_time', '<=', $now - 60)
            ->where('expire_time', '>', $now)
            ->update([
                'status' => 2,
                'next_retry' => $now,
                'error_message' => 'watchdog_recovered: 上次处理未正常结束',
                'update_time' => $now,
            ]);
        $expired = (int) Db::name('qywx_customer_acquisition_event')
            ->whereIn('status', [0, 2])
            ->where('expire_time', '>', 0)
            ->where('expire_time', '<=', $now)
            ->update([
                'status' => 3,
                'next_retry' => 0,
                'error_message' => 'failed_expired: ChatKey 已超过 30 分钟有效期',
                'chat_key' => '',
                'raw_json' => null,
                'update_time' => $now,
            ]);
        $rows = Db::name('qywx_customer_acquisition_event')
            ->where('status', 2)
            ->where('next_retry', '<=', $now)
            ->where('expire_time', '>', $now)
            ->order('next_retry', 'asc')
            ->limit(min(500, max(1, $limit)))
            ->select()->toArray();
        $success = 0;
        $failed = 0;
        foreach ($rows as $row) {
            $message = json_decode((string) ($row['raw_json'] ?? ''), true);
            if (!is_array($message)) {
                self::scheduleRetryOrExpire(
                    (int) $row['id'],
                    (int) ($row['event_time'] ?? 0),
                    '回调原始数据无法解析'
                );
                $failed++;
                continue;
            }
            try {
                $this->handleCallback($message);
                $success++;
            } catch (\Throwable) {
                $failed++;
            }
        }

        return ['selected' => count($rows), 'success' => $success, 'failed' => $failed, 'expired' => $expired];
    }

    public static function eventKey(array $message, string $changeType, string $chatKey, int $eventTime): string
    {
        $parts = [
            (string) ($message['MsgId'] ?? $message['MsgID'] ?? ''),
            $changeType,
            $chatKey,
            (string) $eventTime,
            (string) ($message['LinkID'] ?? $message['LinkId'] ?? ''),
            (string) ($message['ExternalUserID'] ?? $message['ExternalUserId'] ?? ''),
            (string) ($message['UserID'] ?? $message['UserId'] ?? ''),
        ];

        return hash('sha256', implode('|', $parts));
    }

    /** @return array{expire_time:int,next_retry:int,expired:bool} */
    public static function retryDecision(int $eventTime, int $now, int $storedExpireTime = 0): array
    {
        $expireTime = $storedExpireTime > 0
            ? $storedExpireTime
            : ($eventTime > 0 ? $eventTime + 1800 : $now + 1800);
        $expired = $expireTime <= $now;

        return [
            'expire_time' => $expireTime,
            'next_retry' => $expired ? 0 : min($expireTime - 1, $now + 30),
            'expired' => $expired,
        ];
    }

    /** @return array{id:int,duplicate:bool} */
    private static function beginEvent(
        string $eventKey,
        string $changeType,
        string $chatKey,
        int $eventTime,
        array $message
    ): array {
        $now = time();
        $raw = self::encodeJson($message);
        $expireTime = self::retryDecision($eventTime, $now)['expire_time'];
        try {
            $id = (int) Db::name('qywx_customer_acquisition_event')->insertGetId([
                'event_key' => $eventKey,
                'change_type' => $changeType,
                'chat_key' => $chatKey,
                'status' => 0,
                'attempts' => 1,
                'event_time' => max(0, $eventTime),
                'expire_time' => $expireTime,
                'next_retry' => 0,
                'error_message' => '',
                'raw_json' => $raw,
                'create_time' => $now,
                'update_time' => $now,
            ]);

            return ['id' => $id, 'duplicate' => false];
        } catch (\Throwable $e) {
            $existing = Db::name('qywx_customer_acquisition_event')->where('event_key', $eventKey)->find();
            if (!$existing) {
                throw $e;
            }
            if ((int) ($existing['status'] ?? 0) === 1) {
                return ['id' => (int) $existing['id'], 'duplicate' => true];
            }
            if ((int) ($existing['status'] ?? 0) !== 2) {
                return ['id' => (int) $existing['id'], 'duplicate' => true];
            }
            $claimed = Db::name('qywx_customer_acquisition_event')
                ->where('id', (int) $existing['id'])
                ->where('status', 2)
                ->update([
                    'status' => 0,
                    'attempts' => (int) ($existing['attempts'] ?? 0) + 1,
                    'error_message' => '',
                    'raw_json' => $raw,
                    'update_time' => $now,
                ]);
            if ($claimed <= 0) {
                return ['id' => (int) $existing['id'], 'duplicate' => true];
            }

            return ['id' => (int) $existing['id'], 'duplicate' => false];
        }
    }

    private static function finishEvent(
        int $id,
        int $status,
        string $error = '',
        string $remoteLinkId = '',
        string $externalUserId = '',
        string $userId = ''
    ): void {
        if ($id <= 0) {
            return;
        }
        Db::name('qywx_customer_acquisition_event')->where('id', $id)->update([
            'status' => $status,
            'link_id' => $remoteLinkId,
            'external_userid' => $externalUserId,
            'userid' => $userId,
            'error_message' => mb_substr($error, 0, 1000),
            'next_retry' => 0,
            // ChatKey 是短时敏感凭证，终态后不再保留；原始回调也随之清理。
            'chat_key' => '',
            'raw_json' => null,
            'update_time' => time(),
        ]);
    }

    private static function scheduleRetryOrExpire(int $id, int $eventTime, string $error): void
    {
        if ($id <= 0) {
            return;
        }
        $now = time();
        $expireTime = (int) (Db::name('qywx_customer_acquisition_event')
            ->where('id', $id)->value('expire_time') ?? 0);
        $decision = self::retryDecision($eventTime, $now, $expireTime);
        $expireTime = $decision['expire_time'];
        $expired = $decision['expired'];
        Db::name('qywx_customer_acquisition_event')->where('id', $id)->update([
            'status' => $expired ? 3 : 2,
            'expire_time' => $expireTime,
            'next_retry' => $decision['next_retry'],
            'error_message' => mb_substr(
                $expired ? 'failed_expired: ' . $error : $error,
                0,
                1000
            ),
            'chat_key' => $expired ? '' : Db::raw('chat_key'),
            'raw_json' => $expired ? null : Db::raw('raw_json'),
            'update_time' => $now,
        ]);
    }

    /** @return 'created'|'updated' */
    private static function upsertCustomer(string $remoteLinkId, array $customer, bool $messageCountKnown): string
    {
        $externalUserId = trim((string) ($customer['external_userid'] ?? ''));
        $userId = trim((string) ($customer['userid'] ?? ''));
        if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
            throw new RuntimeException('获客客户数据缺少 link_id / external_userid / userid');
        }
        [$ownerAdminId, $deptId] = self::resolveOwner($userId);
        $now = time();
        $existing = Db::name('qywx_customer_acquisition_customer')
            ->where('link_id', $remoteLinkId)
            ->where('external_userid', $externalUserId)
            ->where('userid', $userId)
            ->find();
        $snapshot = $customer['snapshot'] ?? $customer;
        $incomingChatStatus = max(0, min(2, (int) ($customer['chat_status'] ?? 0)));
        $data = [
            'promotion_link_id' => (int) (Db::name('qywx_promotion_link')
                ->where('remote_link_id', $remoteLinkId)->value('id') ?? 0),
            'owner_admin_id' => $ownerAdminId,
            'dept_id' => $deptId,
            'state' => mb_substr((string) ($customer['state'] ?? ''), 0, 255),
            // 已确认发过消息后，列表同步返回的“未发/未知”不得把状态回退。
            'chat_status' => $existing
                ? Db::raw('CASE WHEN chat_status = 1 OR ' . $incomingChatStatus . ' = 1 THEN 1 ELSE ' . $incomingChatStatus . ' END')
                : $incomingChatStatus,
            'last_sync_time' => $now,
            'raw_snapshot' => self::encodeJson($snapshot),
            'update_time' => $now,
        ];
        if ($messageCountKnown) {
            $remoteCount = max(0, (int) ($customer['recv_msg_cnt'] ?? 0));
            // get_chat_info 返回累计值，必须 max/覆盖，绝不按回调次数累加。
            $data['recv_msg_cnt'] = $existing
                ? Db::raw('GREATEST(recv_msg_cnt,' . $remoteCount . ')')
                : $remoteCount;
            $data['message_count_known'] = 1;
        }
        if ($incomingChatStatus === 1 || $messageCountKnown) {
            $eventTime = max(0, (int) ($customer['event_time'] ?? $now));
            $data['last_chat_time'] = $existing
                ? Db::raw('GREATEST(last_chat_time,' . $eventTime . ')')
                : $eventTime;
        }
        if ($existing) {
            Db::name('qywx_customer_acquisition_customer')->where('id', (int) $existing['id'])->update($data);

            return 'updated';
        }
        $data += [
            'link_id' => $remoteLinkId,
            'external_userid' => $externalUserId,
            'userid' => $userId,
            'recv_msg_cnt' => $messageCountKnown ? max(0, (int) ($customer['recv_msg_cnt'] ?? 0)) : 0,
            'message_count_known' => $messageCountKnown ? 1 : 0,
            'first_acquired_time' => max(0, (int) ($customer['create_time'] ?? $customer['event_time'] ?? $now)),
            'last_chat_time' => ($incomingChatStatus === 1 || $messageCountKnown)
                ? max(0, (int) ($customer['event_time'] ?? $now))
                : 0,
            'create_time' => $now,
        ];
        Db::name('qywx_customer_acquisition_customer')->insert($data);

        return 'created';
    }

    /** @return array{0:int,1:int} */
    private static function resolveOwner(string $userId): array
    {
        $adminId = (int) (Db::name('admin')->where('work_wechat_userid', $userId)
            ->whereNull('delete_time')->value('id') ?? 0);
        if ($adminId <= 0) {
            return [0, 0];
        }
        $deptId = (int) (Db::name('admin_dept')->where('admin_id', $adminId)
            ->order('dept_id', 'asc')->value('dept_id') ?? 0);

        return [$adminId, $deptId];
    }

    private static function recordPromotionAssignment(
        string $state,
        string $remoteLinkId,
        string $userId,
        string $externalUserId,
        int $eventTime
    ): void {
        $result = QywxPromotionMemberSchedulerService::recordFromState(
            $state,
            $userId,
            $externalUserId,
            $eventTime,
            'customer_acquisition'
        );
        if ((int) ($result['pool_id'] ?? 0) <= 0) {
            $result = QywxPromotionMemberSchedulerService::recordFromRemoteLink(
                $remoteLinkId,
                $userId,
                $externalUserId,
                $eventTime,
                'customer_acquisition'
            );
        }
        if (($result['status'] ?? '') !== 'counted' || (int) ($result['pool_id'] ?? 0) <= 0) {
            return;
        }
        try {
            // 优先在本次回调完成后切换，后台分钟任务继续承担失败重试。
            (new QywxPromotionRangeSyncService())->syncPool((int) $result['pool_id']);
        } catch (\Throwable) {
            // 同步服务已经保存失败原因和下次重试时间，不能让远端网络错误回滚已记账的客户。
        }
    }

    private static function encodeJson(mixed $value): string
    {
        $json = json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);

        return $json === false ? '{}' : $json;
    }
}