package com.imutil.service.impl;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.imutil.common.MsgKeys;
import com.imutil.common.TenantContext;
import com.imutil.entity.DistQueue;
import com.imutil.entity.ImMessage;
import com.imutil.entity.PullWatermark;
import com.imutil.entity.Tenant;
import com.imutil.mapper.DistQueueMapper;
import com.imutil.mapper.ImMessageMapper;
import com.imutil.mapper.PullWatermarkMapper;
import com.imutil.service.PullService;
import com.imutil.service.TenantService;
import com.imutil.tencent.TencentImClient;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack4.ONode;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;

import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.util.List;

/**
 * 历史消息补拉服务实现
 * <p>
 * 流程：取最久未补拉的 N 个会话 → 按会话类型调腾讯 API 拉最新消息 →
 * 逐条 msg_key 幂等（与回调路径共享 {@link MsgKeys} 算法）→ 本地不存在则落库(source=PULL_BACK)+入队分发 → 推进水位线。
 * <p>
 * 仅"本次新插入"的消息入队分发，避免对回调已正常落库的消息重复推送业务系统。
 * 每轮串行执行 + 会话数/条数上限，受腾讯 API QPS 约束。
 * <p>
 * C2C 会话需 from+to 配对，水位线仅存对端(conv_id)，本端取该会话最近一条消息的 from_account；
 * 多发送方场景仅覆盖最近一个 from（已知限制，由腾讯 get_roam_msg 成对特性决定）。
 *
 * @author imutil
 */
@Slf4j
@Component
public class PullServiceImpl implements PullService {

    @Inject
    private PullWatermarkMapper pullWatermarkMapper;

    @Inject
    private ImMessageMapper imMessageMapper;

    @Inject
    private DistQueueMapper distQueueMapper;

    @Inject
    private TencentImClient tencentImClient;

    @Inject
    private TenantService tenantService;

    @Inject("${imutil.pull.convsPerRound:20}")
    private int convsPerRound;

    @Inject("${imutil.pull.maxMsgPerConv:20}")
    private int maxMsgPerConv;

    @Inject("${imutil.pull.lookbackMinutes:30}")
    private int lookbackMinutes;

    @Override
    public void pullRound() {
        List<PullWatermark> convs = pullWatermarkMapper.selectNeedCheck(convsPerRound);
        if (convs.isEmpty()) {
            return;
        }
        int totalNew = 0;
        for (PullWatermark wm : convs) {
            try {
                totalNew += pullOne(wm);
            } catch (Throwable e) {
                // 单会话异常不影响其他会话；仍推进 updated_at 避免该会话一直排在前
                log.warn("补拉会话异常 tenant={} conv={} : {}", wm.getTenantId(), wm.getConvId(), e.toString());
                advanceWatermark(wm, 0, null);
            }
        }
        log.info("补拉一轮完成 会话数={} 新增消息={}", convs.size(), totalNew);
    }

    /**
     * 补拉单个会话：设置租户上下文（供腾讯 API 审计）→ 按会话类型分发
     */
    private int pullOne(PullWatermark wm) {
        TenantContext.set(wm.getTenantId());
        try {
            if (wm.getConvType() != null && wm.getConvType() == 2) {
                return pullGroup(wm);
            }
            return pullC2C(wm);
        } finally {
            TenantContext.clear();
        }
    }

    /**
     * 补拉群消息：getGroupMsg 取最新 N 条
     */
    private int pullGroup(PullWatermark wm) {
        String resp = tencentImClient.getGroupMsg(wm.getConvId(), maxMsgPerConv);
        ONode list = extractMsgList(resp, "RspMsgList");
        if (list == null) {
            advanceWatermark(wm, 0, null);
            return 0;
        }
        return saveMessages(wm, list, true);
    }

