营销逻辑修改

storlead-api/src/main/java/com/storlead/mail/mkt/MktMailApiController.java

@@ -19,6 +19,7 @@ import com.storlead.sales.mail.mkt.pojo.MktMailQueryDTO;
 import com.storlead.sales.mail.mkt.pojo.MktSendMailDTO;
 import com.storlead.sales.mail.mkt.service.MktMailAttachmentService;
 import com.storlead.sales.mail.mkt.service.MktMailMessageService;
+import com.storlead.sales.mail.mkt.service.MktMailReceiveService;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.log4j.Log4j2;
@@ -59,6 +60,9 @@ public class MktMailApiController {
     @Resource
     private MktMailMessageService mktMailMessageService;
 
+    @Resource
+    private MktMailReceiveService mktMailReceiveService;
+
     @Resource
     private MktMailAttachmentService mktMailAttachmentService;
 
@@ -158,6 +162,29 @@ public class MktMailApiController {
         writeFileResponse(resource.getFilePath(), resource.getFileName(), resource.getFileExt(), response);
     }
 
+    @PostMapping("/receive_mail")
+    @ApiOperation("接收营销邮件（异步：INBOX+Sent Items，落 mkt_mail_message 并用 reply_msg_id 绑定原信）")
+    public Result<?> receiveMail(@RequestParam("accountId") Long accountId) {
+        Long userId = LoginUserUtil.getCurrentUserId();
+        if (Objects.isNull(userId)) {
+            return Result.error("未登录");
+        }
+        if (accountId == null) {
+            return Result.error("账户不能为空");
+        }
+        try {
+            SmtpPopSettingsEntity account = mailAccountBizService.requireAccount(accountId, MailSceneEnum.MKT);
+            if (account.getEnableReceive() != null && !Integer.valueOf(1).equals(account.getEnableReceive())) {
+                return Result.error("该营销账户未开启收信（enable_receive=0）");
+            }
+            mktMailReceiveService.receiveEmails(account);
+            return Result.ok();
+        } catch (Exception e) {
+            log.error("mkt receive_mail error", e);
+            return Result.error(e.getMessage() == null ? "收信失败" : e.getMessage());
+        }
+    }
+
     @PostMapping("/save_draft")
     @ApiOperation("保存营销邮件草稿")
     public Result<?> saveDraft(@RequestBody MktSendMailDTO dto) {
@@ -170,7 +197,7 @@ public class MktMailApiController {
     }
 
     @PostMapping("/send")
-    @ApiOperation("发送营销邮件（单封，支持附件/延时/打开追踪）")
+    @ApiOperation("发送营销邮件（单封；默认同步等待投递结果；async=true 异步投递+回调）")
     public Result<?> send(@RequestBody MktSendMailDTO dto) {
         Long userId = LoginUserUtil.getCurrentUserId();
         if (Objects.isNull(userId)) {

storlead-mail/README.md

@@ -34,6 +34,7 @@ API：
 - `com.storlead.mail.common` — 账户配置
 - `com.storlead.mail.crm` — CRM 邮件接口（原 `/mail/**`）
 - `com.storlead.mail.mkt` — 营销接口（`/mail/mkt/**`）
+  - `POST /mail/mkt/receive_mail?accountId=` — 异步收信（INBOX+Sent Items → `mkt_mail_message.folder`，`reply_msg_id` 绑定原信）
   - `POST /mail/mkt/file/cache_mail_file` — 上传临时附件（`scene=MKT`，需 `accountId`）
   - `POST /mail/mkt/file/list` — 按邮件查正式附件 `mkt_mail_attachment`
   - `POST /mail/mkt/file/download` — 下载正式附件

storlead-mail/storlead-mail-common/src/main/java/com/storlead/sales/mail/common/enums/MailSendStatus.java

@@ -0,0 +1,33 @@
+package com.storlead.sales.mail.common.enums;
+
+import com.baomidou.mybatisplus.annotation.EnumValue;
+import com.fasterxml.jackson.annotation.JsonValue;
+import lombok.Getter;
+
+/**
+ * SMTP 投递结果（对齐 CRM SendStatus，供 CRM/MKT 共用语义）。
+ */
+@Getter
+public enum MailSendStatus {
+
+    SUCCESS("SUCCESS", "成功", 0),
+    FAILED("FAILED", "失败", 0),
+    PARTIAL_SUCCESS("PARTIAL_SUCCESS", "部分成功", 0),
+    SERVER_EXCEPTION_550("SERVER_EXCEPTION", "邮件发送失败，请确认发送的邮箱是否正确", 0),
+    SERVER_LOGIN_FAIL("SERVER_LOGIN_FAIL", "登录失败，请确认账号密码", 401),
+    MTP_SEND_FAILED_EXCEPTION_453("MTP_SEND_FAILED_EXCEPTION_453", "发送异常,请检查附件或邮件大小", 453);
+
+    @EnumValue
+    public final String code;
+
+    @JsonValue
+    public final String desc;
+
+    public final Integer errorCode;
+
+    MailSendStatus(String code, String desc, Integer errorCode) {
+        this.code = code;
+        this.desc = desc;
+        this.errorCode = errorCode;
+    }
+}

storlead-mail/storlead-mail-common/src/main/java/com/storlead/sales/mail/common/util/MailReceiveQueueThreadPool.java

@@ -0,0 +1,74 @@
+package com.storlead.sales.mail.common.util;
+
+import com.storlead.framework.common.thread.ThreadPoolUtil;
+import lombok.extern.log4j.Log4j2;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * 收信任务队列（CRM/MKT 共用）：同一邮箱串行，避免并发拉信。
+ */
+@Log4j2
+public class MailReceiveQueueThreadPool {
+
+    private static volatile MailReceiveQueueThreadPool instance;
+
+    private final Map<String, Queue<Runnable>> taskMap = new HashMap<>();
+    private final Map<String, Boolean> taskInProgress = new HashMap<>();
+    private final ThreadPoolExecutor executorService = ThreadPoolUtil.getThreadExecutor();
+
+    public static MailReceiveQueueThreadPool getInstance() {
+        if (instance == null) {
+            synchronized (MailReceiveQueueThreadPool.class) {
+                if (instance == null) {
+                    instance = new MailReceiveQueueThreadPool();
+                }
+            }
+        }
+        return instance;
+    }
+
+    public Boolean getTaskInProgressByMail(String key) {
+        return taskInProgress.getOrDefault(key, false);
+    }
+
+    public synchronized void addTask(String key, Runnable task) {
+        if (Objects.nonNull(taskMap.get(key))) {
+            return;
+        }
+        taskMap.putIfAbsent(key, new LinkedList<>());
+        taskMap.get(key).offer(task);
+        taskInProgress.putIfAbsent(key, false);
+        processNextTask(key);
+    }
+
+    private synchronized void processNextTask(String key) {
+        if (taskInProgress.getOrDefault(key, false)) {
+            return;
+        }
+        Queue<Runnable> queue = taskMap.get(key);
+        if (queue == null || queue.isEmpty()) {
+            return;
+        }
+        Runnable task = queue.poll();
+        taskInProgress.put(key, true);
+        executorService.submit(() -> {
+            try {
+                task.run();
+            } finally {
+                taskCompleted(key);
+            }
+        });
+    }
+
+    public synchronized void taskCompleted(String key) {
+        taskInProgress.put(key, false);
+        taskMap.remove(key);
+        processNextTask(key);
+    }
+}

storlead-mail/storlead-mail-common/src/main/resources/sql/mail_scene_and_mkt.sql

@@ -25,7 +25,17 @@ CREATE TABLE IF NOT EXISTS mkt_mail_message (
     recipient_cc    text         NULL COMMENT '抄送人，多个用逗号分隔',
     recipient_bcc   text         NULL COMMENT '密送人，多个用逗号分隔',
     content         mediumtext   NULL COMMENT '邮件正文（HTML/文本）；体积大时可后续改为外置只存引用',
-    status          tinyint      NOT NULL DEFAULT 0 COMMENT '投递状态：-2草稿；-1待定时发送；0发送中；1发送成功；2发送失败',
+    folder          varchar(32)  NULL COMMENT '文件夹：INBOX/SENT/SENTING/DRAFT',
+    msg_uid         varchar(128) NULL COMMENT '服务器邮件UID（IMAP/POP）',
+    recipient_date  datetime     NULL COMMENT '收信时间（服务器 ReceivedDate）',
+    sent_date       datetime     NULL COMMENT '邮件头 SentDate',
+    is_only_head    tinyint      NOT NULL DEFAULT 0 COMMENT '是否仅拉头：1是 0已有正文',
+    is_read         tinyint      NOT NULL DEFAULT 0 COMMENT '是否已读：0未读 1已读',
+    in_out_mark     tinyint      NULL COMMENT '进出标记：1收件 2发件',
+    email_size      bigint       NULL COMMENT '邮件大小（字节）',
+    in_reply_to     varchar(512) NULL COMMENT 'In-Reply-To 头，用于匹配回复原信',
+    reply_msg_id    bigint       NULL COMMENT '回复的原邮件ID，对应 mkt_mail_message.id',
+    status          tinyint      NOT NULL DEFAULT 0 COMMENT '投递状态：-2草稿；-1待定时发送；0发送中；1发送成功/已收妥；2发送失败',
     fail_reason     varchar(512) NULL COMMENT '发送失败原因（异常信息摘要）',
     is_track        tinyint      NOT NULL DEFAULT 0 COMMENT '是否开启打开追踪：0否；1是',
     opened          tinyint      NOT NULL DEFAULT 0 COMMENT '是否已打开（像素回调）：0未打开；1已打开；存在网关预取假阳性，仅为弱信号',
@@ -45,7 +55,10 @@ CREATE TABLE IF NOT EXISTS mkt_mail_message (
     UNIQUE KEY uk_mkt_track_token (track_token),
     KEY idx_mkt_biz_ref (biz_ref, status),
     KEY idx_mkt_account_sent (account_id, sent_time),
-    KEY idx_mkt_org_sent (org_id, sent_time)
+    KEY idx_mkt_org_sent (org_id, sent_time),
+    KEY idx_mkt_account_folder_recv (account_id, folder, recipient_date),
+    KEY idx_mkt_account_message_id (account_id, message_id),
+    KEY idx_mkt_reply_msg_id (reply_msg_id)
 ) COMMENT='营销邮件投递记录';
 
 -- 临时附件增加场景字段：默认 CRM，存量与未传 scene 的上传仍走 CRM 逻辑

storlead-mail/storlead-mail-common/src/main/resources/sql/mkt_mail_message_receive.sql

@@ -0,0 +1,19 @@
+-- mkt_mail_message 收信扩展：用 folder 区分收/发，并支持回复原信绑定
+ALTER TABLE mkt_mail_message
+    ADD COLUMN folder varchar(32) NULL COMMENT '文件夹：INBOX/SENT/SENTING/DRAFT' AFTER content,
+    ADD COLUMN msg_uid varchar(128) NULL COMMENT '服务器邮件UID（IMAP/POP）' AFTER folder,
+    ADD COLUMN recipient_date datetime NULL COMMENT '收信时间（服务器 ReceivedDate）' AFTER msg_uid,
+    ADD COLUMN sent_date datetime NULL COMMENT '邮件头 SentDate' AFTER recipient_date,
+    ADD COLUMN is_only_head tinyint NOT NULL DEFAULT 0 COMMENT '是否仅拉头：1是 0已有正文' AFTER sent_date,
+    ADD COLUMN is_read tinyint NOT NULL DEFAULT 0 COMMENT '是否已读：0未读 1已读' AFTER is_only_head,
+    ADD COLUMN in_out_mark tinyint NULL COMMENT '进出标记：1收件 2发件' AFTER is_read,
+    ADD COLUMN email_size bigint NULL COMMENT '邮件大小（字节）' AFTER in_out_mark,
+    ADD COLUMN in_reply_to varchar(512) NULL COMMENT 'In-Reply-To 头，用于匹配回复原信' AFTER email_size,
+    ADD COLUMN reply_msg_id bigint NULL COMMENT '回复的原邮件ID，对应 mkt_mail_message.id' AFTER in_reply_to;
+
+-- 存量发信记录默认视为发件箱
+UPDATE mkt_mail_message SET folder = 'SENTING', in_out_mark = 2 WHERE folder IS NULL AND status IN (-2, -1, 0, 1, 2);
+
+CREATE INDEX idx_mkt_account_folder_recv ON mkt_mail_message (account_id, folder, recipient_date);
+CREATE INDEX idx_mkt_account_message_id ON mkt_mail_message (account_id, message_id);
+CREATE INDEX idx_mkt_reply_msg_id ON mkt_mail_message (reply_msg_id);

storlead-mail/storlead-mail-dispatch/src/main/java/com/storlead/sales/mail/dispatch/MktEmailPullTaskJob.java

@@ -0,0 +1,49 @@
+package com.storlead.sales.mail.dispatch;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.storlead.sales.mail.common.entity.SmtpPopSettingsEntity;
+import com.storlead.sales.mail.common.enums.MailSceneEnum;
+import com.storlead.sales.mail.common.service.SmtpPopSettingsService;
+import com.storlead.sales.mail.mkt.service.MktMailReceiveService;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+/**
+ * 营销收信调度：仅 scene=MKT 且 enable_receive=1 的账户。
+ */
+@Slf4j
+@Component
+@NoArgsConstructor
+public class MktEmailPullTaskJob {
+
+    @Resource
+    private SmtpPopSettingsService smtpPopSettingsService;
+
+    @Resource
+    private MktMailReceiveService mktMailReceiveService;
+
+    //@Scheduled(cron = "${storlead.mail.scheduler.mkt-pull-email-cron:0 */10 * * * ?}")
+    public void pullMktMailTask() {
+        LambdaQueryWrapper<SmtpPopSettingsEntity> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(SmtpPopSettingsEntity::getIsDelete, 0);
+        wrapper.eq(SmtpPopSettingsEntity::getEnabled, 1);
+        wrapper.eq(SmtpPopSettingsEntity::getScene, MailSceneEnum.MKT.getCode());
+        wrapper.eq(SmtpPopSettingsEntity::getEnableReceive, 1);
+        List<SmtpPopSettingsEntity> accounts = smtpPopSettingsService.list(wrapper);
+        if (CollectionUtils.isEmpty(accounts)) {
+            return;
+        }
+        for (SmtpPopSettingsEntity account : accounts) {
+            try {
+                mktMailReceiveService.receiveEmails(account);
+            } catch (Exception e) {
+                log.error("mkt pull task error, accountId={}", account.getId(), e);
+            }
+        }
+    }
+}

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/callback/MktMailSendCallback.java

@@ -0,0 +1,27 @@
+package com.storlead.sales.mail.mkt.callback;
+
+import com.storlead.sales.mail.mkt.entity.MktMailMessageEntity;
+
+/**
+ * 营销邮件 SMTP 投递结果业务回调（可注册多个 Spring Bean）。
+ * <p>对齐 CRM 侧 TransportListener 成功/失败后的后续处理扩展点。
+ */
+public interface MktMailSendCallback {
+
+    /**
+     * 投递成功（含部分成功按成功处理时可覆盖 {@link #onPartialSuccess}）。
+     */
+    void onSuccess(MktMailMessageEntity entity);
+
+    /**
+     * 投递失败。
+     */
+    void onFailure(MktMailMessageEntity entity, String reason);
+
+    /**
+     * 部分成功，默认按成功处理。
+     */
+    default void onPartialSuccess(MktMailMessageEntity entity) {
+        onSuccess(entity);
+    }
+}

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/entity/MktMailMessageEntity.java

@@ -72,7 +72,46 @@ public class MktMailMessageEntity extends SysBaseField {
     @TableField("content")
     private String content;
 
-    @ApiModelProperty("发送状态：-2草稿 -1待定时 0发送中 1成功 2失败")
+    @ApiModelProperty("文件夹：INBOX/SENT/SENTING/DRAFT")
+    @TableField("folder")
+    private String folder;
+
+    @ApiModelProperty("服务器邮件UID")
+    @TableField("msg_uid")
+    private String msgUid;
+
+    @ApiModelProperty("收信时间")
+    @TableField("recipient_date")
+    private Date recipientDate;
+
+    @ApiModelProperty("邮件头发送时间")
+    @TableField("sent_date")
+    private Date sentDate;
+
+    @ApiModelProperty("是否仅拉头：1是 0已有正文")
+    @TableField("is_only_head")
+    private Integer isOnlyHead;
+
+    @ApiModelProperty("是否已读")
+    @TableField("is_read")
+    private Integer isRead;
+
+    @ApiModelProperty("进出标记：1收件 2发件")
+    @TableField("in_out_mark")
+    private Integer inOutMark;
+
+    @TableField("email_size")
+    private Long emailSize;
+
+    @ApiModelProperty("In-Reply-To 头")
+    @TableField("in_reply_to")
+    private String inReplyTo;
+
+    @ApiModelProperty("回复的原邮件ID")
+    @TableField("reply_msg_id")
+    private Long replyMsgId;
+
+    @ApiModelProperty("发送状态：-2草稿 -1待定时 0发送中 1成功 2失败；收信记录一般为1")
     @TableField("status")
     private Integer status;
 

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/enums/MktMailFolderEnum.java

@@ -0,0 +1,39 @@
+package com.storlead.sales.mail.mkt.enums;
+
+import cn.hutool.core.util.StrUtil;
+import com.baomidou.mybatisplus.annotation.EnumValue;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+/**
+ * 营销邮件文件夹（对齐 CRM EmailBoxEnum 常用值）。
+ */
+public enum MktMailFolderEnum {
+
+    INBOX("INBOX", "收件箱"),
+    SENT("SENT", "已发送（服务器）"),
+    SENTING("SENTING", "发件箱/本地发送中"),
+    DRAFT("DRAFT", "草稿箱");
+
+    @EnumValue
+    public final String code;
+
+    @JsonValue
+    public final String desc;
+
+    MktMailFolderEnum(String code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    public static MktMailFolderEnum of(String code) {
+        if (StrUtil.isBlank(code)) {
+            return null;
+        }
+        for (MktMailFolderEnum value : values()) {
+            if (value.code.equalsIgnoreCase(code.trim())) {
+                return value;
+            }
+        }
+        return null;
+    }
+}

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/pojo/MktMailQueryDTO.java

@@ -22,4 +22,7 @@ public class MktMailQueryDTO extends PageQuery {
 
     @ApiModelProperty("收件人")
     private String recipient;
+
+    @ApiModelProperty("文件夹：INBOX/SENT/SENTING/DRAFT")
+    private String folder;
 }

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/pojo/MktSendMailDTO.java

@@ -56,4 +56,7 @@ public class MktSendMailDTO {
     @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm")
     @ApiModelProperty("定时发送时间")
     private Date delaySendTime;
+
+    @ApiModelProperty("是否异步发送：true=落库后立即返回(status=发送中)，SMTP 后台投递并回调；false/空=同步等待投递结果（默认）")
+    private Boolean async;
 }

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/service/MktMailMessageService.java

@@ -17,7 +17,8 @@ public interface MktMailMessageService extends MyBaseService<MktMailMessageEntit
     MktMailMessageEntity saveDraft(MktSendMailDTO dto, Long operatorId);
 
     /**
-     * 发送营销邮件；支持附件、CID 内嵌图、延时发送、打开追踪像素。
+     * 发送营销邮件；支持附件、CID、延时、打开追踪。
+     * 投递走 TransportListener 回调；{@code dto.async=true} 时异步投递。
      */
     MktMailMessageEntity send(MktSendMailDTO dto, Long operatorId);
 

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/service/MktMailReceiveService.java

@@ -0,0 +1,14 @@
+package com.storlead.sales.mail.mkt.service;
+
+import com.storlead.sales.mail.common.entity.SmtpPopSettingsEntity;
+
+/**
+ * 营销邮件收信（落 mkt_mail_message，folder=INBOX/SENT）。
+ */
+public interface MktMailReceiveService {
+
+    /**
+     * 异步收信：拉头 + 补正文（含附件元数据），立即返回。
+     */
+    void receiveEmails(SmtpPopSettingsEntity account);
+}

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

@@ -11,6 +11,7 @@ import com.storlead.framework.web.exception.WebException;
 import com.storlead.sales.mail.common.entity.MailTempAttachmentEntity;
 import com.storlead.sales.mail.common.entity.SmtpPopSettingsEntity;
 import com.storlead.sales.mail.common.enums.MailSceneEnum;
+import com.storlead.sales.mail.common.enums.MailSendStatus;
 import com.storlead.sales.mail.common.properties.MailFileProperties;
 import com.storlead.sales.mail.common.service.MailAccountBizService;
 import com.storlead.sales.mail.common.service.MailTempAttachmentService;
@@ -23,10 +24,14 @@ import com.storlead.sales.mail.mkt.pojo.MktMailQueryDTO;
 import com.storlead.sales.mail.mkt.pojo.MktSendMailDTO;
 import com.storlead.sales.mail.mkt.service.MktMailAttachmentService;
 import com.storlead.sales.mail.mkt.service.MktMailMessageService;
+import com.storlead.sales.mail.mkt.enums.MktMailFolderEnum;
+import com.storlead.sales.mail.mkt.util.MktMailSender;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
@@ -38,6 +43,7 @@ import javax.mail.BodyPart;
 import javax.mail.Message;
 import javax.mail.Multipart;
 import javax.mail.Session;
+import javax.mail.Transport;
 import javax.mail.internet.InternetAddress;
 import javax.mail.internet.MimeBodyPart;
 import javax.mail.internet.MimeMessage;
@@ -47,6 +53,7 @@ import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 /**
@@ -80,6 +87,9 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
     @Resource
     private MailFileProperties mailFileProperties;
 
+    @Resource
+    private MktMailSender mktMailSender;
+
     /**
      * 打开追踪像素根地址，需可被收件方公网访问，例如 https://host/router/mail
      */
@@ -96,6 +106,8 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         MktMailMessageEntity entity = resolveWritableEntity(dto.getId(), operatorId);
         fillMessageFields(entity, dto, account, operatorId);
         entity.setStatus(STATUS_DRAFT);
+        entity.setFolder(MktMailFolderEnum.DRAFT.code);
+        entity.setInOutMark(2);
         entity.setDelaySendTime(dto.getDelaySendTime());
         ensureTrackToken(entity, dto.getIsTrack());
         this.saveOrUpdate(entity);
@@ -123,6 +135,8 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         ensureTrackToken(entity, dto.getIsTrack());
 
         boolean delayed = dto.getDelaySendTime() != null && dto.getDelaySendTime().after(new Date());
+        entity.setFolder(MktMailFolderEnum.SENTING.code);
+        entity.setInOutMark(2);
         if (delayed) {
             entity.setStatus(STATUS_DELAY);
             entity.setDelaySendTime(dto.getDelaySendTime());
@@ -139,7 +153,21 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         // 正式附件拷入临时表，MIME 与成功后转正式统一按 mailId 处理，避免重复挂载
         copyFormalToTemp(entity.getId(), dto.getOldFileIds(), operatorId);
 
-        doSmtpSend(entity, decryptAccount(account));
+        SmtpPopSettingsEntity decrypted = decryptAccount(account);
+        boolean async = Boolean.TRUE.equals(dto.getAsync());
+        if (async) {
+            dispatchSmtpSend(entity.getId(), decrypted, true);
+            return entity;
+        }
+        MailSendStatus status = dispatchSmtpSend(entity.getId(), decrypted, false);
+        MktMailMessageEntity latest = this.getById(entity.getId());
+        if (latest != null) {
+            if (status == MailSendStatus.SERVER_LOGIN_FAIL || status == MailSendStatus.FAILED
+                    || status == MailSendStatus.SERVER_EXCEPTION_550) {
+                // 状态已由 MktMailSender 回调回写
+            }
+            return latest;
+        }
         return entity;
     }
 
@@ -170,13 +198,18 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
             }
             try {
                 SmtpPopSettingsEntity account = mailAccountBizService.requireAccount(entity.getAccountId(), MailSceneEnum.MKT);
-                doSmtpSend(entity, decryptAccount(account));
-                success++;
+                MailSendStatus status = dispatchSmtpSend(entity.getId(), decryptAccount(account), false);
+                if (status == MailSendStatus.SUCCESS || status == MailSendStatus.PARTIAL_SUCCESS) {
+                    success++;
+                }
             } catch (Exception ex) {
                 log.error("mkt delay send fail, id={}", item.getId(), ex);
-                entity.setStatus(STATUS_FAIL);
-                entity.setFailReason(ex.getMessage());
-                this.updateById(entity);
+                LambdaUpdateWrapper<MktMailMessageEntity> fail = new LambdaUpdateWrapper<>();
+                fail.eq(MktMailMessageEntity::getId, item.getId());
+                fail.eq(MktMailMessageEntity::getStatus, STATUS_SENDING);
+                fail.set(MktMailMessageEntity::getStatus, STATUS_FAIL);
+                fail.set(MktMailMessageEntity::getFailReason, ex.getMessage());
+                this.update(fail);
             }
         }
         return success;
@@ -199,6 +232,9 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         if (StringUtils.hasText(query.getRecipient())) {
             wrapper.like(MktMailMessageEntity::getRecipient, query.getRecipient());
         }
+        if (StringUtils.hasText(query.getFolder())) {
+            wrapper.eq(MktMailMessageEntity::getFolder, query.getFolder());
+        }
         wrapper.orderByDesc(MktMailMessageEntity::getId);
         return this.page(page, wrapper);
     }
@@ -250,26 +286,67 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         return this.updateById(entity);
     }
 
-    private void doSmtpSend(MktMailMessageEntity entity, SmtpPopSettingsEntity account) {
+    /**
+     * 对齐 CRM：Session + Transport + MktMailSender（TransportListener 回调）。
+     *
+     * @param async true 时事务提交后异步投递并立即返回；false 同步等待结果
+     */
+    private MailSendStatus dispatchSmtpSend(Long mailId, SmtpPopSettingsEntity account, boolean async) {
+        if (async) {
+            Runnable task = () -> {
+                try {
+                    MktMailMessageEntity entity = this.getById(mailId);
+                    if (entity == null || !Integer.valueOf(STATUS_SENDING).equals(entity.getStatus())) {
+                        return;
+                    }
+                    Session session = MailTransportSupport.createSmtpSession(account);
+                    Transport transport = session.getTransport("smtp");
+                    MimeMessage message = buildMimeMessage(session, entity, account);
+                    mktMailSender.asyncSendMailMessage(transport, message, entity);
+                } catch (Exception e) {
+                    log.error("mkt async prepare error, mailId={}", mailId, e);
+                    markSendingFailed(mailId, e.getMessage());
+                }
+            };
+            if (TransactionSynchronizationManager.isSynchronizationActive()) {
+                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                    @Override
+                    public void afterCommit() {
+                        task.run();
+                    }
+                });
+            } else {
+                task.run();
+            }
+            return MailSendStatus.SUCCESS;
+        }
+
         try {
+            MktMailMessageEntity entity = this.getById(mailId);
+            if (entity == null) {
+                return MailSendStatus.FAILED;
+            }
             Session session = MailTransportSupport.createSmtpSession(account);
+            Transport transport = session.getTransport("smtp");
             MimeMessage message = buildMimeMessage(session, entity, account);
-            MailTransportSupport.send(message);
-
-            entity.setMessageId(message.getMessageID());
-            entity.setStatus(STATUS_SUCCESS);
-            entity.setSentTime(new Date());
-            entity.setFailReason(null);
-            this.updateById(entity);
-            promoteTempAttachments(entity);
-        } catch (Exception ex) {
-            entity.setStatus(STATUS_FAIL);
-            entity.setFailReason(ex.getMessage());
-            this.updateById(entity);
-            throw new WebException("营销邮件发送失败: " + ex.getMessage(), ex);
+            CompletableFuture<MailSendStatus> future = mktMailSender.sendMailMessage(transport, message, entity);
+            return future.get();
+        } catch (Exception e) {
+            log.error("mkt sync send error, mailId={}", mailId, e);
+            markSendingFailed(mailId, e.getMessage());
+            return MailSendStatus.FAILED;
         }
     }
 
+    private void markSendingFailed(Long mailId, String reason) {
+        LambdaUpdateWrapper<MktMailMessageEntity> fail = new LambdaUpdateWrapper<>();
+        fail.eq(MktMailMessageEntity::getId, mailId);
+        fail.eq(MktMailMessageEntity::getStatus, STATUS_SENDING);
+        fail.set(MktMailMessageEntity::getStatus, STATUS_FAIL);
+        fail.set(MktMailMessageEntity::getFailReason, reason);
+        this.update(fail);
+    }
+
     private MimeMessage buildMimeMessage(Session session,
                                          MktMailMessageEntity entity,
                                          SmtpPopSettingsEntity account) throws Exception {
@@ -343,34 +420,6 @@ public class MktMailMessageServiceImpl extends MyBaseServiceImpl<MktMailMessageM
         return html + pixel;
     }
 
-    private void promoteTempAttachments(MktMailMessageEntity entity) {
-        List<MailTempAttachmentEntity> temps = tempAttachmentService.list(
-                new LambdaQueryWrapper<MailTempAttachmentEntity>()
-                        .eq(MailTempAttachmentEntity::getEmailId, entity.getId())
-                        .eq(MailTempAttachmentEntity::getScene, MailSceneEnum.MKT.getCode()));
-        if (CollectionUtils.isEmpty(temps)) {
-            return;
-        }
-        List<MktMailAttachmentEntity> formals = new ArrayList<>();
-        for (MailTempAttachmentEntity temp : temps) {
-            MktMailAttachmentEntity formal = new MktMailAttachmentEntity();
-            formal.setMailId(entity.getId());
-            formal.setAccountId(entity.getAccountId());
-            formal.setFileCode(temp.getFileCode());
-            formal.setFileName(temp.getFileName());
-            formal.setFilePath(temp.getFilePath());
-            formal.setFileSize(temp.getFileSize());
-            formal.setFileExt(temp.getFileExt());
-            formal.setOwnerBy(entity.getOwnerBy());
-            formal.setCreateBy(entity.getCreateBy());
-            formal.setIsDelete(CommonConstant.DEL_FLAG_0);
-            formal.setEnabled(Boolean.TRUE);
-            formals.add(formal);
-        }
-        mktMailAttachmentService.saveBatch(formals);
-        tempAttachmentService.removeByIds(temps.stream().map(MailTempAttachmentEntity::getId).collect(Collectors.toList()));
-    }
-
     private void bindTempAttachments(Long mailId, List<Long> fileIds, Long operatorId) {
         if (CollectionUtils.isEmpty(fileIds) || mailId == null) {
             return;

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/service/impl/MktMailReceiveServiceImpl.java

@@ -0,0 +1,513 @@
+package com.storlead.sales.mail.mkt.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.storlead.framework.common.constant.CommonConstant;
+import com.storlead.framework.common.util.DateUtil;
+import com.storlead.framework.common.util.DateUtils;
+import com.storlead.framework.common.util.RandomGenerateHelper;
+import com.storlead.sales.mail.common.connection.MailConnection;
+import com.storlead.sales.mail.common.connection.client.MailConnectionUtil;
+import com.storlead.sales.mail.common.connection.config.MailProperties;
+import com.storlead.sales.mail.common.entity.SmtpPopSettingsEntity;
+import com.storlead.sales.mail.common.properties.MailFileProperties;
+import com.storlead.sales.mail.common.util.MailReceiveQueueThreadPool;
+import com.storlead.sales.mail.mkt.entity.MktMailAttachmentEntity;
+import com.storlead.sales.mail.mkt.entity.MktMailMessageEntity;
+import com.storlead.sales.mail.mkt.enums.MktMailFolderEnum;
+import com.storlead.sales.mail.mkt.service.MktMailAttachmentService;
+import com.storlead.sales.mail.mkt.service.MktMailMessageService;
+import com.storlead.sales.mail.mkt.service.MktMailReceiveService;
+import com.sun.mail.imap.IMAPFolder;
+import com.sun.mail.pop3.POP3Folder;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.io.FilenameUtils;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.Resource;
+import javax.mail.Address;
+import javax.mail.BodyPart;
+import javax.mail.Flags;
+import javax.mail.Folder;
+import javax.mail.Message;
+import javax.mail.Multipart;
+import javax.mail.Part;
+import javax.mail.internet.InternetAddress;
+import javax.mail.internet.MimeUtility;
+import javax.mail.search.ComparisonTerm;
+import javax.mail.search.ReceivedDateTerm;
+import javax.mail.search.SearchTerm;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * 营销收信：对齐 CRM receiveEmails，落 mkt_mail_message（folder 区分），并绑定 reply_msg_id。
+ */
+@Log4j2
+@Service
+public class MktMailReceiveServiceImpl implements MktMailReceiveService {
+
+    private static final int STATUS_RECEIVED = 1;
+
+    @Resource
+    private MktMailMessageService mktMailMessageService;
+
+    @Resource
+    private MktMailAttachmentService mktMailAttachmentService;
+
+    @Resource
+    private MailFileProperties mailFileProperties;
+
+    @Override
+    public void receiveEmails(SmtpPopSettingsEntity account) {
+        if (account == null || account.getId() == null) {
+            return;
+        }
+        if (account.getEnableReceive() != null && !Integer.valueOf(1).equals(account.getEnableReceive())) {
+            log.warn("mkt receive skipped, enable_receive!=1, accountId={}", account.getId());
+            return;
+        }
+        String taskKey = "mkt_" + account.getEmailAddress() + "_" + account.getId();
+        MailReceiveQueueThreadPool pool = MailReceiveQueueThreadPool.getInstance();
+        if (Boolean.TRUE.equals(pool.getTaskInProgressByMail(taskKey))) {
+            log.warn("mkt receive already running: {}", taskKey);
+            return;
+        }
+        pool.addTask(taskKey, () -> {
+            try {
+                receiveHeadEmails(account);
+                loadMailContents(account);
+            } catch (Exception e) {
+                log.error("mkt receiveEmails error, accountId={}", account.getId(), e);
+            }
+        });
+    }
+
+    private void receiveHeadEmails(SmtpPopSettingsEntity account) {
+        pullFolder(account, "INBOX", MktMailFolderEnum.INBOX);
+        pullFolder(account, "Sent Items", MktMailFolderEnum.SENT);
+    }
+
+    private void pullFolder(SmtpPopSettingsEntity account, String serverFolder, MktMailFolderEnum folderEnum) {
+        try {
+            List<Message> messages = searchMessages(account, serverFolder, folderEnum);
+            if (CollectionUtils.isEmpty(messages)) {
+                return;
+            }
+            List<String> recentIds = listRecentMessageIds(account.getId(), folderEnum.code);
+            for (Message message : messages) {
+                try {
+                    String[] headers = message.getHeader("Message-ID");
+                    if (headers == null || headers.length == 0 || !StringUtils.hasText(headers[0])) {
+                        continue;
+                    }
+                    String messageId = normalizeMessageId(headers[0]);
+                    if (recentIds.contains(messageId) || existsByMessageId(account.getId(), messageId)) {
+                        continue;
+                    }
+                    MktMailMessageEntity entity = convertMessage(message, folderEnum);
+                    if (entity == null) {
+                        continue;
+                    }
+                    entity.setMessageId(messageId);
+                    entity.setAccountId(account.getId());
+                    entity.setOrgId(account.getOrgId());
+                    entity.setOwnerBy(account.getOwnerBy());
+                    entity.setCreateBy(account.getOwnerBy());
+                    entity.setStatus(STATUS_RECEIVED);
+                    entity.setIsDelete(CommonConstant.DEL_FLAG_0);
+                    entity.setEnabled(Boolean.TRUE);
+                    entity.setIsTrack(0);
+                    entity.setOpened(0);
+                    entity.setOpenCount(0);
+                    fillMsgUid(message, account, entity);
+                    bindReply(account.getId(), message, entity);
+
+                    if (mktMailMessageService.save(entity)) {
+                        recentIds.add(messageId);
+                    }
+                } catch (Exception ex) {
+                    log.error("mkt pull one mail error, accountId={}", account.getId(), ex);
+                }
+            }
+        } catch (Exception e) {
+            log.error("mkt pullFolder error, folder={}, accountId={}", serverFolder, account.getId(), e);
+        }
+    }
+
+    private List<Message> searchMessages(SmtpPopSettingsEntity account, String folderName, MktMailFolderEnum folderEnum) {
+        try {
+            QueryWrapper<MktMailMessageEntity> qw = new QueryWrapper<>();
+            qw.eq("account_id", account.getId());
+            qw.eq("folder", folderEnum.code);
+            qw.select("max(recipient_date) as recipient_date");
+            MktMailMessageEntity latest = mktMailMessageService.getOne(qw);
+
+            MailProperties properties = new MailProperties(account);
+            MailConnection connection = MailConnectionUtil.receiveEmailsConnection(properties);
+            if (connection == null) {
+                return null;
+            }
+            Folder inbox = connection.getFolder(folderName);
+            if (inbox == null) {
+                return null;
+            }
+            inbox.open(Folder.READ_ONLY);
+
+            Date beginDate = latest == null ? null : latest.getRecipientDate();
+            if (beginDate == null) {
+                LocalDate currentDate = LocalDate.now();
+                if (account.getPullDay() == null || Integer.valueOf(0).equals(account.getPullDay())) {
+                    beginDate = DateUtil.getDateByFormat(currentDate.minusYears(1).toString(), "yyyy-MM-dd");
+                } else {
+                    beginDate = DateUtil.getDateByFormat(currentDate.minusDays(account.getPullDay()).toString(), "yyyy-MM-dd");
+                }
+            }
+            SearchTerm searchTerm = new ReceivedDateTerm(ComparisonTerm.GT, beginDate);
+            Message[] messages = inbox.search(searchTerm);
+            if (messages == null || messages.length == 0) {
+                return null;
+            }
+            return Arrays.asList(messages);
+        } catch (Exception e) {
+            log.error("mkt searchMessages error, folder={}", folderName, e);
+            return null;
+        }
+    }
+
+    private void loadMailContents(SmtpPopSettingsEntity account) {
+        List<MktMailMessageEntity> heads = mktMailMessageService.list(new LambdaQueryWrapper<MktMailMessageEntity>()
+                .eq(MktMailMessageEntity::getAccountId, account.getId())
+                .eq(MktMailMessageEntity::getIsOnlyHead, 1)
+                .eq(MktMailMessageEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                .in(MktMailMessageEntity::getFolder, MktMailFolderEnum.INBOX.code, MktMailFolderEnum.SENT.code));
+        if (CollectionUtils.isEmpty(heads)) {
+            return;
+        }
+        try {
+            MailProperties properties = new MailProperties(account);
+            MailConnection connection = MailConnectionUtil.receiveEmailsConnection(properties);
+            if (connection == null) {
+                return;
+            }
+            for (MktMailMessageEntity mail : heads) {
+                try {
+                    if (!StringUtils.hasText(mail.getMsgUid())) {
+                        continue;
+                    }
+                    String serverFolder = MktMailFolderEnum.INBOX.code.equals(mail.getFolder()) ? "INBOX" : "Sent Items";
+                    Message message = connection.getMessageByUid(mail.getMsgUid(), serverFolder);
+                    if (message == null) {
+                        continue;
+                    }
+                    String body = extractBody(message);
+                    LambdaUpdateWrapper<MktMailMessageEntity> update = new LambdaUpdateWrapper<>();
+                    update.eq(MktMailMessageEntity::getId, mail.getId());
+                    update.set(MktMailMessageEntity::getContent, body);
+                    update.set(MktMailMessageEntity::getIsOnlyHead, 0);
+                    mktMailMessageService.update(update);
+                    saveAttachments(mail, message, account);
+                } catch (Exception ex) {
+                    log.error("mkt load content error, mailId={}", mail.getId(), ex);
+                }
+            }
+        } catch (Exception e) {
+            log.error("mkt loadMailContents error, accountId={}", account.getId(), e);
+        }
+    }
+
+    private void bindReply(Long accountId, Message message, MktMailMessageEntity entity) {
+        try {
+            String inReplyTo = firstHeader(message, "In-Reply-To");
+            if (!StringUtils.hasText(inReplyTo)) {
+                String references = firstHeader(message, "References");
+                if (StringUtils.hasText(references)) {
+                    String[] parts = references.trim().split("\\s+");
+                    inReplyTo = parts[parts.length - 1];
+                }
+            }
+            if (!StringUtils.hasText(inReplyTo)) {
+                return;
+            }
+            String normalized = normalizeMessageId(inReplyTo);
+            entity.setInReplyTo(normalized);
+            MktMailMessageEntity parent = findByMessageId(accountId, normalized);
+            if (parent != null) {
+                entity.setReplyMsgId(parent.getId());
+                // 继承业务单号，便于营销侧回查
+                if (!StringUtils.hasText(entity.getBizRef()) && StringUtils.hasText(parent.getBizRef())) {
+                    entity.setBizRef(parent.getBizRef());
+                }
+                if (!StringUtils.hasText(entity.getBizBatchNo()) && StringUtils.hasText(parent.getBizBatchNo())) {
+                    entity.setBizBatchNo(parent.getBizBatchNo());
+                }
+            }
+        } catch (Exception e) {
+            log.error("mkt bindReply error", e);
+        }
+    }
+
+    private MktMailMessageEntity findByMessageId(Long accountId, String messageId) {
+        if (!StringUtils.hasText(messageId)) {
+            return null;
+        }
+        String normalized = normalizeMessageId(messageId);
+        List<MktMailMessageEntity> list = mktMailMessageService.list(new LambdaQueryWrapper<MktMailMessageEntity>()
+                .eq(MktMailMessageEntity::getAccountId, accountId)
+                .eq(MktMailMessageEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
+                .and(w -> w.eq(MktMailMessageEntity::getMessageId, normalized)
+                        .or().eq(MktMailMessageEntity::getMessageId, "<" + stripBrackets(normalized) + ">")
+                        .or().eq(MktMailMessageEntity::getMessageId, stripBrackets(normalized)))
+                .orderByDesc(MktMailMessageEntity::getId)
+                .last("limit 1"));
+        return CollectionUtils.isEmpty(list) ? null : list.get(0);
+    }
+
+    private boolean existsByMessageId(Long accountId, String messageId) {
+        return findByMessageId(accountId, messageId) != null;
+    }
+
+    private List<String> listRecentMessageIds(Long accountId, String folder) {
+        QueryWrapper<MktMailMessageEntity> qw = new QueryWrapper<>();
+        qw.select("message_id");
+        qw.eq("folder", folder);
+        qw.eq("account_id", accountId);
+        qw.isNotNull("message_id");
+        qw.last("order by recipient_date desc limit 200");
+        List<Object> objs = mktMailMessageService.listObjs(qw);
+        if (CollectionUtils.isEmpty(objs)) {
+            return new ArrayList<>();
+        }
+        return objs.stream().filter(Objects::nonNull).map(o -> normalizeMessageId(String.valueOf(o))).collect(Collectors.toList());
+    }
+
+    private MktMailMessageEntity convertMessage(Message message, MktMailFolderEnum folderEnum) {
+        try {
+            MktMailMessageEntity entity = new MktMailMessageEntity();
+            String subject = message.getSubject() == null ? "" : MimeUtility.decodeText(message.getSubject());
+            entity.setSubject(subject);
+            entity.setFolder(folderEnum.code);
+            entity.setInOutMark(folderEnum == MktMailFolderEnum.INBOX ? 1 : 2);
+
+            Map<String, String> fromMap = decodeAddresses(message.getFrom());
+            entity.setFromAddr(joinKeys(fromMap));
+            entity.setFromName(joinValues(fromMap));
+
+            Map<String, String> toMap = decodeAddresses(message.getRecipients(Message.RecipientType.TO));
+            String recipient = joinKeys(toMap);
+            entity.setRecipient(StringUtils.hasText(recipient) ? recipient : "");
+
+            Map<String, String> ccMap = decodeAddresses(message.getRecipients(Message.RecipientType.CC));
+            entity.setRecipientCc(joinKeys(ccMap));
+
+            entity.setEmailSize((long) Math.max(message.getSize(), 0));
+            if (message.getSentDate() != null) {
+                entity.setSentDate(message.getSentDate());
+                entity.setSentTime(message.getSentDate());
+            }
+            if (message.getReceivedDate() != null) {
+                entity.setRecipientDate(message.getReceivedDate());
+            } else if (message.getSentDate() != null) {
+                entity.setRecipientDate(message.getSentDate());
+            } else {
+                entity.setRecipientDate(new Date());
+            }
+            entity.setIsRead(message.isSet(Flags.Flag.SEEN) ? 1 : 0);
+
+            try {
+                Object content = message.getContent();
+                if (content instanceof String) {
+                    entity.setIsOnlyHead(0);
+                    entity.setContent(content.toString());
+                } else {
+                    entity.setIsOnlyHead(1);
+                }
+            } catch (Exception e) {
+                entity.setIsOnlyHead(1);
+            }
+            return entity;
+        } catch (Exception e) {
+            log.error("mkt convertMessage error", e);
+            return null;
+        }
+    }
+
+    private void fillMsgUid(Message message, SmtpPopSettingsEntity account, MktMailMessageEntity entity) {
+        try {
+            if ("IMAP".equalsIgnoreCase(account.getProtocolType()) && message.getFolder() instanceof IMAPFolder) {
+                long uid = ((IMAPFolder) message.getFolder()).getUID(message);
+                entity.setMsgUid(String.valueOf(uid));
+            } else if ("POP3".equalsIgnoreCase(account.getProtocolType()) && message.getFolder() instanceof POP3Folder) {
+                entity.setMsgUid(((POP3Folder) message.getFolder()).getUID(message));
+            }
+        } catch (Exception e) {
+            log.error("mkt fillMsgUid error", e);
+        }
+    }
+
+    private void saveAttachments(MktMailMessageEntity mail, Message message, SmtpPopSettingsEntity account) {
+        try {
+            int exist = mktMailAttachmentService.count(new LambdaQueryWrapper<MktMailAttachmentEntity>()
+                    .eq(MktMailAttachmentEntity::getMailId, mail.getId())
+                    .eq(MktMailAttachmentEntity::getIsDelete, CommonConstant.DEL_FLAG_0));
+            if (exist > 0) {
+                return;
+            }
+            Object content = message.getContent();
+            if (!(content instanceof Multipart)) {
+                return;
+            }
+            String day = DateUtils.date2Str(
+                    mail.getRecipientDate() == null ? new Date() : mail.getRecipientDate(), DateUtils.yyyyMMdd);
+            String downloadDir = mailFileProperties.getPath().getPath() + File.separator
+                    + account.getEmailAddress() + File.separator + day + File.separator;
+            Multipart multipart = (Multipart) content;
+            for (int i = 0; i < multipart.getCount(); i++) {
+                BodyPart bodyPart = multipart.getBodyPart(i);
+                if (Part.ATTACHMENT.equalsIgnoreCase(bodyPart.getDisposition())
+                        || bodyPart.isMimeType("application/octet-stream")) {
+                    saveOneAttachment(bodyPart, downloadDir, mail, account);
+                }
+            }
+        } catch (Exception e) {
+            log.error("mkt saveAttachments error, mailId={}", mail.getId(), e);
+        }
+    }
+
+    private void saveOneAttachment(BodyPart bodyPart, String downloadDir, MktMailMessageEntity mail, SmtpPopSettingsEntity account) {
+        try {
+            String fileName = bodyPart.getFileName();
+            if (!StringUtils.hasText(fileName)) {
+                return;
+            }
+            fileName = MimeUtility.decodeText(fileName).replace("\n", "").replace("\r", "");
+            String code = RandomGenerateHelper.generateRandomString(6);
+            if (!new File(downloadDir).exists()) {
+                new File(downloadDir).mkdirs();
+            }
+            String storedName = fileName + "." + code;
+            File file = new File(downloadDir + File.separator + storedName);
+            try (FileOutputStream output = new FileOutputStream(file)) {
+                bodyPart.getInputStream().transferTo(output);
+            }
+            MktMailAttachmentEntity att = new MktMailAttachmentEntity();
+            att.setMailId(mail.getId());
+            att.setAccountId(account.getId());
+            att.setFileCode(code);
+            att.setFileName(FilenameUtils.getBaseName(fileName));
+            att.setFileExt(FilenameUtils.getExtension(fileName));
+            att.setFilePath(file.getAbsolutePath().replace(mailFileProperties.getPath().getPath(), ""));
+            att.setFileSize(file.length());
+            att.setOwnerBy(mail.getOwnerBy());
+            att.setCreateBy(mail.getOwnerBy());
+            att.setIsDelete(CommonConstant.DEL_FLAG_0);
+            att.setEnabled(Boolean.TRUE);
+            mktMailAttachmentService.save(att);
+        } catch (Exception e) {
+            log.error("mkt saveOneAttachment error", e);
+        }
+    }
+
+    private static String extractBody(Message message) {
+        try {
+            Object content = message.getContent();
+            if (content instanceof String) {
+                return content.toString();
+            }
+            if (content instanceof Multipart) {
+                return parseMultipart((Multipart) content);
+            }
+        } catch (Exception e) {
+            log.error("mkt extractBody error", e);
+        }
+        return "";
+    }
+
+    private static String parseMultipart(Multipart multipart) throws Exception {
+        StringBuilder html = new StringBuilder();
+        StringBuilder plain = new StringBuilder();
+        for (int i = 0; i < multipart.getCount(); i++) {
+            BodyPart part = multipart.getBodyPart(i);
+            Object content = part.getContent();
+            if (content instanceof Multipart) {
+                String nested = parseMultipart((Multipart) content);
+                html.append(nested);
+            } else if (part.isMimeType("text/html")) {
+                html.append(content == null ? "" : content.toString());
+            } else if (part.isMimeType("text/plain")) {
+                plain.append(content == null ? "" : content.toString());
+            }
+        }
+        return html.length() > 0 ? html.toString() : plain.toString();
+    }
+
+    private static Map<String, String> decodeAddresses(Address[] addresses) throws Exception {
+        Map<String, String> map = new HashMap<>();
+        if (addresses == null) {
+            return map;
+        }
+        for (Address address : addresses) {
+            InternetAddress internetAddress = (InternetAddress) address;
+            String personal = internetAddress.getPersonal();
+            if (personal != null) {
+                personal = MimeUtility.decodeText(personal);
+            }
+            map.put(internetAddress.getAddress(), personal != null ? personal : "N/A");
+        }
+        return map;
+    }
+
+    private static String joinKeys(Map<String, String> map) {
+        if (map == null || map.isEmpty()) {
+            return "";
+        }
+        return String.join(",", map.keySet());
+    }
+
+    private static String joinValues(Map<String, String> map) {
+        if (map == null || map.isEmpty()) {
+            return "";
+        }
+        return map.values().stream()
+                .filter(v -> v != null && !"N/A".equals(v))
+                .collect(Collectors.joining(","));
+    }
+
+    private static String firstHeader(Message message, String name) throws Exception {
+        String[] values = message.getHeader(name);
+        if (values == null || values.length == 0) {
+            return null;
+        }
+        return values[0];
+    }
+
+    private static String normalizeMessageId(String messageId) {
+        if (!StringUtils.hasText(messageId)) {
+            return messageId;
+        }
+        return messageId.trim();
+    }
+
+    private static String stripBrackets(String messageId) {
+        if (!StringUtils.hasText(messageId)) {
+            return messageId;
+        }
+        String v = messageId.trim();
+        if (v.startsWith("<") && v.endsWith(">") && v.length() > 2) {
+            return v.substring(1, v.length() - 1);
+        }
+        return v;
+    }
+}

storlead-mail/storlead-mail-mkt/src/main/java/com/storlead/sales/mail/mkt/util/MktMailSender.java

@@ -0,0 +1,301 @@
+package com.storlead.sales.mail.mkt.util;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.storlead.framework.common.constant.CommonConstant;
+import com.storlead.sales.mail.common.entity.MailTempAttachmentEntity;
+import com.storlead.sales.mail.common.enums.MailSceneEnum;
+import com.storlead.sales.mail.common.enums.MailSendStatus;
+import com.storlead.sales.mail.common.service.MailTempAttachmentService;
+import com.storlead.sales.mail.mkt.callback.MktMailSendCallback;
+import com.storlead.sales.mail.mkt.entity.MktMailAttachmentEntity;
+import com.storlead.sales.mail.mkt.entity.MktMailMessageEntity;
+import com.storlead.sales.mail.mkt.service.MktMailAttachmentService;
+import com.storlead.sales.mail.mkt.service.MktMailMessageService;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.Resource;
+import javax.mail.AuthenticationFailedException;
+import javax.mail.Message;
+import javax.mail.MessagingException;
+import javax.mail.SendFailedException;
+import javax.mail.Transport;
+import javax.mail.event.TransportEvent;
+import javax.mail.event.TransportListener;
+import javax.mail.internet.MimeMessage;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * 营销邮件 SMTP 发送器：对齐 CRM {@code EmailSenderWithThreadLocal}，
+ * 使用 TransportListener 回调更新状态，并支持同步 / 异步两种调用方式。
+ */
+@Log4j2
+@Component
+public class MktMailSender {
+
+    private static final int STATUS_SENDING = 0;
+    private static final int STATUS_SUCCESS = 1;
+    private static final int STATUS_FAIL = 2;
+
+    @Lazy
+    @Resource
+    private MktMailMessageService mktMailMessageService;
+
+    @Resource
+    private MktMailAttachmentService mktMailAttachmentService;
+
+    @Resource
+    private MailTempAttachmentService tempAttachmentService;
+
+    private final List<MktMailSendCallback> sendCallbacks;
+
+    public MktMailSender(ObjectProvider<MktMailSendCallback> callbackProvider) {
+        this.sendCallbacks = callbackProvider.orderedStream().collect(Collectors.toList());
+    }
+
+    /**
+     * 异步发送：立即返回，结果由 TransportListener / 兜底逻辑回写库并触发业务回调。
+     */
+    public void asyncSendMailMessage(Transport transport, Message message, MktMailMessageEntity mail) {
+        CompletableFuture.runAsync(() -> {
+            try {
+                doSend(transport, message, mail, null);
+            } catch (Exception e) {
+                log.error("mkt async send error, mailId={}", mail == null ? null : mail.getId(), e);
+                onEmailSentFailure(mail, e.getMessage());
+            }
+        });
+    }
+
+    /**
+     * 同步发送：等待投递结果，返回 {@link MailSendStatus}。
+     */
+    public CompletableFuture<MailSendStatus> sendMailMessage(Transport transport, Message message, MktMailMessageEntity mail) {
+        CompletableFuture<MailSendStatus> future = new CompletableFuture<>();
+        try {
+            doSend(transport, message, mail, future);
+        } catch (Exception e) {
+            log.error("mkt sync send error, mailId={}", mail == null ? null : mail.getId(), e);
+            if (!future.isDone()) {
+                onEmailSentFailure(mail, e.getMessage());
+                future.complete(MailSendStatus.FAILED);
+            }
+        }
+        return future;
+    }
+
+    private void doSend(Transport transport,
+                        Message message,
+                        MktMailMessageEntity mail,
+                        CompletableFuture<MailSendStatus> future) {
+        TransportListener listener = new TransportListener() {
+            @Override
+            public void messageDelivered(TransportEvent e) {
+                onEmailSentSuccess(mail, safeMessageId(message));
+                completeQuietly(future, MailSendStatus.SUCCESS);
+            }
+
+            @Override
+            public void messageNotDelivered(TransportEvent e) {
+                onEmailSentFailure(mail, "messageNotDelivered");
+                completeQuietly(future, MailSendStatus.FAILED);
+            }
+
+            @Override
+            public void messagePartiallyDelivered(TransportEvent e) {
+                onEmailSentPartialSuccess(mail, safeMessageId(message));
+                completeQuietly(future, MailSendStatus.PARTIAL_SUCCESS);
+            }
+        };
+
+        try {
+            transport.addTransportListener(listener);
+            transport.connect();
+            transport.sendMessage(message, message.getAllRecipients());
+            transport.close();
+            // 部分 SMTP 不触发 delivered 回调：无异常且仍为发送中则视为成功
+            if (future != null && !future.isDone()) {
+                onEmailSentSuccess(mail, safeMessageId(message));
+                future.complete(MailSendStatus.SUCCESS);
+            } else if (future == null) {
+                onEmailSentSuccess(mail, safeMessageId(message));
+            }
+        } catch (SendFailedException e) {
+            log.error("mkt SendFailedException, mailId={}", mail == null ? null : mail.getId(), e);
+            onEmailSentFailure(mail, e.getMessage());
+            completeQuietly(future, MailSendStatus.SERVER_EXCEPTION_550);
+        } catch (AuthenticationFailedException e) {
+            log.error("mkt AuthenticationFailedException, mailId={}", mail == null ? null : mail.getId(), e);
+            onEmailSentFailure(mail, e.getMessage());
+            completeQuietly(future, MailSendStatus.SERVER_LOGIN_FAIL);
+        } catch (MessagingException e) {
+            log.error("mkt MessagingException, mailId={}", mail == null ? null : mail.getId(), e);
+            onEmailSentFailure(mail, e.getMessage());
+            completeQuietly(future, MailSendStatus.SERVER_LOGIN_FAIL);
+        }
+    }
+
+    private void onEmailSentSuccess(MktMailMessageEntity entity, String messageId) {
+        if (!markDelivered(entity, messageId, false)) {
+            return;
+        }
+        fireSuccess(entity);
+    }
+
+    private void onEmailSentPartialSuccess(MktMailMessageEntity entity, String messageId) {
+        if (!markDelivered(entity, messageId, true)) {
+            return;
+        }
+        firePartial(entity);
+    }
+
+    /**
+     * @return true 表示本次首次落成功态（可触发业务回调）
+     */
+    private boolean markDelivered(MktMailMessageEntity entity, String messageId, boolean partial) {
+        if (entity == null || entity.getId() == null) {
+            return false;
+        }
+        try {
+            Date now = new Date();
+            LambdaUpdateWrapper<MktMailMessageEntity> update = new LambdaUpdateWrapper<>();
+            update.eq(MktMailMessageEntity::getId, entity.getId());
+            update.eq(MktMailMessageEntity::getStatus, STATUS_SENDING);
+            update.set(MktMailMessageEntity::getStatus, STATUS_SUCCESS);
+            update.set(MktMailMessageEntity::getSentTime, now);
+            update.set(MktMailMessageEntity::getFailReason, null);
+            if (StringUtils.hasText(messageId)) {
+                update.set(MktMailMessageEntity::getMessageId, messageId);
+            }
+            boolean updated = mktMailMessageService.update(update);
+            if (!updated) {
+                return false;
+            }
+            entity.setStatus(STATUS_SUCCESS);
+            entity.setSentTime(now);
+            entity.setFailReason(null);
+            if (StringUtils.hasText(messageId)) {
+                entity.setMessageId(messageId);
+            }
+            promoteTempAttachments(entity);
+            return true;
+        } catch (Exception e) {
+            log.error("mkt markDelivered error, mailId={}, partial={}", entity.getId(), partial, e);
+            return false;
+        }
+    }
+
+    private void onEmailSentFailure(MktMailMessageEntity entity, String reason) {
+        if (entity == null || entity.getId() == null) {
+            return;
+        }
+        try {
+            String failReason = StringUtils.hasText(reason) ? truncate(reason, 500) : "send failed";
+            LambdaUpdateWrapper<MktMailMessageEntity> update = new LambdaUpdateWrapper<>();
+            update.eq(MktMailMessageEntity::getId, entity.getId());
+            update.eq(MktMailMessageEntity::getStatus, STATUS_SENDING);
+            update.set(MktMailMessageEntity::getStatus, STATUS_FAIL);
+            update.set(MktMailMessageEntity::getFailReason, failReason);
+            boolean updated = mktMailMessageService.update(update);
+            if (!updated) {
+                return;
+            }
+            entity.setStatus(STATUS_FAIL);
+            entity.setFailReason(failReason);
+            fireFailure(entity, failReason);
+        } catch (Exception e) {
+            log.error("mkt onEmailSentFailure error, mailId={}", entity.getId(), e);
+        }
+    }
+
+    private void promoteTempAttachments(MktMailMessageEntity entity) {
+        List<MailTempAttachmentEntity> temps = tempAttachmentService.list(
+                new LambdaQueryWrapper<MailTempAttachmentEntity>()
+                        .eq(MailTempAttachmentEntity::getEmailId, entity.getId())
+                        .eq(MailTempAttachmentEntity::getScene, MailSceneEnum.MKT.getCode()));
+        if (CollectionUtils.isEmpty(temps)) {
+            return;
+        }
+        List<MktMailAttachmentEntity> formals = new ArrayList<>();
+        for (MailTempAttachmentEntity temp : temps) {
+            MktMailAttachmentEntity formal = new MktMailAttachmentEntity();
+            formal.setMailId(entity.getId());
+            formal.setAccountId(entity.getAccountId());
+            formal.setFileCode(temp.getFileCode());
+            formal.setFileName(temp.getFileName());
+            formal.setFilePath(temp.getFilePath());
+            formal.setFileSize(temp.getFileSize());
+            formal.setFileExt(temp.getFileExt());
+            formal.setOwnerBy(entity.getOwnerBy());
+            formal.setCreateBy(entity.getCreateBy());
+            formal.setIsDelete(CommonConstant.DEL_FLAG_0);
+            formal.setEnabled(Boolean.TRUE);
+            formals.add(formal);
+        }
+        mktMailAttachmentService.saveBatch(formals);
+        tempAttachmentService.removeByIds(temps.stream().map(MailTempAttachmentEntity::getId).collect(Collectors.toList()));
+    }
+
+    private void fireSuccess(MktMailMessageEntity entity) {
+        for (MktMailSendCallback callback : sendCallbacks) {
+            try {
+                callback.onSuccess(entity);
+            } catch (Exception ex) {
+                log.error("mkt success callback error, mailId={}", entity.getId(), ex);
+            }
+        }
+    }
+
+    private void firePartial(MktMailMessageEntity entity) {
+        for (MktMailSendCallback callback : sendCallbacks) {
+            try {
+                callback.onPartialSuccess(entity);
+            } catch (Exception ex) {
+                log.error("mkt partial callback error, mailId={}", entity.getId(), ex);
+            }
+        }
+    }
+
+    private void fireFailure(MktMailMessageEntity entity, String reason) {
+        for (MktMailSendCallback callback : sendCallbacks) {
+            try {
+                callback.onFailure(entity, reason);
+            } catch (Exception ex) {
+                log.error("mkt failure callback error, mailId={}", entity.getId(), ex);
+            }
+        }
+    }
+
+    private static void completeQuietly(CompletableFuture<MailSendStatus> future, MailSendStatus status) {
+        if (future != null && !future.isDone()) {
+            future.complete(status);
+        }
+    }
+
+    private static String safeMessageId(Message message) {
+        try {
+            if (message instanceof MimeMessage) {
+                return ((MimeMessage) message).getMessageID();
+            }
+            return null;
+        } catch (Exception e) {
+            return null;
+        }
+    }
+
+    private static String truncate(String text, int max) {
+        if (text == null || text.length() <= max) {
+            return text;
+        }
+        return text.substring(0, max);
+    }
+}