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

@@ -29,7 +29,7 @@ public class MktMailStatusServiceImpl
         }
         LambdaQueryWrapper<MktMailStatusEntity> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(MktMailStatusEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
-                .apply("update_time > status_updated_at or status_updated_at is null ")
+                .apply("(status_updated_at > synced_at or synced_at is null)")
                 .gt(lastId != null, MktMailStatusEntity::getId, lastId)
                 .orderByAsc(MktMailStatusEntity::getId)
                 .last("limit " + limit);
@@ -44,7 +44,7 @@ public class MktMailStatusServiceImpl
         LambdaUpdateWrapper<MktMailStatusEntity> update = new LambdaUpdateWrapper<>();
         update.in(MktMailStatusEntity::getMailId, mailIds)
                 .eq(MktMailStatusEntity::getIsDelete, CommonConstant.DEL_FLAG_0)
-                .set(MktMailStatusEntity::getStatusUpdatedAt, new Date());
+                .set(MktMailStatusEntity::getSyncedAt, new Date());
         return update(update);
     }
 }

java/storlead-mail/storlead-mail-core/src/main/java/com/storlead/mail/pojo/entity/MktMailStatusEntity.java

@@ -50,11 +50,21 @@ public class MktMailStatusEntity extends SysBaseField {
     @TableField("opened")
     private Integer opened;
 
+    @ApiModelProperty("打开次数")
+    @TableField("open_count")
+    private Integer openCount;
+
     @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
     @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     @TableField("first_open_time")
     private Date firstOpenTime;
 
+    @ApiModelProperty("最近打开时间")
+    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @TableField("last_open_time")
+    private Date lastOpenTime;
+
     @ApiModelProperty("是否已读：0否 1是")
     @TableField("is_read")
     private Integer isRead;
@@ -78,4 +88,10 @@ public class MktMailStatusEntity extends SysBaseField {
     @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
     @TableField("status_updated_at")
     private Date statusUpdatedAt;
+
+    @ApiModelProperty("同步完成时间；待同步条件：status_updated_at > synced_at or synced_at is null")
+    @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @TableField("synced_at")
+    private Date syncedAt;
 }

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

@@ -11,7 +11,7 @@ import java.util.List;
 public interface MktMailStatusService extends MyBaseService<MktMailStatusEntity> {
 
     /**
-     * 待同步：update_time &gt; status_updated_at，按 id 升序分页。
+     * 待同步：status_updated_at &gt; synced_at or synced_at is null，按 id 升序分页。
      *
      * @param lastId 上一批最后一条 id，首页传 null
      * @param limit  批量大小
@@ -19,7 +19,7 @@ public interface MktMailStatusService extends MyBaseService<MktMailStatusEntity>
     List<MktMailStatusEntity> listPendingChanges(Long lastId, int limit);
 
     /**
-     * 同步成功后回写 status_updated_at = now()。
+     * 同步成功后回写 synced_at = now()。
      */
     boolean ackStatusSync(List<Long> mailIds);
 }

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

@@ -1,14 +1,14 @@
 package com.storlead.trade.service;
 
 /**
- * 业务同步：mkt_mail_status → market_emails，再回写 status_updated_at。
+ * 业务同步：mkt_mail_status → market_emails，再回写 synced_at。
  */
 public interface MarketEmailsStatusSyncService {
 
     /**
-     * 1. 查 mkt_mail_status（update_time &gt; status_updated_at）<br>
+     * 1. 查 mkt_mail_status（status_updated_at &gt; synced_at or synced_at is null）<br>
      * 2. 按 mail_id 更新 market_emails<br>
-     * 3. 回写 mkt_mail_status.status_updated_at
+     * 3. 回写 mkt_mail_status.synced_at
      *
      * @return 实际更新 market_emails 条数
      */

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

@@ -16,10 +16,11 @@ import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 
 /**
- * 跨库同步：邮件库 mkt_mail_status → 业务库 market_emails，再回写 status_updated_at。
+ * 跨库同步：邮件库 mkt_mail_status → 业务库 market_emails，再回写 synced_at。
  * <p>
  * 邮件侧走 {@link MktMailStatusService}（@DS mail），业务侧走 trade 数据源；不做跨库事务。
  */
@@ -47,7 +48,7 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         Long lastId = null;
 
         for (int page = 0; page < MAX_PAGES; page++) {
-            // 1. 邮件库：update_time > status_updated_at
+            // 1. 邮件库：status_updated_at > synced_at or synced_at is null
             List<MktMailStatusEntity> changes;
             try {
                 changes = mktMailStatusService.listPendingChanges(lastId, BATCH_LIMIT);
@@ -79,16 +80,16 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
                 // UNMATCHED：尚无 email_id 关联，不 ack，下次继续等
             }
 
-            // 3. 邮件库：回写 status_updated_at
+            // 3. 邮件库：回写 synced_at
             if (!syncedMailIds.isEmpty()) {
                 try {
                     boolean ackOk = mktMailStatusService.ackStatusSync(syncedMailIds);
                     if (!ackOk) {
-                        log.warn("回写 mkt_mail_status.status_updated_at 失败 mailIds={}", syncedMailIds);
+                        log.warn("回写 mkt_mail_status.synced_at 失败 mailIds={}", syncedMailIds);
                         break;
                     }
                 } catch (Exception e) {
-                    log.error("回写 mkt_mail_status.status_updated_at 异常 mailIds={}", syncedMailIds, e);
+                    log.error("回写 mkt_mail_status.synced_at 异常 mailIds={}", syncedMailIds, e);
                     break;
                 }
             }
@@ -136,9 +137,11 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
             update.set(MarketEmailsEntity::getHasReply, 1);
             changed = true;
         }
-        //补充首次打开时间（仅在业务表为空时写入，保持“首次”语义）
-        if (existing.getFirstOpenTime() == null && change.getFirstOpenTime() != null) {
-            update.set(MarketEmailsEntity::getFirstOpenTime, change.getFirstOpenTime());
+        // last_open_time 与 FirstOpenTime 不一致时，回写打开时间并同步 open_count
+        if (change.getLastOpenTime() != null
+                && !sameOpenTime(existing.getFirstOpenTime(), change.getLastOpenTime())) {
+            update.set(MarketEmailsEntity::getFirstOpenTime, change.getLastOpenTime());
+            update.set(MarketEmailsEntity::getOpenCount, change.getOpenCount());
             changed = true;
         }
         //补充首次回复时间
@@ -175,6 +178,16 @@ public class MarketEmailsStatusSyncServiceImpl implements MarketEmailsStatusSync
         return TRADE_STATUS_DRAFT;
     }
 
+    private static boolean sameOpenTime(Date existingFirstOpenTime, Date lastOpenTime) {
+        if (existingFirstOpenTime == lastOpenTime) {
+            return true;
+        }
+        if (existingFirstOpenTime == null || lastOpenTime == null) {
+            return false;
+        }
+        return existingFirstOpenTime.getTime() == lastOpenTime.getTime();
+    }
+
     private static Integer resolveReader(MktMailStatusEntity change) {
         if (Integer.valueOf(1).equals(change.getOpened()) || Integer.valueOf(1).equals(change.getIsRead())) {
             return 1;

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（有更新）→ market_emails，再回写 status_updated_at。
+ * 定时：mkt_mail_status（有更新）→ market_emails，再回写 synced_at。
  */
 @Slf4j
 @Component