fix

netflix-dao/src/main/java/com/cyksj/mapper/corp/kf/CorpKfAccountSyncMsgProgressMapper.java

@@ -0,0 +1,13 @@
+package com.cyksj.mapper.corp.kf;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.cyksj.model.entity.CorpKfAccountSyncMsgProgress;
+
+/**
+ * 项目名: yhlxj2
+ * 文件名: CorpKfAccountSyncMsgProgressMapper
+ * 创建者: JavaZou
+ * 创建时间:2025/1/2 11:54
+ */
+public interface CorpKfAccountSyncMsgProgressMapper extends BaseMapper<CorpKfAccountSyncMsgProgress> {
+}

netflix-dao/src/main/java/com/cyksj/model/entity/CorpKfAccountSyncMsgProgress.java

@@ -0,0 +1,21 @@
+package com.cyksj.model.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * 项目名: yhlxj2
+ * 文件名: CorpKfAccountSyncMsgProgress
+ * 创建者: JavaZou
+ * 创建时间:2025/1/2 11:53
+ */
+@Getter
+@Setter
+public class CorpKfAccountSyncMsgProgress extends BaseEntity{
+	private String openKfId;
+
+	/**
+	 * 上一次调用时返回的next_cursor
+	 */
+	private String cursor;
+}

netflix-dao/src/main/java/com/cyksj/model/kf/WxCpKfMsgListResp.java

