|
|
@@ -0,0 +1,177 @@
|
|
|
+package com.storlead.trade.service.impl;
|
|
|
+
|
|
|
+import cn.hutool.core.util.StrUtil;
|
|
|
+import com.baomidou.dynamic.datasource.annotation.DS;
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
+import com.storlead.framework.common.constant.CommonConstant;
|
|
|
+import com.storlead.framework.common.constant.DSConstants;
|
|
|
+import com.storlead.mail.pojo.entity.MktMailMessageEntity;
|
|
|
+import com.storlead.mail.service.MktMailMessageService;
|
|
|
+import com.storlead.trade.entity.MktMailInboxCursorEntity;
|
|
|
+import com.storlead.trade.enums.TradeMessageTypeEnum;
|
|
|
+import com.storlead.trade.service.MktMailInboxCursorService;
|
|
|
+import com.storlead.trade.service.MktMailNewNotifyService;
|
|
|
+import com.storlead.trade.service.TradeMessageAssembleService;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.util.CollectionUtils;
|
|
|
+
|
|
|
+import javax.annotation.Resource;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.HashSet;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.Set;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 用游标表记录收件箱未读最大 id,按 id 增量判断是否有新邮件后发提醒。
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Service
|
|
|
+@DS(DSConstants.DATASOURCE_TRADE)
|
|
|
+public class MktMailNewNotifyServiceImpl implements MktMailNewNotifyService {
|
|
|
+
|
|
|
+ private static final int BATCH_LIMIT = 200;
|
|
|
+ private static final int MAX_PAGES = 50;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private MktMailMessageService mktMailMessageService;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private MktMailInboxCursorService mktMailInboxCursorService;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private TradeMessageAssembleService tradeMessageAssembleService;
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public synchronized int syncOnce() {
|
|
|
+ MktMailInboxCursorEntity cursor = currentCursor();
|
|
|
+ if (cursor == null) {
|
|
|
+ initCursorWithoutNotify();
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ Long lastId = cursor.getLastMailId() == null ? 0L : cursor.getLastMailId();
|
|
|
+ int notified = 0;
|
|
|
+ Long maxId = lastId;
|
|
|
+ for (int page = 0; page < MAX_PAGES; page++) {
|
|
|
+ List<MktMailMessageEntity> newMails;
|
|
|
+ try {
|
|
|
+ newMails = mktMailMessageService.listNewInboxMails(lastId, BATCH_LIMIT);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("查询营销收件箱新邮件失败 lastId={}", lastId, e);
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (CollectionUtils.isEmpty(newMails)) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ Long batchLastId = newMails.get(newMails.size() - 1).getId();
|
|
|
+ notified += notifyByOwner(newMails);
|
|
|
+ if (batchLastId != null) {
|
|
|
+ lastId = batchLastId;
|
|
|
+ maxId = batchLastId;
|
|
|
+ }
|
|
|
+ if (newMails.size() < BATCH_LIMIT) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (maxId != null && maxId > (cursor.getLastMailId() == null ? 0L : cursor.getLastMailId())) {
|
|
|
+ cursor.setLastMailId(maxId);
|
|
|
+ mktMailInboxCursorService.updateById(cursor);
|
|
|
+ }
|
|
|
+ return notified;
|
|
|
+ }
|
|
|
+
|
|
|
+ private MktMailInboxCursorEntity currentCursor() {
|
|
|
+ return mktMailInboxCursorService.getOne(new LambdaQueryWrapper<MktMailInboxCursorEntity>()
|
|
|
+ .eq(MktMailInboxCursorEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
|
|
|
+ .orderByAsc(MktMailInboxCursorEntity::getId)
|
|
|
+ .last("limit 1"));
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 游标表无记录:写入当前未读最大 id,本次不提醒。
|
|
|
+ */
|
|
|
+ private void initCursorWithoutNotify() {
|
|
|
+ Long maxId = 0L;
|
|
|
+ try {
|
|
|
+ Long inboxMax = mktMailMessageService.maxInboxUnreadId();
|
|
|
+ if (inboxMax != null) {
|
|
|
+ maxId = inboxMax;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("初始化营销收件箱游标时查询最大未读 id 失败", e);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ MktMailInboxCursorEntity cursor = new MktMailInboxCursorEntity();
|
|
|
+ cursor.setLastMailId(maxId);
|
|
|
+ cursor.setIsDelete(CommonConstant.DEL_FLAG_0);
|
|
|
+ mktMailInboxCursorService.save(cursor);
|
|
|
+ log.info("营销收件箱新邮件游标首次初始化 lastMailId={},本次不提醒", maxId);
|
|
|
+ }
|
|
|
+
|
|
|
+ private int notifyByOwner(List<MktMailMessageEntity> newMails) {
|
|
|
+ Map<Long, List<MktMailMessageEntity>> byOwner = newMails.stream()
|
|
|
+ .filter(mail -> resolveOwner(mail) != null)
|
|
|
+ .collect(Collectors.groupingBy(MktMailNewNotifyServiceImpl::resolveOwner));
|
|
|
+ if (byOwner.isEmpty()) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ int sent = 0;
|
|
|
+ for (Map.Entry<Long, List<MktMailMessageEntity>> entry : byOwner.entrySet()) {
|
|
|
+ List<MktMailMessageEntity> ownerMails = entry.getValue();
|
|
|
+ Map<String, Object> payload = new HashMap<>();
|
|
|
+ payload.put("messageContent", buildContent(ownerMails));
|
|
|
+ payload.put("callbackUrl", "/marketing");
|
|
|
+ payload.put("id", "/marketing");
|
|
|
+ Set<Long> toUserIds = new HashSet<>();
|
|
|
+ toUserIds.add(entry.getKey());
|
|
|
+ try {
|
|
|
+ tradeMessageAssembleService.sendEventMessage(
|
|
|
+ TradeMessageTypeEnum.MKT_REPLY_NEW_MAIL.getCode(),
|
|
|
+ "MKT_REPLY_NEW_MAIL",
|
|
|
+ payload,
|
|
|
+ toUserIds);
|
|
|
+ sent++;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("发送营销新邮件提醒失败 ownerBy={} count={}", entry.getKey(), ownerMails.size(), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return sent;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String buildContent(List<MktMailMessageEntity> ownerMails) {
|
|
|
+ Set<String> froms = ownerMails.stream()
|
|
|
+ .map(MktMailMessageEntity::getFromAddr)
|
|
|
+ .filter(StrUtil::isNotBlank)
|
|
|
+ .collect(Collectors.toCollection(HashSet::new));
|
|
|
+ int count = ownerMails.size();
|
|
|
+ if (froms.isEmpty()) {
|
|
|
+ return "您有" + count + "封新邮件,请前往查看!";
|
|
|
+ }
|
|
|
+ StringBuilder sb = new StringBuilder();
|
|
|
+ int i = 0;
|
|
|
+ for (String from : froms) {
|
|
|
+ if (i >= 3) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ if (sb.length() > 0) {
|
|
|
+ sb.append("、");
|
|
|
+ }
|
|
|
+ sb.append("【").append(from).append("】");
|
|
|
+ i++;
|
|
|
+ }
|
|
|
+ String more = froms.size() > 3 ? "等" : "";
|
|
|
+ return "您有来自" + sb + more + "的" + count + "封新邮件,请前往查看!";
|
|
|
+ }
|
|
|
+
|
|
|
+ private static Long resolveOwner(MktMailMessageEntity mail) {
|
|
|
+ if (mail.getOwnerBy() != null && mail.getOwnerBy() > 0) {
|
|
|
+ return mail.getOwnerBy();
|
|
|
+ }
|
|
|
+ if (mail.getCreateBy() != null && mail.getCreateBy() > 0) {
|
|
|
+ return mail.getCreateBy();
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+}
|