    /**
     * 补拉单聊消息：getRoamMsg 按时间窗拉取，本端取该会话最近 from
     */
    private int pullC2C(PullWatermark wm) {
        String to = wm.getConvId();
        String from = latestFrom(wm.getTenantId(), to);
        if (from == null) {
            // 无历史记录无法确定本端账号，跳过本轮
            advanceWatermark(wm, 0, null);
            return 0;
        }
        long maxInterval = lookbackMinutes * 60L;
        long minTime = Instant.now().getEpochSecond() - maxInterval;
        String resp = tencentImClient.getRoamMsg(from, to, maxMsgPerConv, minTime, maxInterval);
        ONode list = extractMsgList(resp, "MsgList");
        if (list == null) {
            advanceWatermark(wm, 0, null);
            return 0;
        }
        return saveMessages(wm, list, false);
    }

    /**
     * 逐条幂等落库 + 入队分发，返回新增条数
     *
     * @param wm      会话水位线
     * @param list    消息数组节点
     * @param isGroup 是否群消息（决定解析方式）
     */
    private int saveMessages(PullWatermark wm, ONode list, boolean isGroup) {
        long maxSeq = 0;
        OffsetDateTime maxTime = null;
        int newCount = 0;
        for (int i = 0, sz = list.size(); i < sz; i++) {
            ONode m = list.get(i);
            ImMessage msg = isGroup ? parseGroupMsg(m, wm.getTenantId()) : parseC2CMsg(m, wm.getTenantId());
            if (msg == null) {
                continue;
            }
            // 跟踪本轮最大 seq/time 用于推进水位线
            long seq = m.get("MsgSeq").getLong();
            if (seq > maxSeq) {
                maxSeq = seq;
            }
            if (msg.getMsgTime() != null && (maxTime == null || msg.getMsgTime().isAfter(maxTime))) {
                maxTime = msg.getMsgTime();
            }
            // 幂等：本地已存在则跳过（不重复分发）
            long exists = imMessageMapper.selectCount(Wrappers.<ImMessage>lambdaQuery()
                    .eq(ImMessage::getMsgKey, msg.getMsgKey()));
            if (exists > 0) {
                continue;
            }
            imMessageMapper.insert(msg);
            newCount++;
            // 仅新插入的消息入队分发
            enqueue(wm.getTenantId(), msg);
        }
        advanceWatermark(wm, maxSeq, maxTime);
        return newCount;
    }

    /**
     * 写入分发队列（复用回调分发链路，payload 用结构化 JSON）
     */
    private void enqueue(String tenantId, ImMessage msg) {
        Tenant t = tenantService.getById(tenantId);
        if (t == null || t.getCallbackUrl() == null || t.getCallbackUrl().isEmpty()) {
            return;
        }
        ONode payload = ONode.ofJson("{}");
        payload.set("source", msg.getSource());
        payload.set("msgKey", msg.getMsgKey());
        payload.set("tenantId", tenantId);
        payload.set("convType", msg.getConvType());
        payload.set("convId", msg.getConvId());
        payload.set("fromAccount", msg.getFromAccount());
        payload.set("toAccount", msg.getToAccount());
        payload.set("groupId", msg.getGroupId());
        payload.set("msgTime", msg.getMsgTime());
        payload.set("msgType", msg.getMsgType());
        if (msg.getMsgBody() != null) {
            try {
                payload.set("msgBody", ONode.ofJson(msg.getMsgBody()));
            } catch (Exception e) {
                payload.set("msgBody", msg.getMsgBody());
            }
        }
        DistQueue q = new DistQueue();
        q.setMsgKey(msg.getMsgKey());
        q.setTenantId(tenantId);
        q.setConvId(msg.getConvId());
        q.setTargetUrl(t.getCallbackUrl());
        q.setPayload(payload.toJson());
        q.setStatus(0);
        q.setRetryCount(0);
        q.setNextRetryAt(OffsetDateTime.now());
        distQueueMapper.insert(q);
    }

    /**
     * 解析群消息节点为 ImMessage
     */
    private ImMessage parseGroupMsg(ONode m, String tenantId) {
        String groupId = m.get("GroupId").getString();
        if (groupId == null || groupId.isEmpty()) {
            return null;
        }
        long msgSeq = m.get("MsgSeq").getLong();
        long msgRandom = m.get("MsgRandom").getLong();
        long msgTs = m.get("MsgTimeStamp").getLong();
        String from = m.get("From_Account").getString();
        ONode body = m.get("MsgBody");
        ImMessage msg = new ImMessage();
        msg.setMsgKey(MsgKeys.build(from, groupId, msgSeq, msgRandom));
        msg.setTenantId(tenantId);
        msg.setMsgTime(toTime(msgTs));
        msg.setConvType(2);
        msg.setConvId(groupId);
        msg.setFromAccount(from);
        msg.setGroupId(groupId);
        msg.setMsgType(firstBodyType(body));
        msg.setMsgBody(body.toString());
        msg.setSource("PULL_BACK");
        msg.setIsCrossTenant(false);
        msg.setDistStatus(0);
        return msg;
    }

