营销回复消息

java/storlead-mail/storlead-mail-biz/src/main/java/com/storlead/mail/service/impl/MktMailMessageServiceImpl.java

@@ -1,14 +1,17 @@
 package com.storlead.mail.service.impl;
 
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.storlead.framework.common.constant.CommonConstant;
+import com.storlead.mail.enums.EmailBoxEnum;
 import com.storlead.mail.mapper.MktMailMessageMapper;
 import com.storlead.mail.pojo.entity.MktMailMessageEntity;
 import com.storlead.mail.service.MktMailMessageService;
 import com.storlead.mail.service.impl.support.MailDataSourceServiceImpl;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.stereotype.Service;
-import javax.mail.*;
-import java.util.*;
 
+import java.util.Collections;
+import java.util.List;
 
 /**
  * 营销邮件：对齐 CRM 草稿 / 附件 / 延时 / 发送，成功后写入 mkt_mail_attachment。
@@ -17,4 +20,44 @@ import java.util.*;
 @Service
 public class MktMailMessageServiceImpl extends MailDataSourceServiceImpl<MktMailMessageMapper, MktMailMessageEntity> implements MktMailMessageService {
 
+    @Override
+    public List<MktMailMessageEntity> listNewInboxMails(Long lastId, int limit) {
+        if (limit <= 0) {
+            return Collections.emptyList();
+        }
+        LambdaQueryWrapper<MktMailMessageEntity> wrapper = new LambdaQueryWrapper<>();
+        wrapper.select(
+                        MktMailMessageEntity::getId,
+                        MktMailMessageEntity::getFromAddr,
+                        MktMailMessageEntity::getOwnerBy,
+                        MktMailMessageEntity::getCreateBy)
+                .eq(MktMailMessageEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                .eq(MktMailMessageEntity::getFolder, EmailBoxEnum.INBOX.code)
+                .and(w -> w.eq(MktMailMessageEntity::getIsRead, 0).or().isNull(MktMailMessageEntity::getIsRead))
+                .gt(lastId != null, MktMailMessageEntity::getId, lastId)
+                .orderByAsc(MktMailMessageEntity::getId)
+                .last("limit " + limit);
+        return list(wrapper);
+    }
+
+    @Override
+    public Long maxInboxUnreadId() {
+        MktMailMessageEntity unreadLatest = getOne(new LambdaQueryWrapper<MktMailMessageEntity>()
+                .select(MktMailMessageEntity::getId)
+                .eq(MktMailMessageEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                .eq(MktMailMessageEntity::getFolder, EmailBoxEnum.INBOX.code)
+                .and(w -> w.eq(MktMailMessageEntity::getIsRead, 0).or().isNull(MktMailMessageEntity::getIsRead))
+                .orderByDesc(MktMailMessageEntity::getId)
+                .last("limit 1"));
+        if (unreadLatest != null && unreadLatest.getId() != null) {
+            return unreadLatest.getId();
+        }
+        MktMailMessageEntity inboxLatest = getOne(new LambdaQueryWrapper<MktMailMessageEntity>()
+                .select(MktMailMessageEntity::getId)
+                .eq(MktMailMessageEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                .eq(MktMailMessageEntity::getFolder, EmailBoxEnum.INBOX.code)
+                .orderByDesc(MktMailMessageEntity::getId)
+                .last("limit 1"));
+        return inboxLatest == null ? 0L : inboxLatest.getId();
+    }
 }

java/storlead-mail/storlead-mail-spi/src/main/java/com/storlead/mail/service/MktMailMessageService.java

@@ -3,10 +3,24 @@ package com.storlead.mail.service;
 import com.storlead.framework.mybatis.service.MyBaseService;
 import com.storlead.mail.pojo.entity.MktMailMessageEntity;
 
+import java.util.List;
+
 /**
  * 营销邮件投递业务（对齐 CRM：草稿、附件、延时、发送）。
  */
 public interface MktMailMessageService extends MyBaseService<MktMailMessageEntity> {
 
+    /**
+     * 营销收件箱未读新邮件（不含正文），按 id 升序分页。
+     *
+     * @param lastId 上一批最后一条 id，首页传 null
+     * @param limit  批量大小
+     */
+    List<MktMailMessageEntity> listNewInboxMails(Long lastId, int limit);
 
+    /**
+     * 收件箱未读邮件当前最大 id；没有未读则取收件箱最大 id。
+     */
+    Long maxInboxUnreadId();
 }
+

java/storlead-sasa/storlead-trade/sql/mkt_mail_inbox_cursor.sql