@@ -22,6 +22,9 @@ public class WxCpKfMsgListResp {
   @JsonProperty("next_cursor")
   private String nextCursor;
 
+  /**
+   * 是否还有更多数据。0-否；1-是。
+   */
   @JsonProperty("has_more")
   private Integer hasMore;
 

netflix-service/src/main/java/com/cyksj/service/corp/impl/CorpServiceImpl.java

@@ -24,10 +24,7 @@ import com.cyksj.mapper.corp.kf.*;
 import com.cyksj.model.dto.CorpKfServicerDto;
 import com.cyksj.model.dto.WxCorpEncryptDto;
 import com.cyksj.model.entity.*;
-import com.cyksj.model.kf.CorpKfServicerResp;
-import com.cyksj.model.kf.WxCorpKfMsgSendRequest;
-import com.cyksj.model.kf.WxCpKfEventMsg;
-import com.cyksj.model.kf.WxCpKfMsgListResp;
+import com.cyksj.model.kf.*;
 import com.cyksj.model.request.corp.Attachments;
 import com.cyksj.model.request.corp.CorpXmlMessage;
 import com.cyksj.model.request.corp.CorpXmlOutMessage;
@@ -50,10 +47,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.PostConstruct;
 import java.io.File;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
@@ -122,6 +116,9 @@ public class CorpServiceImpl implements CorpService {
 	@Autowired
 	private CorpKfServicerSessionRecordMapper corpKfServicerSessionRecordMapper;
 
+	@Autowired
+	private CorpKfAccountSyncMsgProgressMapper corpKfAccountSyncMsgProgressMapper;
+
 	@PostConstruct
 	private void init() {
 		corpMessageRouter = new CorpMessageRouter(this)
@@ -531,16 +528,33 @@ public class CorpServiceImpl implements CorpService {
 	private CorpXmlOutMessage kfMsgOrEvent(CorpXmlMessage message, Map<String, Object> context) {
 		try {
 			String openKfId = message.getOpenKfId();
-			WxCpKfMsgListResp wxCpKfMsgListResp = wxCorpOps.syncMsg(message.getToUserName(), null, message.getToken(), null, openKfId);
-			List<WxCpKfMsgListResp.WxCpKfMsgItem> msgList = wxCpKfMsgListResp.getMsgList();
-			//进入会话事件
-			enterSessionEvent(message, msgList);
-			//会话更变事件
-			sessionStatusChange(message, msgList);
-			//消息发送失败事件
-			//拒收客户消息变更事件
-			//同步客服聊天记录
-			syncChatMsg(message, msgList);
+			CorpKfAccount corpKfAccount = corpKfAccountMapper.selectOne(Wrappers.lambdaQuery(CorpKfAccount.class)
+					.eq(CorpKfAccount::getOpenKfId, openKfId)
+					.last("limit 1"));
+			if (corpKfAccount == null) {
+				log.info("客服账号:{}已被删除或不存在", openKfId);
+				return null;
+			}
+			//账号会话设置
+			CorpKfAccountSessionConfig corpKfAccountSessionConfig = corpKfAccountSessionConfigMapper.selectOne(Wrappers.lambdaQuery(CorpKfAccountSessionConfig.class)
+					.eq(CorpKfAccountSessionConfig::getOpenKfId, openKfId)
+					.last("limit 1"));
+			//同步消息进度
+			CorpKfAccountSyncMsgProgress corpKfAccountSyncMsgProgress = Optional.ofNullable(corpKfAccountSyncMsgProgressMapper.selectOne(Wrappers.lambdaQuery(CorpKfAccountSyncMsgProgress.class)
+					.eq(CorpKfAccountSyncMsgProgress::getOpenKfId, openKfId)
+					.last("limit 1"))).orElse(new CorpKfAccountSyncMsgProgress());
+			String cursor = null;
+			if (corpKfAccountSyncMsgProgress.getCursor() != null) {
+				cursor = corpKfAccountSyncMsgProgress.getCursor();
+			}
+			String corpId = message.getToUserName();
+			String token = message.getToken();
+			syncCorpKfMsg(corpKfAccount, corpKfAccountSessionConfig, corpId, cursor, token, null, corpKfAccountSyncMsgProgress);
+			corpKfAccountSyncMsgProgress.setOpenKfId(openKfId);
+			//同步记录
+			if (corpKfAccountSyncMsgProgress.getId() == null) {
+				corpKfAccountSyncMsgProgressMapper.insert(corpKfAccountSyncMsgProgress);
+			} else corpKfAccountSyncMsgProgressMapper.updateById(corpKfAccountSyncMsgProgress);
 		} catch (Exception e) {
 			log.error("企业微信客服 事件消息回调 错误");
 			throw BusinessRuntimeException.getInstance(StringUtil.getErrorMsg(e));
@@ -548,10 +562,52 @@ public class CorpServiceImpl implements CorpService {
 		return null;
 	}
 
-	private void sessionStatusChange(CorpXmlMessage message, List<WxCpKfMsgListResp.WxCpKfMsgItem> msgList) throws Exception {
-		String openKfId = message.getOpenKfId();
-		List<WxCpKfMsgListResp.WxCpKfMsgItem> sessionStatusChanges = msgList.stream()
-				.filter((item) -> item.getEvent() != null && item.getEvent().getEventType().equals("session_status_change")).collect(Collectors.toList());
+	private void syncCorpKfMsg(CorpKfAccount corpKfAccount, CorpKfAccountSessionConfig corpKfAccountSessionConfig, String corpId, String cursor, String token, Integer limit, CorpKfAccountSyncMsgProgress corpKfAccountSyncMsgProgress) throws Exception {
+		String openKfId = corpKfAccount.getOpenKfId();
+		WxCpKfMsgListResp wxCpKfMsgListResp = wxCorpOps.syncMsg(corpId, cursor, token, limit, openKfId);
+		List<WxCpKfMsgListResp.WxCpKfMsgItem> msgList = wxCpKfMsgListResp.getMsgList();
+		List<WxCpKfMsgListResp.WxCpKfMsgItem> enterSessionList = null;
+		List<WxCpKfMsgListResp.WxCpKfMsgItem> sessionStatusChanges = null;
+		List<WxCorpChatMsg> mergedMsgs = null;
+		for (WxCpKfMsgListResp.WxCpKfMsgItem item : msgList) {
+			if (item.getEvent() != null && item.getEvent().getEventType().equals("enter_session")) {
+				//会话事件
+				if (item.getEvent().getEventType().equals("enter_session")) {
+					if (enterSessionList == null) {
+						enterSessionList = new ArrayList<>();
+					}
+					enterSessionList.add(item);
+				}
+				//会话变更事件
+				if (item.getEvent().getEventType().equals("session_status_change")) {
+					if (sessionStatusChanges == null) {
+						sessionStatusChanges = new ArrayList<>();
+					}
+					sessionStatusChanges.add(item);
+				}
+			}
+			if (item.getMergedMsg() != null) {
+				if (mergedMsgs == null) {
+					mergedMsgs = new ArrayList<>();
+				}
+				mergedMsgs.add(item.getMergedMsg());
+			}
+		}
+		//进入会话事件
+		enterSessionEvent(corpKfAccount, corpKfAccountSessionConfig, corpId, enterSessionList);
+		//会话更变事件
+		sessionStatusChange(corpId, openKfId, sessionStatusChanges);
+		//消息发送失败事件
+		//拒收客户消息变更事件
+		//同步客服聊天记录
+		syncChatMsg(mergedMsgs);
+		corpKfAccountSyncMsgProgress.setCursor(cursor);
+		while (wxCpKfMsgListResp.getHasMore() == 1) {
+			syncCorpKfMsg(corpKfAccount, corpKfAccountSessionConfig, corpId, wxCpKfMsgListResp.getNextCursor(), token, limit, corpKfAccountSyncMsgProgress);
+		}
+	}
+
+	private void sessionStatusChange(String corpId, String openKfId, List<WxCpKfMsgListResp.WxCpKfMsgItem> sessionStatusChanges) throws Exception {
 		if (CollUtil.isNotEmpty(sessionStatusChanges)) {
 			WxCpKfMsgListResp.WxCpKfMsgItem wxCpKfMsgItem = sessionStatusChanges.get(sessionStatusChanges.size() - 1);
 			String externalUserId = wxCpKfMsgItem.getExternalUserId();
@@ -575,7 +631,7 @@ public class CorpServiceImpl implements CorpService {
 					.last("limit 1"));
 			if (corpKfServicerSessionRecord1 == null) {
 				//记录新的客服会话
-				transCorpKfSessionToServicer(message.getToUserName(), openKfId, newServicerUserid, externalUserId);
+				transCorpKfSessionToServicer(corpId, openKfId, newServicerUserid, externalUserId);
 			}
 		}
 	}
@@ -583,21 +639,8 @@ public class CorpServiceImpl implements CorpService {
 	/**
 	 * enter_session 会话事件
 	 */
-	public void enterSessionEvent(CorpXmlMessage message, List<WxCpKfMsgListResp.WxCpKfMsgItem> msgList) throws Exception {
-		String openKfId = message.getOpenKfId();
-		List<WxCpKfMsgListResp.WxCpKfMsgItem> enterSessionList = msgList.stream()
-				.filter((item) -> item.getEvent() != null && item.getEvent().getEventType().equals("enter_session")).collect(Collectors.toList());
-		CorpKfAccount corpKfAccount = corpKfAccountMapper.selectOne(Wrappers.lambdaQuery(CorpKfAccount.class)
-				.eq(CorpKfAccount::getOpenKfId, openKfId)
-				.last("limit 1"));
-		if (corpKfAccount == null) {
-			log.info("客服账号:{}已被删除或不存在", openKfId);
-			return;
-		}
-		//账号会话设置
-		CorpKfAccountSessionConfig corpKfAccountSessionConfig = corpKfAccountSessionConfigMapper.selectOne(Wrappers.lambdaQuery(CorpKfAccountSessionConfig.class)
-				.eq(CorpKfAccountSessionConfig::getOpenKfId, openKfId)
-				.last("limit 1"));
+	public void enterSessionEvent(CorpKfAccount corpKfAccount, CorpKfAccountSessionConfig corpKfAccountSessionConfig, String corpId, List<WxCpKfMsgListResp.WxCpKfMsgItem> enterSessionList) throws Exception {
+		String openKfId = corpKfAccount.getOpenKfId();
 		if (enterSessionList.size() > 0) {
 			WxCpKfMsgListResp.WxCpKfMsgItem wxCpKfMsgItem = enterSessionList.get(enterSessionList.size() - 1);
 			WxCpKfEventMsg event = wxCpKfMsgItem.getEvent();
@@ -621,7 +664,7 @@ public class CorpServiceImpl implements CorpService {
 				wxCpKfMsgSendRequest.setText(textMsg);
 				wxCpKfMsgSendRequest.setToUser(welcomeCode);
 				wxCpKfMsgSendRequest.setOpenKfid(event.getOpenKfid());
-				wxCorpOps.sendMsgOnEvent(message.getToUserName(), wxCpKfMsgSendRequest);
+				wxCorpOps.sendMsgOnEvent(corpId, wxCpKfMsgSendRequest);
 			} else {
 				//用户是否已有会话
 				Integer selectCount = corpKfServicerSessionRecordMapper.selectCount(Wrappers.lambdaQuery(CorpKfServicerSessionRecord.class)
@@ -640,11 +683,11 @@ public class CorpServiceImpl implements CorpService {
 						.le(CorpKfAccountFollow::getStartTime, hour)
 						.ge(CorpKfAccountFollow::getEndTime, hour));
 				//获取接待人员列表
-				CorpKfServicerResp corpKfServicerResp = wxCorpOps.getKfServicerList(message.getToUserName(), openKfId);
+				CorpKfServicerResp corpKfServicerResp = wxCorpOps.getKfServicerList(corpId, openKfId);
 				List<CorpKfServicerResp.CorpKfServicer> servicerList = corpKfServicerResp.getServicerList();
 				//目前值班客服id
 				List<String> corpKfServiceIds = new ArrayList<>();
-				List<CorpKfServicerResp.CorpKfServicer> corpKfServicers = servicerList.stream().filter(servicer ->{
+				List<CorpKfServicerResp.CorpKfServicer> corpKfServicers = servicerList.stream().filter(servicer -> {
 					if (servicer.getUserid() != null && servicer.getStatus() == 0) {
 						corpKfServiceIds.add(servicer.getUserid());
 						return true;
@@ -653,11 +696,11 @@ public class CorpServiceImpl implements CorpService {
 				}).collect(Collectors.toList());
 				if (CollUtil.isEmpty(corpKfServicers)) {
 					//非值班 移入等待池并发送非值班提示消息
-					moveWaitingQueue(message.getToUserName(), openKfId, event.getExternalUserId());
+					moveWaitingQueue(corpId, openKfId, event.getExternalUserId());
 					String noDutyTips = corpKfAccountSessionConfig.getNoDutyTips();
 					if (StrUtil.isNotEmpty(noDutyTips)) {
 						//发送提示
-						sendCorpKfTextTips(message.getToUserName(), event.getExternalUserId(), openKfId, noDutyTips);
+						sendCorpKfTextTips(corpId, event.getExternalUserId(), openKfId, noDutyTips);
 					}
 					return;
 				}
@@ -668,7 +711,7 @@ public class CorpServiceImpl implements CorpService {
 					//随机分配一个客服
 					CorpKfServicerResp.CorpKfServicer corpKfServicer = corpKfServicers.get(RandomUtil.randomInt(corpKfServicers.size()));
 					//变更会话状态
-					transCorpKfSessionToServicer(message.getToUserName(), openKfId, corpKfServicer.getUserid(), event.getExternalUserId());
+					transCorpKfSessionToServicer(corpId, openKfId, corpKfServicer.getUserid(), event.getExternalUserId());
 				}
 				//筛选符合条件的客服
 				//默认接入最少的优先分配 接待人最少的在集合前面
@@ -677,15 +720,15 @@ public class CorpServiceImpl implements CorpService {
 				List<CorpKfServicerDto> filterKfCs = corpKfServicerDtos.stream().filter(kfCs -> corpKfServiceIds.contains(kfCs.getServicerUserid()) && kfCs.getNum() < maxNum).collect(Collectors.toList());
 				if (CollUtil.isEmpty(filterKfCs)) {
 					//未有可接待的客服 移入等待池并发送进入等待池消息
-					moveWaitingQueue(message.getToUserName(), openKfId, event.getExternalUserId());
+					moveWaitingQueue(corpId, openKfId, event.getExternalUserId());
 					String queueTips = corpKfAccountSessionConfig.getQueueTips();
-					sendCorpKfTextTips(message.getToUserName(), event.getExternalUserId(), openKfId, queueTips);
+					sendCorpKfTextTips(corpId, event.getExternalUserId(), openKfId, queueTips);
 					return;
 				}
 				//选择第一个分配接待
 				CorpKfServicerDto corpKfServicerDto = filterKfCs.get(0);
 				//变更会话状态
-				transCorpKfSessionToServicer(message.getToUserName(), openKfId, corpKfServicerDto.getServicerUserid(), event.getExternalUserId());
+				transCorpKfSessionToServicer(corpId, openKfId, corpKfServicerDto.getServicerUserid(), event.getExternalUserId());
 			}
 		}
 	}