    /**
     * 解析单聊消息节点为 ImMessage
     */
    private ImMessage parseC2CMsg(ONode m, String tenantId) {
        long msgSeq = m.get("MsgSeq").getLong();
        long msgRandom = m.get("MsgRandom").getLong();
        long msgTs = m.get("MsgTimeStamp").getLong();
        String from = m.get("From_Account").getString();
        String to = m.get("To_Account").getString();
        ONode body = m.get("MsgBody");
        ImMessage msg = new ImMessage();
        msg.setMsgKey(MsgKeys.build(from, to, msgSeq, msgRandom));
        msg.setTenantId(tenantId);
        msg.setMsgTime(toTime(msgTs));
        msg.setConvType(1);
        msg.setConvId(to != null ? to : "");
        msg.setFromAccount(from);
        msg.setToAccount(to);
        msg.setMsgType(firstBodyType(body));
        msg.setMsgBody(body.toString());
        msg.setSource("PULL_BACK");
        msg.setIsCrossTenant(false);
        msg.setDistStatus(0);
        return msg;
    }

    /**
     * 推进水位线（last_seq/last_time 取本轮最大，updated_at 始终刷新驱动轮询顺序）
     */
    private void advanceWatermark(PullWatermark wm, long maxSeq, OffsetDateTime maxTime) {
        int convType = wm.getConvType() == null ? 1 : wm.getConvType();
        pullWatermarkMapper.upsert(wm.getTenantId(), wm.getConvId(), convType, maxSeq, maxTime, OffsetDateTime.now());
    }

    /**
     * 查该 C2C 会话最近发送方（配对 get_roam_msg 的本端账号）
     */
    private String latestFrom(String tenantId, String to) {
        ImMessage latest = imMessageMapper.selectOne(Wrappers.<ImMessage>lambdaQuery()
                .eq(ImMessage::getTenantId, tenantId)
                .eq(ImMessage::getConvId, to)
                .isNotNull(ImMessage::getFromAccount)
                .ne(ImMessage::getFromAccount, "")
                .orderByDesc(ImMessage::getMsgTime)
                .last("LIMIT 1"));
        return latest == null ? null : latest.getFromAccount();
    }

    /**
     * 解析腾讯响应，校验 ActionStatus=OK 并返回消息数组节点；失败/空返回 null
     *
     * @param resp    腾讯原始响应 JSON
     * @param listKey 消息列表字段名（C2C=MsgList，GROUP=RspMsgList）
     */
    private ONode extractMsgList(String resp, String listKey) {
        if (resp == null || resp.isEmpty()) {
            return null;
        }
        try {
            ONode root = ONode.ofJson(resp);
            if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
                log.warn("补拉腾讯API返回非OK : {}", resp);
                return null;
            }
            ONode list = root.get(listKey);
            return list.isArray() ? list : null;
        } catch (Exception e) {
            log.warn("补拉响应解析失败 : {}", e.toString());
            return null;
        }
    }

    /**
     * 取消息体首个元素的类型（MsgBody 为数组，如 TIMTextElem/TIMImageElem）
     */
    private String firstBodyType(ONode body) {
        if (body == null || !body.isArray() || body.size() == 0) {
            return null;
        }
        return body.get(0).get("MsgType").getString();
    }

    /**
     * 秒级时间戳转 OffsetDateTime（0/负数兜底为当前时间）
     */
    private OffsetDateTime toTime(long epochSecond) {
        if (epochSecond <= 0) {
            return OffsetDateTime.now();
        }
        return OffsetDateTime.ofInstant(Instant.ofEpochSecond(epochSecond), ZoneId.systemDefault());
    }
}