@@ -0,0 +1,14 @@
+-- 营销收件箱新邮件提醒游标（记录已处理的未读邮件最大 id）
+CREATE TABLE IF NOT EXISTS mkt_mail_inbox_cursor (
+    id            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
+    last_mail_id  BIGINT       NOT NULL DEFAULT 0 COMMENT '已处理的收件箱未读邮件最大id（mkt_mail_message.id）',
+    owner_by      BIGINT       DEFAULT NULL,
+    create_by     BIGINT       DEFAULT NULL,
+    update_by     BIGINT       DEFAULT NULL,
+    create_time   DATETIME     DEFAULT NULL,
+    update_time   DATETIME     DEFAULT NULL,
+    is_delete     INT          DEFAULT 0,
+    enabled       TINYINT      DEFAULT 1,
+    sort          INT          DEFAULT NULL,
+    PRIMARY KEY (id)
+) COMMENT='营销收件箱新邮件提醒游标';

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/entity/MktMailInboxCursorEntity.java

@@ -0,0 +1,28 @@
+package com.storlead.trade.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.storlead.framework.mybatis.entity.SysBaseField;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 营销收件箱新邮件提醒游标：记录已处理的未读邮件最大 id。
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("mkt_mail_inbox_cursor")
+@ApiModel(value = "MktMailInboxCursorEntity", description = "营销收件箱新邮件提醒游标")
+public class MktMailInboxCursorEntity extends SysBaseField {
+
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    @ApiModelProperty("已处理的收件箱未读邮件最大 id（mkt_mail_message.id）")
+    @TableField("last_mail_id")
+    private Long lastMailId;
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/enums/TradeMessageTypeEnum.java

@@ -13,6 +13,7 @@ import java.util.Map;
 public enum  TradeMessageTypeEnum {
 
     CUSTOMIZE_EXAMPLE("0", "业务自定义"),
+    MKT_REPLY_NEW_MAIL("10", "新回复邮件"),
     OTHER_EXTEND("999", "其他"),
     ;
 

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/mapper/MktMailInboxCursorMapper.java

@@ -0,0 +1,9 @@
+package com.storlead.trade.mapper;
+
+import com.storlead.trade.entity.MktMailInboxCursorEntity;
+import com.storlead.trade.mapper.support.TradeBaseMapper;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface MktMailInboxCursorMapper extends TradeBaseMapper<MktMailInboxCursorEntity> {
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/MktMailInboxCursorService.java

@@ -0,0 +1,7 @@
+package com.storlead.trade.service;
+
+import com.storlead.framework.mybatis.service.MyBaseService;
+import com.storlead.trade.entity.MktMailInboxCursorEntity;
+
+public interface MktMailInboxCursorService extends MyBaseService<MktMailInboxCursorEntity> {
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/MktMailNewNotifyService.java

@@ -0,0 +1,12 @@
+package com.storlead.trade.service;
+
+/**
+ * 定时扫描营销收件箱新邮件并发消息提醒。
+ */
+public interface MktMailNewNotifyService {
+
+    /**
+     * @return 本次发出提醒的条数（按接收人聚合后的次数）
+     */
+    int syncOnce();
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/impl/MktMailInboxCursorServiceImpl.java

@@ -0,0 +1,16 @@
+package com.storlead.trade.service.impl;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.storlead.framework.common.constant.DSConstants;
+import com.storlead.trade.entity.MktMailInboxCursorEntity;
+import com.storlead.trade.mapper.MktMailInboxCursorMapper;
+import com.storlead.trade.service.MktMailInboxCursorService;
+import com.storlead.trade.service.impl.support.TradeDataSourceServiceImpl;
+import org.springframework.stereotype.Service;
+
+@Service
+@DS(DSConstants.DATASOURCE_TRADE)
+public class MktMailInboxCursorServiceImpl
+        extends TradeDataSourceServiceImpl<MktMailInboxCursorMapper, MktMailInboxCursorEntity>
+        implements MktMailInboxCursorService {
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/service/impl/MktMailNewNotifyServiceImpl.java

@@ -0,0 +1,176 @@
+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");
+            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;
+    }
+}

java/storlead-sasa/storlead-trade/src/main/java/com/storlead/trade/task/MktMailNewNotifyTask.java

@@ -0,0 +1,33 @@
+package com.storlead.trade.task;
+
+import com.storlead.trade.service.MktMailNewNotifyService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 定时查询营销收件箱（mkt_mail_message）是否有新邮件，有则发消息提醒。
+ */
+@Slf4j
+@Component
+public class MktMailNewNotifyTask {
+
+    @Resource
+    private MktMailNewNotifyService mktMailNewNotifyService;
+
+    @Scheduled(cron = "0 */2 * * * ?")
+    public void sync() {
+        long start = System.currentTimeMillis();
+        try {
+            int notified = mktMailNewNotifyService.syncOnce();
+            if (notified > 0) {
+                log.info("营销收件箱新邮件提醒完成 notified={} cost={}ms",
+                        notified, System.currentTimeMillis() - start);
+            }
+        } catch (Exception e) {
+            log.error("营销收件箱新邮件提醒失败", e);
+        }
+    }
+}