package cn.iocoder.yudao.module.im.service.message;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
import cn.iocoder.yudao.module.im.controller.admin.manager.message.vo.privates.ImPrivateMessageManagerPageReqVO;
import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageListReqVO;
import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageSendReqVO;
import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
import cn.iocoder.yudao.module.im.dal.mysql.message.ImPrivateMessageMapper;
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
import cn.iocoder.yudao.module.im.service.conversation.ImConversationReadService;
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReadNotification;
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReceiptNotification;
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImPrivateMessageNotification;
import cn.iocoder.yudao.module.im.dal.dataobject.message.content.QuoteMessage;
import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
import cn.iocoder.yudao.module.im.util.ImMessageUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.validation.annotation.Validated;

import java.time.LocalDateTime;
import java.util.List;

import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;

/**
 * IM 私聊消息 Service 实现类
 *
 * @author 芋道源码
 */
@Service
@Validated
@Slf4j
public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {

    @Resource
    private ImPrivateMessageMapper privateMessageMapper;

    @Resource
    private ImFriendService friendService;
    @Resource
    private ImSensitiveWordService sensitiveWordService;
    @Resource
    private ImConversationReadService conversationReadService;

    @Resource
    private ImWebSocketService imWebSocketService;

    @Resource
    private ImProperties imProperties;

    @Override
    public ImPrivateMessageDO sendPrivateMessage(Long senderId, ImPrivateMessageSendReqVO reqVO) {
        // 1.1 幂等校验：根据 senderId + clientMessageId 查重
        ImPrivateMessageDO existing = privateMessageMapper.selectBySenderIdAndClientMessageId(
                senderId, reqVO.getClientMessageId());
        if (existing != null) {
            log.info("[sendPrivateMessage][幂等命中 senderId({}) clientMessageId({}) 已存在消息({})]",
                    senderId, reqVO.getClientMessageId(), existing.getId());
            return existing;
        }
        // 1.2 消息内容校验
        ImMessageUtils.validateUserMessageContent(reqVO.getType(), reqVO.getContent());
        // 1.3 好友校验
        friendService.validateFriend(senderId, reqVO.getReceiverId());
        // 1.4 文本消息敏感词过滤
        if (ImContentTypeEnum.TEXT.getType().equals(reqVO.getType())) {
            sensitiveWordService.validateText(reqVO.getContent());
        }

        // 2.1 引用 quote 消息规范化
        reqVO.setContent(normalizeQuoteContent(reqVO, senderId));
        // 2.2 构建并保存消息；唯一键冲突时回查已存在消息返回
        // 用户私聊消息默认需要回执（receipt 不传按 true）；系统通知走 DTO 通道，默认不回执
        Boolean receipt = reqVO.getReceipt() != null ? reqVO.getReceipt() : Boolean.TRUE;
        ImPrivateMessageDO message = BeanUtils.toBean(reqVO, ImPrivateMessageDO.class, m -> m
                .setSenderId(senderId).setStatus(ImMessageStatusEnum.NORMAL.getStatus())
                .setReceiptStatus(resolveReceiptStatus(receipt)).setSendTime(LocalDateTime.now()));
        try {
            privateMessageMapper.insert(message);
        } catch (DuplicateKeyException e) {
            log.warn("[sendPrivateMessage][senderId({}) clientMessageId({}) 并发插入冲突，回查返回]",
                    senderId, reqVO.getClientMessageId());
            return privateMessageMapper.selectBySenderIdAndClientMessageId(senderId, reqVO.getClientMessageId());
        }

        // 3. WebSocket 异步推送：接收方 + 发送方多端同步
        ImPrivateMessageNotification notification = ImPrivateMessageNotification.ofSend(message);
        imWebSocketService.sendNotificationAsync(message.getReceiverId(), ImConversationTypeEnum.PRIVATE.getType(),
                notification.getType(), notification);
        imWebSocketService.sendNotificationAsync(senderId, ImConversationTypeEnum.PRIVATE.getType(),
                notification.getType(), notification);
        return message;
    }

    @Override
    public ImPrivateMessageDO sendPrivateMessage(Long senderId, ImPrivateMessageSendDTO dto) {
        // 1.1 content 序列化：null / String 透传，POJO 走 JSON
        Object payload = dto.getContent();
        String contentString = payload == null || payload instanceof String
                ? (String) payload
                : JsonUtils.toJsonString(payload);
        // 1.2 构建消息
        ImPrivateMessageDO message = new ImPrivateMessageDO().setClientMessageId(IdUtil.fastSimpleUUID())
                .setSenderId(senderId).setReceiverId(dto.getReceiverId())
                .setType(dto.getType()).setContent(contentString)
                .setStatus(ImMessageStatusEnum.NORMAL.getStatus())
                .setReceiptStatus(resolveReceiptStatus(dto.getReceipt())).setSendTime(LocalDateTime.now());
        // 1.3 决定是否持久化：dto.persistent 优先；为 null 时按 type 默认
        boolean persistent = dto.getPersistent() != null
                ? dto.getPersistent()
                : ImContentTypeEnum.validate(dto.getType()).isPersistent();
        if (persistent) {
            privateMessageMapper.insert(message);
        }

        // 2. WebSocket 异步推送：双向（默认）；单边语义（persistent=false）下仅推 sender 多端，对方不感知
        ImPrivateMessageNotification notification = ImPrivateMessageNotification.ofSend(message);
        if (persistent) {
            imWebSocketService.sendNotificationAsync(dto.getReceiverId(), ImConversationTypeEnum.PRIVATE.getType(),
                    notification.getType(), notification);
        }
        imWebSocketService.sendNotificationAsync(senderId, ImConversationTypeEnum.PRIVATE.getType(),
                notification.getType(), notification);
        return message;
    }

    /**
     * 计算私聊消息回执 status：私聊已读关闭时强制 NO_RECEIPT，忽略发送方传入的 receipt（receipt 为 null 等价 false）
     */
    private Integer resolveReceiptStatus(Boolean receipt) {
        if (BooleanUtil.isFalse(imProperties.getMessage().isPrivateReadEnabled())) {
            return ImMessageReceiptStatusEnum.NO_RECEIPT.getStatus();
        }
        return BooleanUtil.isTrue(receipt)
                ? ImMessageReceiptStatusEnum.PENDING.getStatus()
                : ImMessageReceiptStatusEnum.NO_RECEIPT.getStatus();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public ImPrivateMessageDO recallPrivateMessage(Long userId, Long messageId) {
        // 1.1 校验消息存在
        ImPrivateMessageDO message = privateMessageMapper.selectById(messageId);
        if (message == null) {
            throw exception(MESSAGE_NOT_EXISTS);
        }
        // 1.2 只能撤回自己发送的消息
        if (ObjUtil.notEqual(message.getSenderId(), userId)) {
            throw exception(MESSAGE_RECALL_DENIED);
        }
        // 1.3 不能重复撤回
        if (ImMessageStatusEnum.RECALL.getStatus().equals(message.getStatus())) {
            throw exception(MESSAGE_ALREADY_RECALLED);
        }
        // 1.4 只允许撤回限定时间内的消息
        int recallTimeoutMinutes = imProperties.getMessage().getRecallTimeoutMinutes();
        if (message.getSendTime().plusMinutes(recallTimeoutMinutes).isBefore(LocalDateTime.now())) {
            throw exception(MESSAGE_RECALL_TIMEOUT, recallTimeoutMinutes);
        }

        // 2. 更新原消息状态为撤回
        privateMessageMapper.updateById(new ImPrivateMessageDO().setId(messageId)
                .setStatus(ImMessageStatusEnum.RECALL.getStatus()));

        // 3. 发送撤回事件
        return sendPrivateMessage(userId, new ImPrivateMessageSendDTO().setReceiverId(message.getReceiverId())
                .setType(ImContentTypeEnum.RECALL.getType()).setContent(new RecallMessage().setMessageId(messageId)));
    }

    /**
     * 私聊引用消息规范化
     *
     * @param reqVO 发送请求
     * @param senderId 发送人编号
     * @return 规范化后的 content
     */
    private String normalizeQuoteContent(ImPrivateMessageSendReqVO reqVO, Long senderId) {
        // 解析客户端 content 里的 quote.messageId
        Long quoteMessageId = ImMessageUtils.parseQuoteMessageId(reqVO.getContent());

        // 情况一：没有 quoteMessageId，直接 remove 掉 content 里可能伪造的 quote 字段
        if (quoteMessageId == null) {
            return ImMessageUtils.removeQuote(reqVO.getContent());
        }

        // 情况二：有 quoteMessageId，加载原消息并校验
        ImPrivateMessageDO original = privateMessageMapper.selectById(quoteMessageId);
        if (original == null
                || ImMessageStatusEnum.RECALL.getStatus().equals(original.getStatus())) {
            throw exception(MESSAGE_QUOTE_INVALID);
        }
        // 校验是同对话
        boolean sameConversation = (ObjUtil.equal(original.getSenderId(), senderId) // 发送人是当前用户，接收人是对方
                && ObjUtil.equal(original.getReceiverId(), reqVO.getReceiverId()))
                || (ObjUtil.equal(original.getSenderId(), reqVO.getReceiverId()) // 发送人是对方，接收人是当前用户
                        && ObjUtil.equal(original.getReceiverId(), senderId));
        if (!sameConversation) {
            throw exception(MESSAGE_QUOTE_INVALID);
        }
        // 构建 quote 对象并注入 content
        QuoteMessage quote = ImMessageUtils.buildQuote(original.getId(),
                original.getSenderId(), original.getType(), original.getContent());
        return ImMessageUtils.appendQuote(reqVO.getContent(), quote);
    }

    @Override
    public List<ImPrivateMessageDO> pullPrivateMessageList(Long userId, Long minId, Integer size) {
        int maxPullSize = imProperties.getMessage().getMaxPullSize();
        if (size > maxPullSize) {
            throw exception(MESSAGE_PULL_SIZE_EXCEEDED, maxPullSize);
        }
        // 0. 拉取时间窗；超过窗口的老消息不再通过离线通道推送
        LocalDateTime minSendTime = LocalDateTime.now().minusDays(imProperties.getMessage().getPrivatePullMaxDays());

        // 根据 minId 和 minSendTime 拉取消息，避免 minId 恰好被发出后才拉取，导致漏消息
        List<ImPrivateMessageDO> messages = privateMessageMapper.selectListByMinId(userId, minId, minSendTime, size);
        log.info("[pullPrivateMessageList][userId({}) minId({}) size({}) result({})]",
                userId, minId, size, messages.size());
        return messages;
    }

    @Override
    public void readPrivateMessages(Long userId, Long receiverId, Long messageId) {
        // 1. 全局开关校验
        if (BooleanUtil.isFalse(imProperties.getMessage().isPrivateReadEnabled())) {
            throw exception(MESSAGE_PRIVATE_READ_DISABLED);
        }
        Assert.notNull(messageId, "已读消息编号不能为空");

        // 2. 回执置 DONE：把 (receiverId → userId) 上 id <= messageId、待回执(PENDING) 的消息标记为已完成
        // status 不再表达已读（保持 NORMAL）；只翻 PENDING 行，避免覆盖 NO_RECEIPT / 已 DONE
        privateMessageMapper.updateBySenderIdAndReceiverIdAndIdLeAndReceiptStatus(
                receiverId, userId, messageId, ImMessageReceiptStatusEnum.PENDING.getStatus(),
                new ImPrivateMessageDO().setReceiptStatus(ImMessageReceiptStatusEnum.DONE.getStatus()));

        // 3. 同步写 im_conversation_read（读位置唯一权威，单调递增）；读位置前进才下发事件
        boolean advanced = conversationReadService.updateConversationReadPosition(
                userId, ImConversationTypeEnum.PRIVATE.getType(), receiverId, messageId);
        if (!advanced) {
            return;
        }

        // 4. 异步发送 READ + RECEIPT 事件（已读位置以前端上报为准，与多端 / 对方 UI 显示一致）
        imWebSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.PRIVATE.getType(),
                ImContentTypeEnum.READ.getType(), ImMessageReadNotification.ofPrivate(userId, receiverId, messageId));
        imWebSocketService.sendNotificationAsync(receiverId, ImConversationTypeEnum.PRIVATE.getType(),
                ImContentTypeEnum.RECEIPT.getType(),
                ImMessageReceiptNotification.ofPrivate(userId, receiverId, messageId));
    }

    @Override
    public Long getMaxReadMessageId(Long userId, Long peerId) {
        if (BooleanUtil.isFalse(imProperties.getMessage().isPrivateReadEnabled())) {
            throw exception(MESSAGE_PRIVATE_READ_DISABLED);
        }
        // 对端 peer 在「与 userId 的会话」里的读位置 = peer 把 userId 发的消息读到哪
        return conversationReadService.getConversationReadMessageId(
                peerId, ImConversationTypeEnum.PRIVATE.getType(), userId);
    }

    @Override
    public List<ImPrivateMessageDO> getPrivateMessageList(Long userId, ImPrivateMessageListReqVO reqVO) {
        return privateMessageMapper.selectHistoryList(userId, reqVO.getReceiverId(), reqVO.getMaxId(), reqVO.getLimit());
    }

    // ==================== 管理后台 ====================

    @Override
    public PageResult<ImPrivateMessageDO> getPrivateMessagePage(ImPrivateMessageManagerPageReqVO reqVO) {
        return privateMessageMapper.selectPage(reqVO);
    }

    @Override
    public ImPrivateMessageDO getPrivateMessage(Long id) {
        return privateMessageMapper.selectById(id);
    }

}