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

@@ -29,11 +29,12 @@ public class MktMailMessageServiceImpl extends MailDataSourceServiceImpl<MktMail
         wrapper.select(
                         MktMailMessageEntity::getId,
                         MktMailMessageEntity::getFromAddr,
+                        MktMailMessageEntity::getSubject,
+                        MktMailMessageEntity::getReplyMsgId,
                         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);

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

@@ -8,25 +8,16 @@ import com.storlead.framework.common.constant.DSConstants;
 import com.storlead.mail.pojo.entity.MktMailStatusEntity;
 import com.storlead.mail.service.MktMailStatusService;
 import com.storlead.trade.entity.MarketEmailsEntity;
-import com.storlead.trade.entity.MarketingCampaignEntity;
-import com.storlead.trade.enums.TradeMessageTypeEnum;
 import com.storlead.trade.service.MarketEmailsService;
 import com.storlead.trade.service.MarketEmailsStatusSyncService;
-import com.storlead.trade.service.MarketingCampaignService;
-import com.storlead.trade.service.TradeMessageAssembleService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
-import org.springframework.util.StringUtils;
 
 import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
 
 /**
  * 跨库同步：邮件库 mkt_mail_status → 业务库 market_emails，再回写 synced_at。
@@ -51,12 +42,6 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
     @Resource
     private MarketEmailsService marketEmailsService;
 
-    @Resource
-    private MarketingCampaignService marketingCampaignService;
-
-    @Resource
-    private TradeMessageAssembleService tradeMessageAssembleService;
-
     @Override
     public int syncOnce() {
         int totalUpdated = 0;
@@ -167,11 +152,9 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
             changed = true;
         }
         //补充首次回复邮件id
-        boolean firstReplyMailFilled = false;
         if (existing.getFirstReplyEmailId() == null && change.getFirstReplyEmailId() != null) {
             update.set(MarketEmailsEntity::getFirstReplyEmailId, change.getFirstReplyEmailId());
             changed = true;
-            firstReplyMailFilled = true;
         }
         //补充回复邮件主题
         if (existing.getFirstReplyTheme() == null && change.getFirstReplyTheme() != null) {
@@ -181,82 +164,7 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         if (!changed) {
             return ApplyResult.MATCHED_NO_CHANGE;
         }
-        boolean updated = marketEmailsService.update(update);
-        if (updated && firstReplyMailFilled) {
-            notifyCampaignSupervisor(existing, change);
-        }
-        return updated ? ApplyResult.UPDATED : ApplyResult.MATCHED_NO_CHANGE;
-    }
-
-    /**
-     * 首次同步到回复邮件时，提醒营销活动负责人。
-     */
-    private void notifyCampaignSupervisor(MarketEmailsEntity email, MktMailStatusEntity change) {
-        if (email.getMarketingCampaignId() == null) {
-            log.warn("营销回复提醒跳过：无营销活动 id, marketEmailId={}", email.getId());
-            return;
-        }
-        MarketingCampaignEntity campaign = marketingCampaignService.getById(email.getMarketingCampaignId());
-        if (campaign == null) {
-            log.warn("营销回复提醒跳过：活动不存在 campaignId={}", email.getMarketingCampaignId());
-            return;
-        }
-        Set<Long> toUserIds = resolveCampaignSupervisors(campaign);
-        if (toUserIds.isEmpty()) {
-            log.warn("营销回复提醒无接收人 campaignId={}", campaign.getId());
-            return;
-        }
-        Map<String, Object> payload = new HashMap<>();
-        payload.put("messageContent", buildFirstReplyContent(email, change, campaign));
-        payload.put("id", email.getEmailId());
-        try {
-            tradeMessageAssembleService.sendEventMessage(
-                    TradeMessageTypeEnum.MKT_REPLY_NEW_MAIL.getCode(),
-                    "MKT_REPLY_NEW_MAIL",
-                    payload,
-                    toUserIds);
-        } catch (Exception e) {
-            log.error("发送营销回复提醒失败 marketEmailId={} campaignId={}", email.getId(), campaign.getId(), e);
-        }
-    }
-
-    private static Set<Long> resolveCampaignSupervisors(MarketingCampaignEntity campaign) {
-        Set<Long> toUserIds = new HashSet<>();
-        if (StringUtils.hasText(campaign.getSupervisor())) {
-            for (String part : campaign.getSupervisor().split(",")) {
-                if (!StringUtils.hasText(part)) {
-                    continue;
-                }
-                try {
-                    Long userId = Long.valueOf(part.trim());
-                    if (userId > 0) {
-                        toUserIds.add(userId);
-                    }
-                } catch (NumberFormatException ignored) {
-                    // skip
-                }
-            }
-        }
-        if (campaign.getOwnerBy() != null && campaign.getOwnerBy() > 0) {
-            toUserIds.add(campaign.getOwnerBy());
-        }
-        return toUserIds;
-    }
-
-    private static String buildFirstReplyContent(MarketEmailsEntity email, MktMailStatusEntity change,
-                                                 MarketingCampaignEntity campaign) {
-        String from = StringUtils.hasText(email.getRecipient()) ? email.getRecipient() : "";
-        String subject = StringUtils.hasText(change.getFirstReplyTheme())
-                ? change.getFirstReplyTheme()
-                : (StringUtils.hasText(email.getSubject()) ? email.getSubject() : "");
-        String campaignName = StringUtils.hasText(campaign.getName()) ? campaign.getName() : "";
-        if (StringUtils.hasText(from) && StringUtils.hasText(subject)) {
-            return "客户【" + from + "】回复了营销邮件《" + subject + "》，请前往查看！";
-        }
-        if (StringUtils.hasText(from)) {
-            return "客户【" + from + "】回复了营销活动【" + campaignName + "】的邮件，请前往查看！";
-        }
-        return "营销活动【" + campaignName + "】有新的客户回复邮件，请前往查看！";
+        return marketEmailsService.update(update) ? ApplyResult.UPDATED : ApplyResult.MATCHED_NO_CHANGE;
     }
 
     private static Integer mapSendStatus(Integer sendStatus) {

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

@@ -7,20 +7,26 @@ 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.MarketEmailsEntity;
+import com.storlead.trade.entity.MarketingCampaignEntity;
 import com.storlead.trade.entity.MktMailInboxCursorEntity;
 import com.storlead.trade.enums.TradeMessageTypeEnum;
+import com.storlead.trade.service.MarketEmailsService;
+import com.storlead.trade.service.MarketingCampaignService;
 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 org.springframework.util.StringUtils;
 
 import javax.annotation.Resource;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -44,6 +50,12 @@ public class MktMailNewNotifyServiceImpl implements MktMailNewNotifyService {
     @Resource
     private TradeMessageAssembleService tradeMessageAssembleService;
 
+    @Resource
+    private MarketEmailsService marketEmailsService;
+
+    @Resource
+    private MarketingCampaignService marketingCampaignService;
+
     @Override
     public synchronized int syncOnce() {
         MktMailInboxCursorEntity cursor = currentCursor();
@@ -66,7 +78,7 @@ public class MktMailNewNotifyServiceImpl implements MktMailNewNotifyService {
                 break;
             }
             Long batchLastId = newMails.get(newMails.size() - 1).getId();
-            notified += notifyByOwner(newMails);
+            notified += notifyByCampaignSupervisor(newMails);
             if (batchLastId != null) {
                 lastId = batchLastId;
                 maxId = batchLastId;
@@ -110,22 +122,36 @@ public class MktMailNewNotifyServiceImpl implements MktMailNewNotifyService {
         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()) {
+    /**
+     * 每一封回复单独发一条消息，接收人为对应营销活动负责人。
+     */
+    private int notifyByCampaignSupervisor(List<MktMailMessageEntity> newMails) {
+        Map<Long, Long> mailToCampaign = resolveMailCampaignIds(newMails);
+        if (mailToCampaign.isEmpty()) {
+            log.warn("营销新邮件未匹配到活动，跳过提醒 count={}", newMails.size());
             return 0;
         }
+        Map<Long, MarketingCampaignEntity> campaignMap = loadCampaigns(new HashSet<>(mailToCampaign.values()));
         int sent = 0;
-        for (Map.Entry<Long, List<MktMailMessageEntity>> entry : byOwner.entrySet()) {
-            List<MktMailMessageEntity> ownerMails = entry.getValue();
+        for (MktMailMessageEntity mail : newMails) {
+            Long campaignId = mailToCampaign.get(mail.getId());
+            if (campaignId == null) {
+                log.warn("营销回复提醒跳过：未匹配活动 mailId={}", mail.getId());
+                continue;
+            }
+            MarketingCampaignEntity campaign = campaignMap.get(campaignId);
+            if (campaign == null) {
+                log.warn("营销回复提醒跳过：活动不存在 campaignId={} mailId={}", campaignId, mail.getId());
+                continue;
+            }
+            Set<Long> toUserIds = resolveCampaignSupervisors(campaign);
+            if (toUserIds.isEmpty()) {
+                log.warn("营销回复提醒无接收人 campaignId={} mailId={}", campaign.getId(), mail.getId());
+                continue;
+            }
             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());
+            payload.put("messageContent", buildContent(mail, campaign));
+            payload.put("id",mail.getId());
             try {
                 tradeMessageAssembleService.sendEventMessage(
                         TradeMessageTypeEnum.MKT_REPLY_NEW_MAIL.getCode(),
@@ -134,44 +160,105 @@ public class MktMailNewNotifyServiceImpl implements MktMailNewNotifyService {
                         toUserIds);
                 sent++;
             } catch (Exception e) {
-                log.error("发送营销新邮件提醒失败 ownerBy={} count={}", entry.getKey(), ownerMails.size(), e);
+                log.error("发送营销回复提醒失败 campaignId={} mailId={}", campaign.getId(), mail.getId(), 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 + "封新邮件，请前往查看！";
+    /**
+     * 优先用回复原邮件 id（replyMsgId = market_emails.email_id），否则用 first_reply_email_id。
+     */
+    private Map<Long, Long> resolveMailCampaignIds(List<MktMailMessageEntity> newMails) {
+        Map<Long, Long> mailToCampaign = new HashMap<>();
+        Set<Long> replyMsgIds = newMails.stream()
+                .map(MktMailMessageEntity::getReplyMsgId)
+                .filter(id -> id != null && id > 0)
+                .collect(Collectors.toSet());
+        Set<Long> inboxIds = newMails.stream()
+                .map(MktMailMessageEntity::getId)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toSet());
+        if (!replyMsgIds.isEmpty()) {
+            List<MarketEmailsEntity> bySent = marketEmailsService.list(new LambdaQueryWrapper<MarketEmailsEntity>()
+                    .select(MarketEmailsEntity::getEmailId, MarketEmailsEntity::getMarketingCampaignId)
+                    .in(MarketEmailsEntity::getEmailId, replyMsgIds)
+                    .eq(MarketEmailsEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                    .isNotNull(MarketEmailsEntity::getMarketingCampaignId));
+            Map<Long, Long> sentToCampaign = bySent.stream()
+                    .filter(e -> e.getEmailId() != null && e.getMarketingCampaignId() != null)
+                    .collect(Collectors.toMap(MarketEmailsEntity::getEmailId,
+                            MarketEmailsEntity::getMarketingCampaignId, (a, b) -> a));
+            for (MktMailMessageEntity mail : newMails) {
+                Long campaignId = sentToCampaign.get(mail.getReplyMsgId());
+                if (campaignId != null) {
+                    mailToCampaign.put(mail.getId(), campaignId);
+                }
+            }
         }
-        StringBuilder sb = new StringBuilder();
-        int i = 0;
-        for (String from : froms) {
-            if (i >= 3) {
-                break;
+        Set<Long> unmatchedIds = inboxIds.stream()
+                .filter(id -> !mailToCampaign.containsKey(id))
+                .collect(Collectors.toSet());
+        if (!unmatchedIds.isEmpty()) {
+            List<MarketEmailsEntity> byReply = marketEmailsService.list(new LambdaQueryWrapper<MarketEmailsEntity>()
+                    .select(MarketEmailsEntity::getFirstReplyEmailId, MarketEmailsEntity::getMarketingCampaignId)
+                    .in(MarketEmailsEntity::getFirstReplyEmailId, unmatchedIds)
+                    .eq(MarketEmailsEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                    .isNotNull(MarketEmailsEntity::getMarketingCampaignId));
+            for (MarketEmailsEntity email : byReply) {
+                if (email.getFirstReplyEmailId() != null && email.getMarketingCampaignId() != null) {
+                    mailToCampaign.putIfAbsent(email.getFirstReplyEmailId(), email.getMarketingCampaignId());
+                }
             }
-            if (sb.length() > 0) {
-                sb.append("、");
+        }
+        return mailToCampaign;
+    }
+
+    private Map<Long, MarketingCampaignEntity> loadCampaigns(Set<Long> campaignIds) {
+        if (CollectionUtils.isEmpty(campaignIds)) {
+            return new HashMap<>();
+        }
+        return marketingCampaignService.listByIds(campaignIds).stream()
+                .filter(c -> c.getId() != null)
+                .collect(Collectors.toMap(MarketingCampaignEntity::getId, c -> c, (a, b) -> a));
+    }
+
+    private static Set<Long> resolveCampaignSupervisors(MarketingCampaignEntity campaign) {
+        Set<Long> toUserIds = new HashSet<>();
+        if (StringUtils.hasText(campaign.getSupervisor())) {
+            for (String part : campaign.getSupervisor().split(",")) {
+                if (!StringUtils.hasText(part)) {
+                    continue;
+                }
+                try {
+                    Long userId = Long.valueOf(part.trim());
+                    if (userId > 0) {
+                        toUserIds.add(userId);
+                    }
+                } catch (NumberFormatException ignored) {
+                    // skip
+                }
             }
-            sb.append("【").append(from).append("】");
-            i++;
         }
-        String more = froms.size() > 3 ? "等" : "";
-        return "您有来自" + sb + more + "的" + count + "封新邮件，请前往查看！";
+        if (toUserIds.isEmpty() && campaign.getOwnerBy() != null && campaign.getOwnerBy() > 0) {
+            toUserIds.add(campaign.getOwnerBy());
+        }
+        return toUserIds;
     }
 
-    private static Long resolveOwner(MktMailMessageEntity mail) {
-        if (mail.getOwnerBy() != null && mail.getOwnerBy() > 0) {
-            return mail.getOwnerBy();
+    private static String buildContent(MktMailMessageEntity mail, MarketingCampaignEntity campaign) {
+        String from = StrUtil.blankToDefault(mail.getFromAddr(), "");
+        String subject = StrUtil.blankToDefault(mail.getSubject(), "");
+        String campaignName = StringUtils.hasText(campaign.getName()) ? campaign.getName() : "";
+        if (StringUtils.hasText(from) && StringUtils.hasText(subject)) {
+            return "客户【" + from + "】回复了营销活动【" + campaignName + "】的邮件《" + subject + "》，请前往查看！";
+        }
+        if (StringUtils.hasText(from)) {
+            return "客户【" + from + "】回复了营销活动【" + campaignName + "】的邮件，请前往查看！";
         }
-        if (mail.getCreateBy() != null && mail.getCreateBy() > 0) {
-            return mail.getCreateBy();
+        if (StringUtils.hasText(campaignName)) {
+            return "营销活动【" + campaignName + "】有一封新的客户回复邮件，请前往查看！";
         }
-        return null;
+        return "您有一封新的客户回复邮件，请前往查看！";
     }
 }

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

@@ -14,20 +14,20 @@ import javax.annotation.Resource;
 @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);
-//        }
-//    }
+    @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);
+        }
+    }
 }