邮件业务

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/controller/mkt/MktEmailCccountApiController.java → java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/controller/mkt/MktEmailAccountApiController.java

@@ -16,13 +16,13 @@ import org.springframework.web.bind.annotation.RestController;
 import javax.annotation.Resource;
 
 /**
- * 营销邮箱账户（转发独立邮件服务 /mkt/account/**）。
+ * 营销邮箱账户（路径前缀 /mkt/email/account，能力对齐 CRM {@code SmtpPopSettingsApiController}）。
  */
 @Log4j2
 @RestController
 @RequestMapping("/mkt/email/account")
 @Api(tags = "营销邮件: 邮箱账户")
-public class MktEmailCccountApiController {
+public class MktEmailAccountApiController {
 
     private static final String SCENE_MKT = "MKT";
     private static final String SCOPE_ORG = "ORG";
@@ -62,6 +62,30 @@ public class MktEmailCccountApiController {
         return mktAccountRemoteClient.defaultUse(id);
     }
 
+    @PostMapping("/setReplyContent")
+    @ApiOperation("设置自动回复内容")
+    public Result<?> setReplyContent(Long id, String replyContent, String replyTitle, Integer autoReplyStatus) {
+        return mktAccountRemoteClient.setReplyContent(id, replyContent, replyTitle, autoReplyStatus);
+    }
+
+    @PostMapping("/setAutoReplyStatus")
+    @ApiOperation("开关自动回复")
+    public Result<?> setAutoReplyStatus(Long id, Integer autoReplyStatus) {
+        return mktAccountRemoteClient.setAutoReplyStatus(id, autoReplyStatus);
+    }
+
+    @PostMapping("/isEnabledDelete")
+    @ApiOperation("是否开启删除服务器邮件")
+    public Result<?> isEnabledDelete(Long id, Integer isEnabled) {
+        return mktAccountRemoteClient.isEnabledDelete(id, isEnabled);
+    }
+
+    @PostMapping("/personal_mail_account")
+    @ApiOperation("营销邮箱账户（含未读数）")
+    public Result<?> personalMailAccount() {
+        return mktAccountRemoteClient.personalMailAccount();
+    }
+
     @PostMapping("/test_connect")
     @ApiOperation("测试营销邮箱连通性")
     public Result<?> testConnect(@RequestBody SmtpPopSettingsEntity mailSet) {

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/webclient/client/MktAccountRemoteClient.java

@@ -8,7 +8,7 @@ import com.storlead.mail.webclient.support.MailRemoteHttpSupport;
 import lombok.RequiredArgsConstructor;
 
 /**
- * 营销邮箱账户远程客户端，对齐 B 服务 {@code /mail/mkt/account/**}。
+ * 营销邮箱账户远程客户端，对齐 B 服务 {@code /mkt/account/**}（能力与 CRM 账户对齐）。
  */
 @RequiredArgsConstructor
 public class MktAccountRemoteClient {
@@ -35,6 +35,25 @@ public class MktAccountRemoteClient {
         return http.postForm(MailRemoteApiConstants.MKT_ACCOUNT_DEFAULT_USE, http.formOf("id", id));
     }
 
+    public Result<?> setReplyContent(Long id, String replyContent, String replyTitle, Integer autoReplyStatus) {
+        return http.postForm(MailRemoteApiConstants.MKT_ACCOUNT_SET_REPLY_CONTENT,
+                http.formOf("id", id, "replyContent", replyContent, "replyTitle", replyTitle, "autoReplyStatus", autoReplyStatus));
+    }
+
+    public Result<?> setAutoReplyStatus(Long id, Integer autoReplyStatus) {
+        return http.postForm(MailRemoteApiConstants.MKT_ACCOUNT_SET_AUTO_REPLY_STATUS,
+                http.formOf("id", id, "autoReplyStatus", autoReplyStatus));
+    }
+
+    public Result<?> isEnabledDelete(Long id, Integer isEnabled) {
+        return http.postForm(MailRemoteApiConstants.MKT_ACCOUNT_IS_ENABLED_DELETE,
+                http.formOf("id", id, "isEnabled", isEnabled));
+    }
+
+    public Result<?> personalMailAccount() {
+        return http.postForm(MailRemoteApiConstants.MKT_ACCOUNT_PERSONAL, http.formOf());
+    }
+
     public Result<?> testConnect(SmtpPopSettingsEntity mailSet) {
         return http.postJson(MailRemoteApiConstants.MKT_ACCOUNT_TEST_CONNECT, mailSet);
     }

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/webclient/client/MktMailRemoteClient.java

@@ -67,12 +67,21 @@ public class MktMailRemoteClient {
     }
 
     /**
-     * 按 status_updated_at 增量拉取营销邮件状态变更。
+     * 查询待同步状态：B 侧 update_time > status_updated_at。
      */
     public Result<List<MktMailStatusChangeRemoteDTO>> statusChanges(MktMailStatusSyncQueryRemoteDTO query) {
         return http.postJsonForList(MailRemoteApiConstants.MKT_STATUS_CHANGES, query, MktMailStatusChangeRemoteDTO.class);
     }
 
+    /**
+     * 同步成功后回写 mkt_mail_status.status_updated_at。
+     */
+    public Result<?> ackStatusSync(List<Long> mailIds) {
+        Map<String, Object> body = new HashMap<>();
+        body.put("mailIds", mailIds == null ? java.util.Collections.emptyList() : mailIds);
+        return http.postJson(MailRemoteApiConstants.MKT_STATUS_ACK_SYNC, body);
+    }
+
     /** 打开追踪像素（通常由收件端浏览器直接访问 B 服务）。 */
     public byte[] track(String token) {
         return http.getForBytes(MailRemoteApiConstants.MKT_TRACK_PREFIX + token, null);

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/webclient/constant/MailRemoteApiConstants.java

@@ -44,11 +44,18 @@ public final class MailRemoteApiConstants {
     public static final String MKT_OPEN_STATS = "/mkt/open_stats";
 
     /**
-     * 营销邮件状态增量变更（供 trade 同步 market_emails）。
+     * 查询 mkt_mail_status 待同步数据。
+     * B 条件：update_time > status_updated_at（有比同步时间更新的修改）。
      * JSON：MktMailStatusSyncQueryRemoteDTO；返回 List<MktMailStatusChangeRemoteDTO>
      */
     public static final String MKT_STATUS_CHANGES = "/mkt/status/changes";
 
+    /**
+     * 同步成功后回写 mkt_mail_status.status_updated_at（同步时间）。
+     * JSON：{ "mailIds": [1,2,3] }
+     */
+    public static final String MKT_STATUS_ACK_SYNC = "/mkt/status/ack_sync";
+
     /**
      * 打开追踪像素（免登录，返回 1x1 GIF）。
      * 完整路径：{@value #MKT_TRACK_PREFIX}{token}
@@ -72,6 +79,18 @@ public final class MailRemoteApiConstants {
     /** 设置默认营销发信账户，参数：id */
     public static final String MKT_ACCOUNT_DEFAULT_USE = "/mkt/account/default_use";
 
+    /** 设置营销邮箱自动回复内容，参数：id、replyContent、replyTitle、autoReplyStatus */
+    public static final String MKT_ACCOUNT_SET_REPLY_CONTENT = "/mkt/account/set_reply_content";
+
+    /** 开关营销邮箱自动回复，参数：id、autoReplyStatus */
+    public static final String MKT_ACCOUNT_SET_AUTO_REPLY_STATUS = "/mkt/account/set_auto_reply_status";
+
+    /** 是否开启删除服务器邮件，参数：id、isEnabled */
+    public static final String MKT_ACCOUNT_IS_ENABLED_DELETE = "/mkt/account/is_enabled_delete";
+
+    /** 营销邮箱账户列表（含未读数等，对齐 CRM personal_mail_account） */
+    public static final String MKT_ACCOUNT_PERSONAL = "/mkt/account/personal_mail_account";
+
     /** 测试营销邮箱连通性（以 SMTP 为主），JSON：SmtpPopSettingsEntity */
     public static final String MKT_ACCOUNT_TEST_CONNECT = "/mkt/account/test_connect";
 

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/webclient/dto/MktMailStatusChangeRemoteDTO.java

@@ -8,7 +8,7 @@ import org.springframework.format.annotation.DateTimeFormat;
 import java.time.LocalDateTime;
 
 /**
- * 营销邮件状态变更行（对齐 B 服务 mkt_mail_status 增量字段）。
+ * mkt_mail_status 待同步行（对齐邮件库状态表）。
  */
 @Data
 public class MktMailStatusChangeRemoteDTO {
@@ -33,6 +33,11 @@ public class MktMailStatusChangeRemoteDTO {
 
     @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
     @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    @ApiModelProperty("状态最后更新时间（增量游标）")
+    @ApiModelProperty("记录最新修改时间（与 status_updated_at 比对）")
+    private LocalDateTime updateTime;
+
+    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty("同步时间标记；同步成功后由 SaaS 回写刷新")
     private LocalDateTime statusUpdatedAt;
 }

java/storlead-mail/storlead-mail-api/src/main/java/com/storlead/mail/webclient/dto/MktMailStatusSyncQueryRemoteDTO.java

@@ -1,26 +1,19 @@
 package com.storlead.mail.webclient.dto;
 
-import com.fasterxml.jackson.annotation.JsonFormat;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
-import org.springframework.format.annotation.DateTimeFormat;
-
-import java.time.LocalDateTime;
 
 /**
- * 营销邮件状态增量查询（对齐 B 服务按 status_updated_at 游标拉取）。
+ * 查询 mkt_mail_status 待同步数据。
+ * <p>
+ * B 侧条件：{@code update_time > status_updated_at}（有比同步时间更新的修改）。
  */
 @Data
 public class MktMailStatusSyncQueryRemoteDTO {
 
-    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
-    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    @ApiModelProperty("游标时间：status_updated_at > since；首次为空表示从头拉")
-    private LocalDateTime since;
+    @ApiModelProperty("每批条数")
+    private Integer limit = 200;
 
-    @ApiModelProperty("同秒防漏：status_updated_at = since 时 id > lastId")
+    @ApiModelProperty("分页：上一批最后一条 mkt_mail_status.id")
     private Long lastId;
-
-    @ApiModelProperty("每批条数，默认 200")
-    private Integer limit = 200;
 }

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

@@ -1,41 +0,0 @@
-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.fasterxml.jackson.annotation.JsonFormat;
-import com.storlead.framework.mybatis.entity.SysBaseField;
-import io.swagger.annotations.ApiModelProperty;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import org.springframework.format.annotation.DateTimeFormat;
-
-import java.time.LocalDateTime;
-
-/**
- * 邮件状态同步游标（trade 库）。
- */
-@Data
-@EqualsAndHashCode(callSuper = true)
-@TableName("mail_status_sync_cursor")
-public class MailStatusSyncCursorEntity extends SysBaseField {
-
-    @TableId(type = IdType.AUTO)
-    @ApiModelProperty("主键")
-    private Long id;
-
-    @ApiModelProperty("同步源标识")
-    @TableField("source")
-    private String source;
-
-    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
-    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    @ApiModelProperty("已同步到的 status_updated_at")
-    @TableField("last_status_updated_at")
-    private LocalDateTime lastStatusUpdatedAt;
-
-    @ApiModelProperty("已同步到的 mkt_mail_status.id")
-    @TableField("last_id")
-    private Long lastId;
-}

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

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

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

@@ -1,19 +0,0 @@
-package com.storlead.trade.service;
-
-import com.storlead.framework.mybatis.service.MyBaseService;
-import com.storlead.trade.entity.MailStatusSyncCursorEntity;
-
-import java.time.LocalDateTime;
-
-public interface MailStatusSyncCursorService extends MyBaseService<MailStatusSyncCursorEntity> {
-
-    /**
-     * 获取或初始化指定 source 的游标。
-     */
-    MailStatusSyncCursorEntity getOrInit(String source);
-
-    /**
-     * 推进游标（本批成功写完 market_emails 后调用）。
-     */
-    void advance(String source, LocalDateTime lastStatusUpdatedAt, Long lastId);
-}

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

@@ -1,12 +1,16 @@
 package com.storlead.trade.service;
 
 /**
- * 将邮件服务 mkt_mail_status 增量同步到 trade.market_emails。
+ * 业务同步：mkt_mail_status → market_emails，再回写 status_updated_at。
  */
 public interface MarketEmailsStatusSyncService {
 
     /**
-     * 执行一轮（或多页）增量同步，返回本轮成功回写条数。
+     * 1. 查 mkt_mail_status（update_time &gt; status_updated_at）<br>
+     * 2. 按 mail_id 更新 market_emails<br>
+     * 3. 回写 mkt_mail_status.status_updated_at
+     *
+     * @return 实际更新 market_emails 条数
      */
     int syncOnce();
 }

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

@@ -1,48 +0,0 @@
-package com.storlead.trade.service.impl;
-
-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.trade.entity.MailStatusSyncCursorEntity;
-import com.storlead.trade.mapper.MailStatusSyncCursorMapper;
-import com.storlead.trade.service.MailStatusSyncCursorService;
-import com.storlead.trade.service.impl.support.TradeDataSourceServiceImpl;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-
-import java.time.LocalDateTime;
-
-@Service
-@DS(DSConstants.DATASOURCE_TRADE)
-public class MailStatusSyncCursorServiceImpl
-        extends TradeDataSourceServiceImpl<MailStatusSyncCursorMapper, MailStatusSyncCursorEntity>
-        implements MailStatusSyncCursorService {
-
-    @Override
-    public MailStatusSyncCursorEntity getOrInit(String source) {
-        MailStatusSyncCursorEntity cursor = this.getOne(new LambdaQueryWrapper<MailStatusSyncCursorEntity>()
-                .eq(MailStatusSyncCursorEntity::getSource, source)
-                .eq(MailStatusSyncCursorEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
-                .last("limit 1"));
-        if (cursor != null) {
-            return cursor;
-        }
-        cursor = new MailStatusSyncCursorEntity();
-        cursor.setSource(source);
-        cursor.setLastId(0L);
-        cursor.setIsDelete(CommonConstant.DEL_FLAG_0);
-        cursor.setEnabled(Boolean.TRUE);
-        this.save(cursor);
-        return cursor;
-    }
-
-    @Override
-    @Transactional(rollbackFor = Exception.class)
-    public void advance(String source, LocalDateTime lastStatusUpdatedAt, Long lastId) {
-        MailStatusSyncCursorEntity cursor = getOrInit(source);
-        cursor.setLastStatusUpdatedAt(lastStatusUpdatedAt);
-        cursor.setLastId(lastId);
-        this.updateById(cursor);
-    }
-}

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

@@ -9,9 +9,7 @@ import com.storlead.framework.common.result.Result;
 import com.storlead.mail.webclient.client.MktMailRemoteClient;
 import com.storlead.mail.webclient.dto.MktMailStatusChangeRemoteDTO;
 import com.storlead.mail.webclient.dto.MktMailStatusSyncQueryRemoteDTO;
-import com.storlead.trade.entity.MailStatusSyncCursorEntity;
 import com.storlead.trade.entity.MarketEmailsEntity;
-import com.storlead.trade.service.MailStatusSyncCursorService;
 import com.storlead.trade.service.MarketEmailsService;
 import com.storlead.trade.service.MarketEmailsStatusSyncService;
 import lombok.extern.slf4j.Slf4j;
@@ -19,23 +17,21 @@ import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
-import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.List;
 
 /**
- * 从邮件服务增量拉取 mkt_mail_status，按 email_id 回写 market_emails。
+ * mkt_mail_status → market_emails：
+ * 查待同步（update_time > status_updated_at）→ 按 mail_id 改业务表 → 回写同步时间。
  */
 @Slf4j
 @Service
 @DS(DSConstants.DATASOURCE_TRADE)
 public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSyncService {
 
-    public static final String SOURCE_MKT_MAIL_STATUS = "mkt_mail_status";
-
     private static final int BATCH_LIMIT = 200;
     private static final int MAX_PAGES = 50;
 
-    /** trade.status：0草稿/待发，1已发送，2发送失败 */
     private static final int TRADE_STATUS_DRAFT = 0;
     private static final int TRADE_STATUS_SENT = 1;
     private static final int TRADE_STATUS_FAIL = 2;
@@ -43,34 +39,29 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
     @Resource
     private MktMailRemoteClient mktMailRemoteClient;
 
-    @Resource
-    private MailStatusSyncCursorService mailStatusSyncCursorService;
-
     @Resource
     private MarketEmailsService marketEmailsService;
 
     @Override
     public int syncOnce() {
-        MailStatusSyncCursorEntity cursor = mailStatusSyncCursorService.getOrInit(SOURCE_MKT_MAIL_STATUS);
-        LocalDateTime since = cursor.getLastStatusUpdatedAt();
-        Long lastId = cursor.getLastId() == null ? 0L : cursor.getLastId();
-
         int totalUpdated = 0;
+        Long lastId = null;
+
         for (int page = 0; page < MAX_PAGES; page++) {
+            // 1. 查 mkt_mail_status：有最新修改（update_time > status_updated_at）
             MktMailStatusSyncQueryRemoteDTO query = new MktMailStatusSyncQueryRemoteDTO();
-            query.setSince(since);
-            query.setLastId(lastId);
             query.setLimit(BATCH_LIMIT);
+            query.setLastId(lastId);
 
             Result<List<MktMailStatusChangeRemoteDTO>> result;
             try {
                 result = mktMailRemoteClient.statusChanges(query);
             } catch (Exception e) {
-                log.error("拉取营销邮件状态失败 since={} lastId={}", since, lastId, e);
+                log.error("查询 mkt_mail_status 待同步数据失败 lastId={}", lastId, e);
                 break;
             }
             if (result == null || !result.isSuccess()) {
-                log.warn("拉取营销邮件状态返回失败 message={}", result == null ? null : result.getMessage());
+                log.warn("查询 mkt_mail_status 失败 message={}", result == null ? null : result.getMessage());
                 break;
             }
             List<MktMailStatusChangeRemoteDTO> changes = result.getResult();
@@ -78,52 +69,62 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
                 break;
             }
 
-            LocalDateTime batchMaxTime = since;
-            Long batchMaxId = lastId;
-            int pageUpdated = 0;
+            List<Long> syncedMailIds = new ArrayList<>();
+            Long batchLastId = lastId;
             for (MktMailStatusChangeRemoteDTO change : changes) {
                 if (change == null || change.getMailId() == null) {
                     continue;
                 }
-                if (applyChange(change)) {
-                    pageUpdated++;
+                if (change.getId() != null) {
+                    batchLastId = change.getId();
                 }
-                if (change.getStatusUpdatedAt() != null) {
-                    if (batchMaxTime == null || change.getStatusUpdatedAt().isAfter(batchMaxTime)
-                            || (change.getStatusUpdatedAt().equals(batchMaxTime)
-                            && change.getId() != null && change.getId() > batchMaxId)) {
-                        batchMaxTime = change.getStatusUpdatedAt();
-                        batchMaxId = change.getId() == null ? batchMaxId : change.getId();
-                    }
-                } else if (change.getId() != null && change.getId() > batchMaxId) {
-                    batchMaxId = change.getId();
+                // 2. 按邮件 id 修改 market_emails
+                ApplyResult applyResult = applyToMarketEmails(change);
+                if (applyResult == ApplyResult.UPDATED) {
+                    totalUpdated++;
+                    syncedMailIds.add(change.getMailId());
+                } else if (applyResult == ApplyResult.MATCHED_NO_CHANGE) {
+                    // 业务表已对齐，仍需回写同步时间，避免反复拉取
+                    syncedMailIds.add(change.getMailId());
                 }
+                // UNMATCHED：尚无 email_id 关联，不 ack，下次继续等
             }
 
-            // 本批处理完再推进游标（失败不推进则下次重拉，更新需幂等）
-            since = batchMaxTime;
-            lastId = batchMaxId;
-            mailStatusSyncCursorService.advance(SOURCE_MKT_MAIL_STATUS, since, lastId);
-            totalUpdated += pageUpdated;
+            // 3. 同步后回写 mkt_mail_status.status_updated_at
+            if (!syncedMailIds.isEmpty()) {
+                try {
+                    Result<?> ack = mktMailRemoteClient.ackStatusSync(syncedMailIds);
+                    if (ack == null || !ack.isSuccess()) {
+                        log.warn("回写 mkt_mail_status.status_updated_at 失败 mailIds={} message={}",
+                                syncedMailIds, ack == null ? null : ack.getMessage());
+                        // ack 失败则不再继续翻页，避免漏确认
+                        break;
+                    }
+                } catch (Exception e) {
+                    log.error("回写 mkt_mail_status.status_updated_at 异常 mailIds={}", syncedMailIds, e);
+                    break;
+                }
+            }
 
             if (changes.size() < BATCH_LIMIT) {
                 break;
             }
+            if (batchLastId == null || batchLastId.equals(lastId)) {
+                break;
+            }
+            lastId = batchLastId;
         }
         return totalUpdated;
     }
 
-    /**
-     * 按 email_id 回写；发送状态不降级；已读/已回复只升不降。
-     */
-    private boolean applyChange(MktMailStatusChangeRemoteDTO change) {
+    private ApplyResult applyToMarketEmails(MktMailStatusChangeRemoteDTO change) {
         MarketEmailsEntity existing = marketEmailsService.getOne(new LambdaQueryWrapper<MarketEmailsEntity>()
                 .eq(MarketEmailsEntity::getEmailId, change.getMailId())
                 .eq(MarketEmailsEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
                 .last("limit 1"));
         if (existing == null) {
-            log.debug("market_emails 无 email_id={} 记录，跳过", change.getMailId());
-            return false;
+            log.debug("market_emails 无 email_id={}，跳过且不 ack", change.getMailId());
+            return ApplyResult.UNMATCHED;
         }
 
         Integer mappedStatus = mapSendStatus(change.getSendStatus());
@@ -149,9 +150,9 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
             changed = true;
         }
         if (!changed) {
-            return false;
+            return ApplyResult.MATCHED_NO_CHANGE;
         }
-        return marketEmailsService.update(update);
+        return marketEmailsService.update(update) ? ApplyResult.UPDATED : ApplyResult.MATCHED_NO_CHANGE;
     }
 
     private static Integer mapSendStatus(Integer sendStatus) {
@@ -164,11 +165,9 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         if (sendStatus == 2) {
             return TRADE_STATUS_FAIL;
         }
-        // -2草稿 -1定时 0发送中 → trade 0
         return TRADE_STATUS_DRAFT;
     }
 
-    /** 已打开或已读都视为 has_reader=1 */
     private static Integer resolveReader(MktMailStatusChangeRemoteDTO change) {
         if (Integer.valueOf(1).equals(change.getOpened()) || Integer.valueOf(1).equals(change.getIsRead())) {
             return 1;
@@ -176,9 +175,6 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         return null;
     }
 
-    /**
-     * 发送状态：成功不降级；失败可升为成功；草稿可变为成功/失败。
-     */
     private static boolean shouldUpgradeSendStatus(Integer current, Integer mapped) {
         if (mapped == null) {
             return false;
@@ -195,7 +191,12 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         if (TRADE_STATUS_FAIL == current) {
             return TRADE_STATUS_SENT == mapped;
         }
-        // 当前为草稿/待发
         return true;
     }
+
+    private enum ApplyResult {
+        UPDATED,
+        MATCHED_NO_CHANGE,
+        UNMATCHED
+    }
 }

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

@@ -8,7 +8,7 @@ import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
 
 /**
- * 定时增量同步邮件库 mkt_mail_status → trade.market_emails。
+ * 定时：mkt_mail_status（有更新）→ market_emails，再回写 status_updated_at。
  */
 @Slf4j
 @Component
@@ -17,17 +17,17 @@ public class MarketEmailsStatusSyncTask {
     @Resource
     private MarketEmailsStatusSyncService marketEmailsStatusSyncService;
 
-    /** 每 2 分钟跑一轮 */
-    // @Scheduled(cron = "0 */2 * * * ?")
+    @Scheduled(cron = "0 */2 * * * ?")
     public void sync() {
         long start = System.currentTimeMillis();
         try {
             int updated = marketEmailsStatusSyncService.syncOnce();
             if (updated > 0) {
-                log.info("营销邮件状态同步完成 updated={} cost={}ms", updated, System.currentTimeMillis() - start);
+                log.info("mkt_mail_status → market_emails 同步完成 updated={} cost={}ms",
+                        updated, System.currentTimeMillis() - start);
             }
         } catch (Exception e) {
-            log.error("营销邮件状态同步失败", e);
+            log.error("mkt_mail_status → market_emails 同步失败", e);
         }
     }
 }

java/storlead-sasa/storlead-trade/src/main/resources/sql/mail_status_sync_cursor.sql

@@ -1,17 +0,0 @@
--- trade 库：邮件状态同步游标（消费 mkt_mail_status 增量）
-CREATE TABLE IF NOT EXISTS `mail_status_sync_cursor` (
-  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',
-  `source` varchar(64) NOT NULL COMMENT '同步源，如 mkt_mail_status',
-  `last_status_updated_at` datetime DEFAULT NULL COMMENT '已同步到的 status_updated_at',
-  `last_id` bigint DEFAULT NULL COMMENT '已同步到的 mkt_mail_status.id',
-  `owner_by` bigint DEFAULT NULL COMMENT '数据归属人',
-  `create_by` bigint DEFAULT NULL COMMENT '创建人',
-  `update_by` bigint DEFAULT NULL COMMENT '最后更新人',
-  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
-  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
-  `is_delete` tinyint NOT NULL DEFAULT '0' COMMENT '逻辑删除：0未删除；1已删除',
-  `enabled` tinyint NOT NULL DEFAULT '1' COMMENT '是否启用：1启用；0禁用',
-  `sort` int DEFAULT '0' COMMENT '排序号',
-  PRIMARY KEY (`id`),
-  UNIQUE KEY `uk_source` (`source`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='邮件状态同步游标';