map = new LinkedHashMap<>();
+ map.put("intent", intent.getIntent());
+ map.put("confidence", intent.getConfidence());
+ map.put("slots", intent.getSlots());
+ map.put("missing_slots", intent.getMissingSlots());
+ map.put("requires_confirmation", intent.isRequiresConfirmation());
+ return map;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Object parseJson(String json) {
+ if (json == null || json.isBlank()) {
+ return null;
+ }
+ Object result = JsonUtils.fromJson(json, Object.class);
+ return result != null ? result : json;
+ }
+
+ private String mapDirection(Integer direction) {
+ if (direction == null) {
+ return "in";
+ }
+ return switch (direction) {
+ case 1 -> "in";
+ case 2 -> "out";
+ default -> String.valueOf(direction);
+ };
+ }
+
+ private String formatTime(LocalDateTime time) {
+ if (time == null) {
+ return null;
+ }
+ return time.format(ISO_FORMATTER);
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignature.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignature.java
new file mode 100644
index 0000000..4c81353
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignature.java
@@ -0,0 +1,11 @@
+package com.sino.mci.inbound.webhook;
+
+/**
+ * 入站 webhook 签名结果。
+ *
+ * @param signature Base64 编码的 HMAC 签名
+ * @param algorithm 签名算法，如 hmac-sha256 / hmac-sha512
+ * @param timestamp 签名时的时间戳（Unix 秒）
+ */
+public record WebhookSignature(String signature, String algorithm, long timestamp) {
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignatureService.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignatureService.java
new file mode 100644
index 0000000..7e98786
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/webhook/WebhookSignatureService.java
@@ -0,0 +1,67 @@
+package com.sino.mci.inbound.webhook;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+
+/**
+ * 入站 webhook 签名服务。
+ *
+ * 使用 HMAC-SHA256 / HMAC-SHA512 对原始 payload 进行签名，
+ * 输出 Base64 编码签名、算法标识及时间戳，供业务系统验签使用。
+ */
+@Slf4j
+@Service
+public class WebhookSignatureService {
+
+ private static final String DEFAULT_ALGORITHM = "hmac-sha256";
+ private static final String HMAC_SHA256 = "HmacSHA256";
+ private static final String HMAC_SHA512 = "HmacSHA512";
+
+ /**
+ * 对原始 payload 进行签名。
+ *
+ * @param tenantCode 租户编码（用于日志/追踪，不参与签名计算）
+ * @param payload 待签名的原始 payload 字符串
+ * @param secret 租户 webhook 密钥
+ * @param algorithm 签名算法，支持 hmac-sha256（默认）和 hmac-sha512
+ * @return 签名结果，包含签名值、算法及时间戳
+ */
+ public WebhookSignature sign(String tenantCode, String payload, String secret, String algorithm) {
+ String normalizedAlgorithm = normalizeAlgorithm(algorithm);
+ String signature = hmacSign(tenantCode, payload, secret, normalizedAlgorithm);
+ long timestamp = System.currentTimeMillis() / 1000;
+ return new WebhookSignature(signature, normalizedAlgorithm, timestamp);
+ }
+
+ private String normalizeAlgorithm(String algorithm) {
+ if (algorithm == null || algorithm.isBlank()) {
+ return DEFAULT_ALGORITHM;
+ }
+ return switch (algorithm.toLowerCase().replace("_", "-")) {
+ case "hmac-sha256", "sha256" -> "hmac-sha256";
+ case "hmac-sha512", "sha512" -> "hmac-sha512";
+ default -> DEFAULT_ALGORITHM;
+ };
+ }
+
+ private String hmacSign(String tenantCode, String payload, String secret, String algorithm) {
+ String hmacAlgorithm = algorithm.equals("hmac-sha512") ? HMAC_SHA512 : HMAC_SHA256;
+ try {
+ Mac mac = Mac.getInstance(hmacAlgorithm);
+ SecretKeySpec keySpec = new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), hmacAlgorithm);
+ mac.init(keySpec);
+ byte[] signatureBytes = mac.doFinal(payload.getBytes(StandardCharsets.UTF_8));
+ return Base64.getEncoder().encodeToString(signatureBytes);
+ } catch (NoSuchAlgorithmException | InvalidKeyException e) {
+ log.error("Failed to sign webhook payload for tenant {}", tenantCode, e);
+ throw new IllegalStateException("Webhook signature failed: " + algorithm, e);
+ }
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/mq/config/MqConfig.java b/backend/mci-server/src/main/java/com/sino/mci/mq/config/MqConfig.java
new file mode 100644
index 0000000..ebff1cb
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/mq/config/MqConfig.java
@@ -0,0 +1,119 @@
+package com.sino.mci.mq.config;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 消息发送 RabbitMQ 配置。
+ *
+ * 重试机制：主队列设置死信交换机（DLX）和死信路由键；发送失败需要延迟重试时，
+ * 将消息发送到 DLX 并设置 per-message TTL，到期后由 DLX 重新路由回主队列消费。
+ */
+@Configuration
+public class MqConfig {
+
+ public static final String SEND_EXCHANGE = "mci.send.exchange";
+ public static final String SEND_QUEUE = "mci.send.queue";
+ public static final String SEND_ROUTING_KEY = "mci.send";
+
+ public static final String SEND_DLX_EXCHANGE = "mci.send.dlx.exchange";
+ public static final String SEND_DLX_ROUTING_KEY = "mci.send.retry";
+
+ public static final String CALLBACK_EXCHANGE = "mci.callback.exchange";
+ public static final String CALLBACK_QUEUE = "mci.callback.queue";
+ public static final String CALLBACK_ROUTING_KEY = "mci.callback";
+
+ public static final String INBOUND_EXCHANGE = "mci.inbound.exchange";
+ public static final String INBOUND_QUEUE = "mci.inbound.queue";
+ public static final String INBOUND_ROUTING_KEY = "mci.inbound";
+
+ public static final String INBOUND_FLOW_EXCHANGE = "mci.inbound.flow.exchange";
+ public static final String INBOUND_FLOW_QUEUE = "mci.inbound.flow.queue";
+ public static final String INBOUND_FLOW_ROUTING_KEY = "mci.inbound.flow";
+
+ @Bean
+ public Queue sendQueue() {
+ Map args = new HashMap<>();
+ args.put("x-dead-letter-exchange", SEND_DLX_EXCHANGE);
+ args.put("x-dead-letter-routing-key", SEND_DLX_ROUTING_KEY);
+ return new Queue(SEND_QUEUE, true, false, false, args);
+ }
+
+ @Bean
+ public DirectExchange sendExchange() {
+ return new DirectExchange(SEND_EXCHANGE, true, false);
+ }
+
+ @Bean
+ public Binding sendBinding(Queue sendQueue, DirectExchange sendExchange) {
+ return BindingBuilder.bind(sendQueue).to(sendExchange).with(SEND_ROUTING_KEY);
+ }
+
+ @Bean
+ public DirectExchange sendDlxExchange() {
+ return new DirectExchange(SEND_DLX_EXCHANGE, true, false);
+ }
+
+ @Bean
+ public Binding sendDlxBinding(Queue sendQueue, DirectExchange sendDlxExchange) {
+ return BindingBuilder.bind(sendQueue).to(sendDlxExchange).with(SEND_DLX_ROUTING_KEY);
+ }
+
+ @Bean
+ public Queue callbackQueue() {
+ return new Queue(CALLBACK_QUEUE, true);
+ }
+
+ @Bean
+ public DirectExchange callbackExchange() {
+ return new DirectExchange(CALLBACK_EXCHANGE, true, false);
+ }
+
+ @Bean
+ public Binding callbackBinding(Queue callbackQueue, DirectExchange callbackExchange) {
+ return BindingBuilder.bind(callbackQueue).to(callbackExchange).with(CALLBACK_ROUTING_KEY);
+ }
+
+ @Bean
+ public Queue inboundQueue() {
+ return new Queue(INBOUND_QUEUE, true);
+ }
+
+ @Bean
+ public DirectExchange inboundExchange() {
+ return new DirectExchange(INBOUND_EXCHANGE, true, false);
+ }
+
+ @Bean
+ public Binding inboundBinding(Queue inboundQueue, DirectExchange inboundExchange) {
+ return BindingBuilder.bind(inboundQueue).to(inboundExchange).with(INBOUND_ROUTING_KEY);
+ }
+
+ @Bean
+ public Queue inboundFlowQueue() {
+ return new Queue(INBOUND_FLOW_QUEUE, true);
+ }
+
+ @Bean
+ public DirectExchange inboundFlowExchange() {
+ return new DirectExchange(INBOUND_FLOW_EXCHANGE, true, false);
+ }
+
+ @Bean
+ public Binding inboundFlowBinding(Queue inboundFlowQueue, DirectExchange inboundFlowExchange) {
+ return BindingBuilder.bind(inboundFlowQueue).to(inboundFlowExchange).with(INBOUND_FLOW_ROUTING_KEY);
+ }
+
+ @Bean
+ public RestTemplate restTemplate() {
+ return new RestTemplate();
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/Policy.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/Policy.java
new file mode 100644
index 0000000..adad6e8
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/Policy.java
@@ -0,0 +1,44 @@
+package com.sino.mci.policy.engine;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class Policy {
+
+ private PolicyType type;
+
+ // rate_limit
+ @JsonProperty("scope")
+ private String scope;
+
+ @JsonProperty("limit")
+ private Integer limit;
+
+ @JsonProperty("window")
+ private String window;
+
+ // time_window
+ @JsonProperty("allow")
+ private List allow;
+
+ // blacklist
+ @JsonProperty("person_ids")
+ private List personIds;
+
+ @JsonProperty("conversation_ids")
+ private List conversationIds;
+
+ // retry
+ @JsonProperty("max_retry")
+ private Integer maxRetry;
+
+ @JsonProperty("interval")
+ private List intervals;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyCheckResult.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyCheckResult.java
new file mode 100644
index 0000000..4aa9d8e
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyCheckResult.java
@@ -0,0 +1,18 @@
+package com.sino.mci.policy.engine;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class PolicyCheckResult {
+
+ private boolean allowed;
+ private String reasonCode;
+ private String reasonMessage;
+ private RetryConfig retryConfig;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyEngine.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyEngine.java
new file mode 100644
index 0000000..2b5c2d0
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyEngine.java
@@ -0,0 +1,240 @@
+package com.sino.mci.policy.engine;
+
+import com.sino.mci.send.entity.SendRecord;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Optional;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class PolicyEngine {
+
+ private static final DateTimeFormatter HOURLY_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMddHH");
+ private static final DateTimeFormatter DAILY_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
+ private static final DateTimeFormatter WEEKLY_FORMATTER = DateTimeFormatter.ofPattern("YYYYww");
+
+ private final StringRedisTemplate redisTemplate;
+
+ public PolicyCheckResult check(SendRecord record, PolicyGroupConfig config) {
+ return check(record, config, LocalDateTime.now());
+ }
+
+ public PolicyCheckResult check(SendRecord record, PolicyGroupConfig config, LocalDateTime now) {
+ if (config == null || CollectionUtils.isEmpty(config.getPolicies())) {
+ return allowed(null);
+ }
+
+ PolicyCheckResult blacklistResult = checkBlacklist(record, config);
+ if (blacklistResult != null) {
+ return blacklistResult;
+ }
+
+ PolicyCheckResult timeWindowResult = checkTimeWindow(config, now);
+ if (timeWindowResult != null) {
+ return timeWindowResult;
+ }
+
+ PolicyCheckResult rateLimitResult = checkRateLimit(record, config, now);
+ if (rateLimitResult != null) {
+ return rateLimitResult;
+ }
+
+ return allowed(config.getRetryConfig().orElse(null));
+ }
+
+ public PolicyPreviewResult previewCheck(SendRecord record, PolicyGroupConfig config, LocalDateTime now) {
+ PolicyPreviewResult result = PolicyPreviewResult.builder()
+ .timeWindowOk(true)
+ .blacklistHit(0)
+ .build();
+ if (config == null || CollectionUtils.isEmpty(config.getPolicies())) {
+ return result;
+ }
+
+ if (isBlacklisted(record, config)) {
+ result.setBlacklistHit(1);
+ }
+
+ PolicyCheckResult timeWindowResult = checkTimeWindow(config, now);
+ result.setTimeWindowOk(timeWindowResult == null);
+
+ result.setRateLimitWarning(checkRateLimitDryRun(record, config, now));
+ return result;
+ }
+
+ private PolicyCheckResult allowed(RetryConfig retryConfig) {
+ return PolicyCheckResult.builder()
+ .allowed(true)
+ .retryConfig(retryConfig)
+ .build();
+ }
+
+ private PolicyCheckResult reject(String reasonCode, String reasonMessage) {
+ return PolicyCheckResult.builder()
+ .allowed(false)
+ .reasonCode(reasonCode)
+ .reasonMessage(reasonMessage)
+ .build();
+ }
+
+ private PolicyCheckResult checkBlacklist(SendRecord record, PolicyGroupConfig config) {
+ for (Policy policy : config.getPolicies()) {
+ if (policy.getType() != PolicyType.BLACKLIST) {
+ continue;
+ }
+ if (!CollectionUtils.isEmpty(policy.getPersonIds())
+ && record.getPersonId() != null
+ && policy.getPersonIds().contains(record.getPersonId())) {
+ return reject("blacklist_person", "发送目标人员在黑名单中");
+ }
+ if (!CollectionUtils.isEmpty(policy.getConversationIds())
+ && record.getConversationId() != null
+ && policy.getConversationIds().contains(record.getConversationId())) {
+ return reject("blacklist_conversation", "发送目标会话在黑名单中");
+ }
+ }
+ return null;
+ }
+
+ private boolean isBlacklisted(SendRecord record, PolicyGroupConfig config) {
+ return checkBlacklist(record, config) != null;
+ }
+
+ private PolicyCheckResult checkTimeWindow(PolicyGroupConfig config, LocalDateTime now) {
+ LocalTime current = now.toLocalTime();
+ for (Policy policy : config.getPolicies()) {
+ if (policy.getType() != PolicyType.TIME_WINDOW || CollectionUtils.isEmpty(policy.getAllow())) {
+ continue;
+ }
+ boolean insideAny = policy.getAllow().stream()
+ .map(this::parseTimeRange)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .anyMatch(range -> range.contains(current));
+ if (!insideAny) {
+ return reject("time_window_not_allowed", "当前时间不在允许发送窗口内");
+ }
+ }
+ return null;
+ }
+
+ private Optional parseTimeRange(String range) {
+ if (range == null || !range.contains("-")) {
+ return Optional.empty();
+ }
+ String[] parts = range.split("-", 2);
+ try {
+ LocalTime start = LocalTime.parse(parts[0].trim());
+ LocalTime end = LocalTime.parse(parts[1].trim());
+ return Optional.of(new TimeRange(start, end));
+ } catch (Exception e) {
+ log.warn("时间窗口格式错误: {}", range);
+ return Optional.empty();
+ }
+ }
+
+ private PolicyCheckResult checkRateLimit(SendRecord record, PolicyGroupConfig config, LocalDateTime now) {
+ for (Policy policy : config.getPolicies()) {
+ if (policy.getType() != PolicyType.RATE_LIMIT) {
+ continue;
+ }
+ Long scopeId = resolveScopeId(record, policy.getScope());
+ if (scopeId == null || policy.getLimit() == null || policy.getLimit() <= 0) {
+ continue;
+ }
+
+ String key = buildRateLimitKey(record.getTenantCode(), policy.getScope(), scopeId, policy.getWindow(), now);
+ long ttlSeconds = resolveTtlSeconds(policy.getWindow());
+
+ Long count = redisTemplate.opsForValue().increment(key);
+ if (count != null && count == 1L) {
+ redisTemplate.expire(key, Duration.ofSeconds(ttlSeconds));
+ }
+
+ if (count != null && count > policy.getLimit()) {
+ return reject("rate_limit_exceeded",
+ String.format("发送频率超过限制: scope=%s, limit=%d, window=%s", policy.getScope(), policy.getLimit(), policy.getWindow()));
+ }
+ }
+ return null;
+ }
+
+ private String checkRateLimitDryRun(SendRecord record, PolicyGroupConfig config, LocalDateTime now) {
+ for (Policy policy : config.getPolicies()) {
+ if (policy.getType() != PolicyType.RATE_LIMIT) {
+ continue;
+ }
+ Long scopeId = resolveScopeId(record, policy.getScope());
+ if (scopeId == null || policy.getLimit() == null || policy.getLimit() <= 0) {
+ continue;
+ }
+
+ String key = buildRateLimitKey(record.getTenantCode(), policy.getScope(), scopeId, policy.getWindow(), now);
+ String value = redisTemplate.opsForValue().get(key);
+ long count = 0L;
+ if (value != null) {
+ try {
+ count = Long.parseLong(value);
+ } catch (NumberFormatException e) {
+ log.warn("Redis 限流计数格式错误: key={}, value={}", key, value);
+ }
+ }
+
+ if (count >= policy.getLimit()) {
+ return String.format("scope=%s, id=%d 已超过频率限制: limit=%d, window=%s, current=%d",
+ policy.getScope(), scopeId, policy.getLimit(), policy.getWindow(), count);
+ }
+ if (count >= policy.getLimit() * 0.8) {
+ return String.format("scope=%s, id=%d 接近频率上限: limit=%d, window=%s, current=%d",
+ policy.getScope(), scopeId, policy.getLimit(), policy.getWindow(), count);
+ }
+ }
+ return null;
+ }
+
+ private Long resolveScopeId(SendRecord record, String scope) {
+ return switch (scope) {
+ case "account" -> record.getAccountId();
+ case "person" -> record.getPersonId();
+ case "conversation" -> record.getConversationId();
+ default -> null;
+ };
+ }
+
+ private String buildRateLimitKey(String tenantCode, String scope, Long scopeId, String window, LocalDateTime now) {
+ String period = switch (window) {
+ case "1h" -> now.format(HOURLY_FORMATTER);
+ case "7d" -> now.format(WEEKLY_FORMATTER);
+ default -> now.format(DAILY_FORMATTER);
+ };
+ return String.format("mci:rate:%s:%s:%s:%s", tenantCode, scope, scopeId, period);
+ }
+
+ private long resolveTtlSeconds(String window) {
+ return switch (window) {
+ case "1h" -> 3600L;
+ case "7d" -> 7 * 86400L;
+ default -> 86400L;
+ };
+ }
+
+ private record TimeRange(LocalTime start, LocalTime end) {
+
+ boolean contains(LocalTime time) {
+ if (!start.isAfter(end)) {
+ return !time.isBefore(start) && !time.isAfter(end);
+ }
+ return !time.isBefore(start) || !time.isAfter(end);
+ }
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyGroupConfig.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyGroupConfig.java
new file mode 100644
index 0000000..ae7eb12
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyGroupConfig.java
@@ -0,0 +1,47 @@
+package com.sino.mci.policy.engine;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+@Slf4j
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class PolicyGroupConfig {
+
+ private List policies = new ArrayList<>();
+
+ public static PolicyGroupConfig parseJson(String json) {
+ if (!StringUtils.hasText(json)) {
+ return new PolicyGroupConfig();
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ List list = mapper.readValue(json, new TypeReference>() {
+ });
+ return new PolicyGroupConfig(list != null ? list : new ArrayList<>());
+ } catch (JsonProcessingException e) {
+ log.warn("解析策略组配置失败: {}", json, e);
+ return new PolicyGroupConfig();
+ }
+ }
+
+ public Optional getRetryConfig() {
+ return policies.stream()
+ .filter(p -> p.getType() == PolicyType.RETRY)
+ .findFirst()
+ .map(p -> new RetryConfig(
+ p.getMaxRetry() == null ? 0 : p.getMaxRetry(),
+ p.getIntervals()));
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyPreviewResult.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyPreviewResult.java
new file mode 100644
index 0000000..3c212aa
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyPreviewResult.java
@@ -0,0 +1,17 @@
+package com.sino.mci.policy.engine;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class PolicyPreviewResult {
+
+ private String rateLimitWarning;
+ private boolean timeWindowOk;
+ private int blacklistHit;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyType.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyType.java
new file mode 100644
index 0000000..f6bab1f
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/PolicyType.java
@@ -0,0 +1,18 @@
+package com.sino.mci.policy.engine;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public enum PolicyType {
+
+ @JsonProperty("rate_limit")
+ RATE_LIMIT,
+
+ @JsonProperty("time_window")
+ TIME_WINDOW,
+
+ @JsonProperty("blacklist")
+ BLACKLIST,
+
+ @JsonProperty("retry")
+ RETRY
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/policy/engine/RetryConfig.java b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/RetryConfig.java
new file mode 100644
index 0000000..3274817
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/policy/engine/RetryConfig.java
@@ -0,0 +1,16 @@
+package com.sino.mci.policy.engine;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class RetryConfig {
+
+ private int maxRetry;
+ private List intervals;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/controller/SendController.java b/backend/mci-server/src/main/java/com/sino/mci/send/controller/SendController.java
index c44ad0c..0cd68da 100644
--- a/backend/mci-server/src/main/java/com/sino/mci/send/controller/SendController.java
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/controller/SendController.java
@@ -1,16 +1,33 @@
package com.sino.mci.send.controller;
+import com.baomidou.mybatisplus.core.metadata.IPage;
import com.sino.mci.admin.security.AuthContext;
+import com.sino.mci.send.dto.SendPreviewRequest;
+import com.sino.mci.send.dto.SendPreviewResponse;
import com.sino.mci.send.dto.SendRecordResponse;
import com.sino.mci.send.dto.SendRequest;
import com.sino.mci.send.dto.SendResponse;
+import com.sino.mci.send.dto.SendStatisticsResponse;
+import com.sino.mci.send.dto.SendTaskStatusResponse;
import com.sino.mci.send.dto.SendTestRequest;
+import com.sino.mci.send.service.SendPreviewService;
import com.sino.mci.send.service.SendService;
import com.sino.mci.shared.web.ApiResponse;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.util.StringUtils;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.List;
@RestController
@@ -19,6 +36,9 @@ import java.util.List;
public class SendController {
private final SendService sendService;
+ private final SendPreviewService sendPreviewService;
+
+ private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
@PostMapping
public ApiResponse send(
@@ -34,15 +54,69 @@ public class SendController {
return ApiResponse.ok(sendService.testSend(tenantCode, request));
}
- @GetMapping("/{task_id}/records")
+ @PostMapping("/preview")
+ public ApiResponse preview(
+ @RequestHeader("X-Tenant-Code") String tenantCode,
+ @Valid @RequestBody SendPreviewRequest request) {
+ return ApiResponse.ok(sendPreviewService.preview(tenantCode, request));
+ }
+
+ @RequestMapping(value = "/{task_id}/records", method = {RequestMethod.GET, RequestMethod.POST})
public ApiResponse> listRecords(
@RequestHeader("X-Tenant-Code") String tenantCode,
@PathVariable("task_id") String taskId) {
return ApiResponse.ok(sendService.listRecords(tenantCode, taskId));
}
+ @GetMapping("/{task_id}/status")
+ public ApiResponse getTaskStatus(
+ @RequestHeader("X-Tenant-Code") String tenantCode,
+ @PathVariable("task_id") String taskId) {
+ return ApiResponse.ok(sendService.getTaskStatus(tenantCode, taskId));
+ }
+
+ @PostMapping("/{task_id}/cancel")
+ public ApiResponse cancelTask(
+ @RequestHeader("X-Tenant-Code") String tenantCode,
+ @PathVariable("task_id") String taskId) {
+ return ApiResponse.ok(sendService.cancelTask(tenantCode, taskId));
+ }
+
+ @GetMapping("/records")
+ public ApiResponse> queryRecords(
+ @RequestHeader("X-Tenant-Code") String tenantCode,
+ @RequestParam(required = false) String taskId,
+ @RequestParam(required = false) String start,
+ @RequestParam(required = false) String end,
+ @RequestParam(required = false) String channelType,
+ @RequestParam(required = false) Integer status,
+ @RequestParam(defaultValue = "1") long page,
+ @RequestParam(defaultValue = "20") long size) {
+ LocalDateTime startTime = parseDateTime(start);
+ LocalDateTime endTime = parseDateTime(end);
+ return ApiResponse.ok(sendService.queryRecords(tenantCode, taskId, startTime, endTime,
+ channelType, status, page, size));
+ }
+
+ @GetMapping("/statistics")
+ public ApiResponse statistics(
+ @RequestHeader("X-Tenant-Code") String tenantCode,
+ @RequestParam(required = false) String start,
+ @RequestParam(required = false) String end) {
+ LocalDateTime startTime = parseDateTime(start);
+ LocalDateTime endTime = parseDateTime(end);
+ return ApiResponse.ok(sendService.getStatistics(tenantCode, startTime, endTime));
+ }
+
@GetMapping("/list")
public ApiResponse> listRecentRecords() {
return ApiResponse.ok(sendService.listRecentRecords(AuthContext.currentTenantCode()));
}
+
+ private LocalDateTime parseDateTime(String value) {
+ if (!StringUtils.hasText(value)) {
+ return null;
+ }
+ return LocalDateTime.parse(value, DATE_TIME_FORMATTER);
+ }
}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/ChannelStatistics.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/ChannelStatistics.java
new file mode 100644
index 0000000..21f0f43
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/ChannelStatistics.java
@@ -0,0 +1,14 @@
+package com.sino.mci.send.dto;
+
+import lombok.Data;
+
+@Data
+public class ChannelStatistics {
+
+ private String channelType;
+ private long total;
+ private long pendingCount;
+ private long sendingCount;
+ private long successCount;
+ private long failedCount;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewRequest.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewRequest.java
new file mode 100644
index 0000000..32bb903
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewRequest.java
@@ -0,0 +1,19 @@
+package com.sino.mci.send.dto;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+public class SendPreviewRequest {
+
+ @NotBlank
+ private String targetType;
+
+ @NotNull
+ private Map targetValue;
+
+ private Map channelStrategy;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewResponse.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewResponse.java
new file mode 100644
index 0000000..5e4eb40
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendPreviewResponse.java
@@ -0,0 +1,25 @@
+package com.sino.mci.send.dto;
+
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+public class SendPreviewResponse {
+
+ private int totalPersonCount;
+ private int totalConversationCount;
+ private Map channelBreakdown;
+ private int unreachableCount;
+ private Map accountUsage;
+ private PolicyCheckInfo policyCheck;
+ private int estimatedTimeSeconds;
+
+ @Data
+ public static class PolicyCheckInfo {
+
+ private String rateLimitWarning;
+ private boolean timeWindowOk;
+ private int blacklistHit;
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendRecordResponse.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendRecordResponse.java
index 4ea24d6..efb0cb6 100644
--- a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendRecordResponse.java
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendRecordResponse.java
@@ -18,6 +18,9 @@ public class SendRecordResponse {
private String contentType;
private Integer sendStatus;
private String channelResult;
+ private String contentBody;
+ private String callbackUrl;
private LocalDateTime sendTime;
+ private LocalDateTime finishTime;
private LocalDateTime createTime;
}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendStatisticsResponse.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendStatisticsResponse.java
new file mode 100644
index 0000000..ac7b3e6
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendStatisticsResponse.java
@@ -0,0 +1,17 @@
+package com.sino.mci.send.dto;
+
+import lombok.Data;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+public class SendStatisticsResponse {
+
+ private long total;
+ private long pendingCount;
+ private long sendingCount;
+ private long successCount;
+ private long failedCount;
+ private List channelStats = new ArrayList<>();
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendTaskStatusResponse.java b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendTaskStatusResponse.java
new file mode 100644
index 0000000..8ae5a4b
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/dto/SendTaskStatusResponse.java
@@ -0,0 +1,14 @@
+package com.sino.mci.send.dto;
+
+import lombok.Data;
+
+@Data
+public class SendTaskStatusResponse {
+
+ private String taskId;
+ private long total;
+ private long pendingCount;
+ private long sendingCount;
+ private long successCount;
+ private long failedCount;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/entity/SendRecord.java b/backend/mci-server/src/main/java/com/sino/mci/send/entity/SendRecord.java
index e330deb..9fc282e 100644
--- a/backend/mci-server/src/main/java/com/sino/mci/send/entity/SendRecord.java
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/entity/SendRecord.java
@@ -26,6 +26,9 @@ public class SendRecord extends BaseEntity {
private String contentBody;
private String callbackUrl;
private Integer sendStatus;
+ private Integer retryCount;
+ private Integer maxRetry;
+ private LocalDateTime nextRetryTime;
private String channelResult;
private LocalDateTime sendTime;
private LocalDateTime finishTime;
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageConsumer.java b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageConsumer.java
new file mode 100644
index 0000000..43938dd
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageConsumer.java
@@ -0,0 +1,116 @@
+package com.sino.mci.send.mq;
+
+import com.sino.mci.mq.config.MqConfig;
+import com.sino.mci.policy.engine.RetryConfig;
+import com.sino.mci.send.entity.SendRecord;
+import com.sino.mci.send.repository.SendRecordMapper;
+import com.sino.mci.send.service.SendCallbackService;
+import com.sino.mci.send.service.SendService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class SendMessageConsumer {
+
+ private static final int SEND_STATUS_SUCCESS = 2;
+ private static final int SEND_STATUS_FAILED = 3;
+
+ private final SendRecordMapper sendRecordMapper;
+ private final SendService sendService;
+ private final SendMessageProducer sendMessageProducer;
+ private final SendCallbackService sendCallbackService;
+
+ @RabbitListener(queues = MqConfig.SEND_QUEUE)
+ public void handle(SendRecordMessage message) {
+ log.info("收到异步发送消息: recordId={}, attempt={}", message.getRecordId(), message.getAttempt());
+ SendRecord record = sendRecordMapper.selectById(message.getRecordId());
+ if (record == null) {
+ log.warn("发送记录不存在: recordId={}", message.getRecordId());
+ return;
+ }
+
+ if (record.getSendStatus() != null && record.getSendStatus() == SEND_STATUS_SUCCESS) {
+ sendCallbackService.sendCallback(record);
+ return;
+ }
+
+ if (record.getSendStatus() != null && record.getSendStatus() == SEND_STATUS_FAILED) {
+ sendCallbackService.sendCallback(record);
+ return;
+ }
+
+ int attempt = message.getAttempt();
+ int maxRetry = record.getMaxRetry() != null ? record.getMaxRetry() : 0;
+ int currentRetry = record.getRetryCount() != null ? record.getRetryCount() : 0;
+
+ if (attempt < currentRetry) {
+ log.warn("丢弃过期重试消息: recordId={}, attempt={}, currentRetry={}",
+ record.getId(), attempt, currentRetry);
+ return;
+ }
+
+ if (attempt > maxRetry) {
+ record.setSendStatus(SEND_STATUS_FAILED);
+ record.setFinishTime(LocalDateTime.now());
+ sendRecordMapper.updateById(record);
+ sendCallbackService.sendCallback(record);
+ return;
+ }
+
+ try {
+ sendService.dispatchRecord(record);
+
+ record = sendRecordMapper.selectById(record.getId());
+ if (record == null) {
+ return;
+ }
+
+ if (record.getSendStatus() != null && record.getSendStatus() == SEND_STATUS_SUCCESS) {
+ sendCallbackService.sendCallback(record);
+ } else if (record.getSendStatus() != null && record.getSendStatus() == SEND_STATUS_FAILED) {
+ handleRetry(record, message);
+ }
+ } catch (Exception e) {
+ log.error("异步发送处理异常: recordId={}", message.getRecordId(), e);
+ }
+ }
+
+ private void handleRetry(SendRecord record, SendRecordMessage message) {
+ int currentRetry = record.getRetryCount() != null ? record.getRetryCount() : 0;
+ int maxRetry = record.getMaxRetry() != null ? record.getMaxRetry() : 0;
+ if (currentRetry >= maxRetry) {
+ sendCallbackService.sendCallback(record);
+ return;
+ }
+
+ int attempt = message.getAttempt() + 1;
+ int delaySeconds = resolveDelaySeconds(message.getRetryPlan(), attempt);
+ SendRecordMessage retryMessage = SendRecordMessage.builder()
+ .recordId(message.getRecordId())
+ .taskId(message.getTaskId())
+ .tenantCode(message.getTenantCode())
+ .attempt(attempt)
+ .retryPlan(message.getRetryPlan())
+ .build();
+
+ record.setRetryCount(attempt);
+ record.setNextRetryTime(LocalDateTime.now().plusSeconds(delaySeconds));
+ sendRecordMapper.updateById(record);
+
+ sendMessageProducer.publishRetryMessage(retryMessage, delaySeconds);
+ }
+
+ private int resolveDelaySeconds(RetryConfig retryPlan, int attempt) {
+ if (retryPlan != null && retryPlan.getIntervals() != null && !retryPlan.getIntervals().isEmpty()) {
+ int index = Math.min(attempt - 1, retryPlan.getIntervals().size() - 1);
+ return retryPlan.getIntervals().get(index);
+ }
+ return 5;
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageProducer.java b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageProducer.java
new file mode 100644
index 0000000..344baff
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendMessageProducer.java
@@ -0,0 +1,29 @@
+package com.sino.mci.send.mq;
+
+import com.sino.mci.mq.config.MqConfig;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class SendMessageProducer {
+
+ private final RabbitTemplate rabbitTemplate;
+
+ public void publishSendMessage(SendRecordMessage message) {
+ log.info("发布异步发送消息: recordId={}, taskId={}", message.getRecordId(), message.getTaskId());
+ rabbitTemplate.convertAndSend(MqConfig.SEND_EXCHANGE, MqConfig.SEND_ROUTING_KEY, message);
+ }
+
+ public void publishRetryMessage(SendRecordMessage message, int delaySeconds) {
+ long delayMillis = delaySeconds * 1000L;
+ log.info("发布重试消息: recordId={}, attempt={}, delay={}ms", message.getRecordId(), message.getAttempt(), delayMillis);
+ rabbitTemplate.convertAndSend(MqConfig.SEND_DLX_EXCHANGE, MqConfig.SEND_DLX_ROUTING_KEY, message, msg -> {
+ msg.getMessageProperties().setExpiration(String.valueOf(delayMillis));
+ return msg;
+ });
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendRecordMessage.java b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendRecordMessage.java
new file mode 100644
index 0000000..eaf8219
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/mq/SendRecordMessage.java
@@ -0,0 +1,22 @@
+package com.sino.mci.send.mq;
+
+import com.sino.mci.policy.engine.RetryConfig;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SendRecordMessage implements Serializable {
+
+ private Long recordId;
+ private String taskId;
+ private String tenantCode;
+ private int attempt;
+ private RetryConfig retryPlan;
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/service/SendCallbackService.java b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendCallbackService.java
new file mode 100644
index 0000000..eefdbcc
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendCallbackService.java
@@ -0,0 +1,70 @@
+package com.sino.mci.send.service;
+
+import com.sino.mci.send.entity.SendRecord;
+import com.sino.mci.shared.util.JsonUtils;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+import org.springframework.web.client.RestTemplate;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class SendCallbackService {
+
+ private static final int CALLBACK_STATUS_DELIVERED = 1;
+ private static final int CALLBACK_STATUS_FAILED = 3;
+
+ private final RestTemplate restTemplate;
+
+ public void sendCallback(SendRecord record) {
+ if (record == null || !StringUtils.hasText(record.getCallbackUrl())) {
+ return;
+ }
+ String event;
+ int status;
+ if (record.getSendStatus() != null && record.getSendStatus() == 2) {
+ event = "delivered";
+ status = CALLBACK_STATUS_DELIVERED;
+ } else {
+ event = "failed";
+ status = CALLBACK_STATUS_FAILED;
+ }
+ Map payload = new HashMap<>();
+ payload.put("task_id", record.getTaskId());
+ payload.put("record_id", record.getId());
+ payload.put("conversation_id", record.getConversationId());
+ payload.put("person_id", record.getPersonId());
+ payload.put("channel_type", record.getChannelType());
+ payload.put("event", event);
+ payload.put("status", status);
+ payload.put("channel_result", parseChannelResult(record.getChannelResult()));
+ payload.put("event_time", record.getFinishTime() != null ? record.getFinishTime() : LocalDateTime.now());
+
+ HttpHeaders headers = new HttpHeaders();
+ headers.setContentType(MediaType.APPLICATION_JSON);
+ HttpEntity> entity = new HttpEntity<>(payload, headers);
+ try {
+ restTemplate.postForEntity(record.getCallbackUrl(), entity, String.class);
+ log.info("发送回调成功: recordId={}, event={}", record.getId(), event);
+ } catch (Exception e) {
+ log.warn("发送回调失败: recordId={}, url={}", record.getId(), record.getCallbackUrl(), e);
+ }
+ }
+
+ private Object parseChannelResult(String channelResult) {
+ if (!StringUtils.hasText(channelResult)) {
+ return new HashMap<>();
+ }
+ Object result = JsonUtils.fromJson(channelResult, Object.class);
+ return result != null ? result : new HashMap<>();
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/service/SendPreviewService.java b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendPreviewService.java
new file mode 100644
index 0000000..779adcd
--- /dev/null
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendPreviewService.java
@@ -0,0 +1,200 @@
+package com.sino.mci.send.service;
+
+import com.sino.mci.channel.entity.ChannelAccount;
+import com.sino.mci.channel.repository.ChannelAccountMapper;
+import com.sino.mci.policy.engine.PolicyEngine;
+import com.sino.mci.policy.engine.PolicyGroupConfig;
+import com.sino.mci.policy.engine.PolicyPreviewResult;
+import com.sino.mci.policy.entity.PolicyGroup;
+import com.sino.mci.policy.repository.PolicyGroupMapper;
+import com.sino.mci.send.dto.SendPreviewRequest;
+import com.sino.mci.send.dto.SendPreviewResponse;
+import com.sino.mci.send.entity.SendRecord;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+@Service
+@RequiredArgsConstructor
+public class SendPreviewService {
+
+ private static final int STATUS_ENABLED = 1;
+
+ private final SendService sendService;
+ private final ChannelAccountMapper accountMapper;
+ private final PolicyEngine policyEngine;
+ private final PolicyGroupMapper policyGroupMapper;
+
+ public SendPreviewResponse preview(String tenantCode, SendPreviewRequest request) {
+ List targets = sendService.resolveByType(tenantCode, request.getTargetType(),
+ request.getTargetValue(), request.getChannelStrategy());
+
+ int totalPersonCount = 0;
+ int totalConversationCount = 0;
+ int unreachableCount = 0;
+ Map channelBreakdown = new HashMap<>();
+ Map accountIdUsage = new HashMap<>();
+
+ for (SendTarget target : targets) {
+ if (target.getPersonId() != null) {
+ totalPersonCount++;
+ }
+ if (target.getConversationId() != null) {
+ totalConversationCount++;
+ }
+ if (target.getAccountId() == null) {
+ unreachableCount++;
+ } else {
+ accountIdUsage.merge(target.getAccountId(), 1, Integer::sum);
+ }
+ String channelType = target.getChannelType() != null ? target.getChannelType().name().toLowerCase() : "unknown";
+ channelBreakdown.merge(channelType, 1, Integer::sum);
+ }
+
+ Map accountUsage = buildAccountUsage(accountIdUsage);
+ SendPreviewResponse.PolicyCheckInfo policyCheck = dryRunPolicyCheck(tenantCode, request, targets, accountIdUsage.keySet());
+
+ int estimatedTimeSeconds = (int) (1 + targets.size() * 0.5 + totalConversationCount * 2);
+
+ SendPreviewResponse response = new SendPreviewResponse();
+ response.setTotalPersonCount(totalPersonCount);
+ response.setTotalConversationCount(totalConversationCount);
+ response.setChannelBreakdown(channelBreakdown);
+ response.setUnreachableCount(unreachableCount);
+ response.setAccountUsage(accountUsage);
+ response.setPolicyCheck(policyCheck);
+ response.setEstimatedTimeSeconds(estimatedTimeSeconds);
+ return response;
+ }
+
+ private Map buildAccountUsage(Map accountIdUsage) {
+ Map accountUsage = new HashMap<>();
+ if (accountIdUsage.isEmpty()) {
+ return accountUsage;
+ }
+ Set accountIds = accountIdUsage.keySet();
+ List accounts = accountMapper.selectBatchIds(accountIds);
+ Map accountNames = new HashMap<>();
+ for (ChannelAccount account : accounts) {
+ accountNames.put(account.getId(),
+ StringUtils.hasText(account.getAccountName()) ? account.getAccountName() : String.valueOf(account.getId()));
+ }
+ for (Map.Entry entry : accountIdUsage.entrySet()) {
+ String name = accountNames.getOrDefault(entry.getKey(), String.valueOf(entry.getKey()));
+ accountUsage.put(name, entry.getValue());
+ }
+ return accountUsage;
+ }
+
+ private SendPreviewResponse.PolicyCheckInfo dryRunPolicyCheck(String tenantCode, SendPreviewRequest request,
+ List targets, Set accountIds) {
+ SendPreviewResponse.PolicyCheckInfo info = new SendPreviewResponse.PolicyCheckInfo();
+ info.setTimeWindowOk(true);
+ info.setBlacklistHit(0);
+ info.setRateLimitWarning(null);
+
+ PolicyGroup policyGroup = loadPolicyGroup(tenantCode, extractChannelType(request.getChannelStrategy()));
+ if (policyGroup == null) {
+ return info;
+ }
+
+ PolicyGroupConfig config = PolicyGroupConfig.parseJson(policyGroup.getPolicies());
+ if (CollectionUtils.isEmpty(config.getPolicies())) {
+ return info;
+ }
+
+ LocalDateTime now = LocalDateTime.now();
+ int blacklistHit = 0;
+ boolean timeWindowOk = true;
+ StringBuilder warning = new StringBuilder();
+
+ Set checkedAccounts = new HashSet<>(accountIds);
+ if (checkedAccounts.isEmpty()) {
+ checkedAccounts.add(null);
+ }
+ for (Long accountId : checkedAccounts) {
+ SendTarget representative = targets.stream()
+ .filter(t -> Objects.equals(t.getAccountId(), accountId))
+ .findFirst()
+ .orElse(null);
+ SendRecord dummy = new SendRecord();
+ dummy.setTenantCode(tenantCode);
+ dummy.setAccountId(accountId);
+ dummy.setPersonId(representative != null ? representative.getPersonId() : null);
+ dummy.setConversationId(representative != null ? representative.getConversationId() : null);
+ dummy.setChannelType(extractChannelType(request.getChannelStrategy()));
+
+ PolicyPreviewResult result = policyEngine.previewCheck(dummy, config, now);
+ if (!result.isTimeWindowOk()) {
+ timeWindowOk = false;
+ }
+ blacklistHit += result.getBlacklistHit();
+ if (result.getRateLimitWarning() != null) {
+ if (warning.length() > 0) {
+ warning.append("; ");
+ }
+ warning.append(result.getRateLimitWarning());
+ }
+ }
+
+ info.setTimeWindowOk(timeWindowOk);
+ info.setBlacklistHit(blacklistHit);
+ info.setRateLimitWarning(warning.length() > 0 ? warning.toString() : null);
+ return info;
+ }
+
+ private PolicyGroup loadPolicyGroup(String tenantCode, String channelType) {
+ com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper wrapper =
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<>();
+ wrapper.eq(PolicyGroup::getTenantCode, tenantCode)
+ .eq(PolicyGroup::getStatus, STATUS_ENABLED);
+ if (StringUtils.hasText(channelType)) {
+ wrapper.eq(PolicyGroup::getChannelType, channelType);
+ } else {
+ wrapper.and(w -> w.isNull(PolicyGroup::getChannelType).or().eq(PolicyGroup::getChannelType, ""));
+ }
+ wrapper.orderByDesc(PolicyGroup::getId).last("LIMIT 1");
+ PolicyGroup group = policyGroupMapper.selectOne(wrapper);
+ if (group != null) {
+ return group;
+ }
+
+ group = policyGroupMapper.selectOne(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(PolicyGroup::getTenantCode, tenantCode)
+ .eq(PolicyGroup::getStatus, STATUS_ENABLED)
+ .and(w -> w.isNull(PolicyGroup::getChannelType).or().eq(PolicyGroup::getChannelType, ""))
+ .orderByDesc(PolicyGroup::getId)
+ .last("LIMIT 1"));
+ if (group != null) {
+ return group;
+ }
+
+ return policyGroupMapper.selectOne(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(PolicyGroup::getGroupCode, "default")
+ .eq(PolicyGroup::getStatus, STATUS_ENABLED)
+ .orderByDesc(PolicyGroup::getId)
+ .last("LIMIT 1"));
+ }
+
+ private String extractChannelType(Map channelStrategy) {
+ if (channelStrategy == null) {
+ return "wecom";
+ }
+ Object value = channelStrategy.get("channel_type");
+ if (value == null) {
+ value = channelStrategy.get("channelType");
+ }
+ return value != null ? String.valueOf(value) : "wecom";
+ }
+}
diff --git a/backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java
index 4b9b796..4d345b0 100644
--- a/backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java
+++ b/backend/mci-server/src/main/java/com/sino/mci/send/service/SendService.java
@@ -1,17 +1,18 @@
package com.sino.mci.send.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.sino.mci.channel.adapter.registry.ChannelAdapterRegistry;
import com.sino.mci.channel.common.dto.SendMessageRequest;
import com.sino.mci.channel.common.dto.SendMessageResult;
import com.sino.mci.channel.common.model.ChannelType;
import com.sino.mci.channel.common.spi.ChannelAdapter;
import com.sino.mci.channel.entity.ChannelAccount;
-import com.sino.mci.channel.entity.ChannelAccountConversation;
import com.sino.mci.channel.entity.Conversation;
-import com.sino.mci.channel.repository.ChannelAccountConversationMapper;
import com.sino.mci.channel.repository.ChannelAccountMapper;
import com.sino.mci.channel.repository.ConversationMapper;
+import com.sino.mci.channel.service.AccountSelector;
import com.sino.mci.channel.entity.ConversationTagBinding;
import com.sino.mci.crm.entity.ChannelIdentity;
import com.sino.mci.crm.entity.Person;
@@ -23,11 +24,22 @@ import com.sino.mci.crm.repository.PersonMapper;
import com.sino.mci.crm.repository.PersonTagBindingMapper;
import com.sino.mci.crm.repository.TagMapper;
import com.sino.mci.exception.BusinessException;
+import com.sino.mci.policy.engine.PolicyCheckResult;
+import com.sino.mci.policy.engine.PolicyEngine;
+import com.sino.mci.policy.engine.PolicyGroupConfig;
+import com.sino.mci.policy.entity.PolicyGroup;
+import com.sino.mci.policy.repository.PolicyGroupMapper;
+import com.sino.mci.policy.engine.RetryConfig;
+import com.sino.mci.send.dto.ChannelStatistics;
import com.sino.mci.send.dto.SendRecordResponse;
import com.sino.mci.send.dto.SendRequest;
+import com.sino.mci.send.dto.SendStatisticsResponse;
+import com.sino.mci.send.dto.SendTaskStatusResponse;
import com.sino.mci.send.dto.SendResponse;
import com.sino.mci.send.dto.SendTestRequest;
import com.sino.mci.send.entity.SendRecord;
+import com.sino.mci.send.mq.SendMessageProducer;
+import com.sino.mci.send.mq.SendRecordMessage;
import com.sino.mci.send.repository.SendRecordMapper;
import com.sino.mci.shared.util.JsonUtils;
import lombok.RequiredArgsConstructor;
@@ -41,8 +53,10 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -57,8 +71,6 @@ import java.util.stream.Collectors;
@RequiredArgsConstructor
public class SendService {
- private static final int BINDING_SEND_PRIMARY = 1;
- private static final int BINDING_SEND_BACKUP = 2;
private static final int STATUS_ENABLED = 1;
private static final int SEND_STATUS_PENDING = 0;
private static final int SEND_STATUS_SENDING = 1;
@@ -68,33 +80,36 @@ public class SendService {
private final SendRecordMapper sendRecordMapper;
private final ConversationMapper conversationMapper;
private final ChannelAccountMapper accountMapper;
- private final ChannelAccountConversationMapper accountConversationMapper;
private final ChannelIdentityMapper channelIdentityMapper;
private final PersonMapper personMapper;
private final TagMapper tagMapper;
private final PersonTagBindingMapper personTagBindingMapper;
private final ConversationTagBindingMapper conversationTagBindingMapper;
private final ChannelAdapterRegistry adapterRegistry;
+ private final PolicyGroupMapper policyGroupMapper;
+ private final PolicyEngine policyEngine;
+ private final AccountSelector accountSelector;
+ private final SendMessageProducer sendMessageProducer;
@Transactional
public SendResponse send(String tenantCode, SendRequest request) {
String taskId = generateTaskId();
List targets = resolveTargets(tenantCode, request);
- List records = createRecords(taskId, tenantCode, request, targets);
+ RetryConfig retryConfig = loadRetryConfig(tenantCode, request);
+ List records = createRecords(taskId, tenantCode, request, targets, retryConfig);
- int successCount = 0;
- int failedCount = 0;
for (SendRecord record : records) {
- dispatchRecord(record);
- if (record.getSendStatus() == SEND_STATUS_SUCCESS) {
- successCount++;
- } else if (record.getSendStatus() == SEND_STATUS_FAILED) {
- failedCount++;
- }
+ SendRecordMessage message = SendRecordMessage.builder()
+ .recordId(record.getId())
+ .taskId(record.getTaskId())
+ .tenantCode(record.getTenantCode())
+ .attempt(0)
+ .retryPlan(retryConfig)
+ .build();
+ sendMessageProducer.publishSendMessage(message);
}
- String status = failedCount == 0 ? "success" : (successCount == 0 ? "failed" : "partial");
- return new SendResponse(taskId, records.size(), status);
+ return new SendResponse(taskId, records.size(), "accepted");
}
@Transactional
@@ -122,21 +137,148 @@ public class SendService {
return records.stream().limit(50).map(this::toResponse).collect(Collectors.toList());
}
+ public com.baomidou.mybatisplus.core.metadata.IPage queryRecords(String tenantCode, String taskId, LocalDateTime start,
+ LocalDateTime end, String channelType, Integer status,
+ long page, long size) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper()
+ .eq(SendRecord::getTenantCode, tenantCode)
+ .orderByDesc(SendRecord::getId);
+ if (StringUtils.hasText(taskId)) {
+ wrapper.eq(SendRecord::getTaskId, taskId);
+ }
+ if (start != null) {
+ wrapper.ge(SendRecord::getSendTime, start);
+ }
+ if (end != null) {
+ wrapper.le(SendRecord::getSendTime, end);
+ }
+ if (StringUtils.hasText(channelType)) {
+ wrapper.eq(SendRecord::getChannelType, channelType);
+ }
+ if (status != null) {
+ wrapper.eq(SendRecord::getSendStatus, status);
+ }
+ Page pageParam = new Page<>(page, size);
+ sendRecordMapper.selectPage(pageParam, wrapper);
+ return pageParam.convert(this::toResponse);
+ }
+
+ public SendTaskStatusResponse getTaskStatus(String tenantCode, String taskId) {
+ List records = sendRecordMapper.selectList(
+ new LambdaQueryWrapper()
+ .eq(SendRecord::getTenantCode, tenantCode)
+ .eq(SendRecord::getTaskId, taskId));
+ SendTaskStatusResponse response = new SendTaskStatusResponse();
+ response.setTaskId(taskId);
+ response.setTotal(records.size());
+ for (SendRecord record : records) {
+ switch (record.getSendStatus()) {
+ case SEND_STATUS_PENDING:
+ response.setPendingCount(response.getPendingCount() + 1);
+ break;
+ case SEND_STATUS_SENDING:
+ response.setSendingCount(response.getSendingCount() + 1);
+ break;
+ case SEND_STATUS_SUCCESS:
+ response.setSuccessCount(response.getSuccessCount() + 1);
+ break;
+ case SEND_STATUS_FAILED:
+ response.setFailedCount(response.getFailedCount() + 1);
+ break;
+ default:
+ break;
+ }
+ }
+ return response;
+ }
+
+ @Transactional
+ public int cancelTask(String tenantCode, String taskId) {
+ return sendRecordMapper.update(null,
+ new LambdaUpdateWrapper()
+ .eq(SendRecord::getTenantCode, tenantCode)
+ .eq(SendRecord::getTaskId, taskId)
+ .eq(SendRecord::getSendStatus, SEND_STATUS_PENDING)
+ .set(SendRecord::getSendStatus, SEND_STATUS_FAILED)
+ .set(SendRecord::getFinishTime, LocalDateTime.now())
+ .set(SendRecord::getChannelResult,
+ JsonUtils.toJson(Collections.singletonMap("canceled", true))));
+ }
+
+ public SendStatisticsResponse getStatistics(String tenantCode, LocalDateTime start, LocalDateTime end) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper()
+ .eq(SendRecord::getTenantCode, tenantCode);
+ if (start != null) {
+ wrapper.ge(SendRecord::getSendTime, start);
+ }
+ if (end != null) {
+ wrapper.le(SendRecord::getSendTime, end);
+ }
+ List records = sendRecordMapper.selectList(wrapper);
+
+ SendStatisticsResponse response = new SendStatisticsResponse();
+ Map channelMap = new HashMap<>();
+ for (SendRecord record : records) {
+ response.setTotal(response.getTotal() + 1);
+ switch (record.getSendStatus()) {
+ case SEND_STATUS_PENDING:
+ response.setPendingCount(response.getPendingCount() + 1);
+ break;
+ case SEND_STATUS_SENDING:
+ response.setSendingCount(response.getSendingCount() + 1);
+ break;
+ case SEND_STATUS_SUCCESS:
+ response.setSuccessCount(response.getSuccessCount() + 1);
+ break;
+ case SEND_STATUS_FAILED:
+ response.setFailedCount(response.getFailedCount() + 1);
+ break;
+ default:
+ break;
+ }
+ ChannelStatistics channelStats = channelMap.computeIfAbsent(record.getChannelType(),
+ k -> {
+ ChannelStatistics cs = new ChannelStatistics();
+ cs.setChannelType(k);
+ return cs;
+ });
+ channelStats.setTotal(channelStats.getTotal() + 1);
+ switch (record.getSendStatus()) {
+ case SEND_STATUS_PENDING:
+ channelStats.setPendingCount(channelStats.getPendingCount() + 1);
+ break;
+ case SEND_STATUS_SENDING:
+ channelStats.setSendingCount(channelStats.getSendingCount() + 1);
+ break;
+ case SEND_STATUS_SUCCESS:
+ channelStats.setSuccessCount(channelStats.getSuccessCount() + 1);
+ break;
+ case SEND_STATUS_FAILED:
+ channelStats.setFailedCount(channelStats.getFailedCount() + 1);
+ break;
+ default:
+ break;
+ }
+ }
+ response.setChannelStats(new ArrayList<>(channelMap.values()));
+ return response;
+ }
+
private String generateTaskId() {
return "SND" + UUID.randomUUID().toString().replace("-", "").toUpperCase();
}
- private List resolveTargets(String tenantCode, SendRequest request) {
+ List resolveTargets(String tenantCode, SendRequest request) {
return resolveByType(tenantCode, request.getTargetType(), request.getTargetValue(), request.getChannelStrategy());
}
- private List resolveTestTargets(String tenantCode, SendTestRequest request) {
+ List resolveTestTargets(String tenantCode, SendTestRequest request) {
return resolveByType(tenantCode, request.getTargetType(), request.getTargetValue(), request.getChannelStrategy());
}
@SuppressWarnings("unchecked")
- private List resolveByType(String tenantCode, String targetType, Map targetValue,
- Map channelStrategy) {
+ List resolveByType(String tenantCode, String targetType, Map targetValue,
+ Map channelStrategy) {
String channelTypeStr = extractChannelType(channelStrategy);
ChannelType channelType = parseChannelType(channelTypeStr);
@@ -183,7 +325,7 @@ public class SendService {
if (conversation == null || !tenantCode.equals(conversation.getTenantCode())) {
throw new BusinessException("会话不存在: " + conversationId);
}
- Long accountId = selectSendAccount(conversationId, channelType, channelStrategy);
+ Long accountId = selectSendAccount(tenantCode, conversationId, channelType, channelStrategy);
if (accountId == null) {
accountId = conversation.getOwnerAccountId();
}
@@ -532,22 +674,8 @@ public class SendService {
return value != null ? String.valueOf(value) : null;
}
- private Long selectSendAccount(Long conversationId, ChannelType channelType, Map channelStrategy) {
- String strategy = extractStrategy(channelStrategy);
- int bindingType = "backup".equals(strategy) ? BINDING_SEND_BACKUP : BINDING_SEND_PRIMARY;
- ChannelAccountConversation binding = accountConversationMapper.selectOne(
- new LambdaQueryWrapper()
- .eq(ChannelAccountConversation::getConversationId, conversationId)
- .eq(ChannelAccountConversation::getBindingType, bindingType)
- .eq(ChannelAccountConversation::getStatus, STATUS_ENABLED));
- if (binding != null) {
- ChannelAccount account = accountMapper.selectById(binding.getAccountId());
- if (account != null && STATUS_ENABLED == account.getStatus()
- && channelType.name().equalsIgnoreCase(account.getChannelType())) {
- return account.getId();
- }
- }
- return null;
+ private Long selectSendAccount(String tenantCode, Long conversationId, ChannelType channelType, Map channelStrategy) {
+ return accountSelector.selectSendAccount(tenantCode, conversationId, channelType, extractStrategy(channelStrategy));
}
private Long selectDefaultAccount(String tenantCode, ChannelType channelType) {
@@ -589,9 +717,20 @@ public class SendService {
}
}
+ private RetryConfig loadRetryConfig(String tenantCode, SendRequest request) {
+ String channelTypeStr = extractChannelType(request.getChannelStrategy());
+ PolicyGroup policyGroup = loadPolicyGroup(tenantCode, channelTypeStr);
+ if (policyGroup == null) {
+ return null;
+ }
+ PolicyGroupConfig config = PolicyGroupConfig.parseJson(policyGroup.getPolicies());
+ return config.getRetryConfig().orElse(null);
+ }
+
private List createRecords(String taskId, String tenantCode, SendRequest request,
- List targets) {
+ List targets, RetryConfig retryConfig) {
List records = new ArrayList<>();
+ int maxRetry = retryConfig != null ? retryConfig.getMaxRetry() : 0;
for (SendTarget target : targets) {
SendRecord record = new SendRecord();
record.setTaskId(taskId);
@@ -607,6 +746,8 @@ public class SendService {
record.setContentBody(JsonUtils.toJson(request.getContent()));
record.setCallbackUrl(request.getCallbackUrl());
record.setSendStatus(SEND_STATUS_PENDING);
+ record.setRetryCount(0);
+ record.setMaxRetry(maxRetry);
record.setSendTime(LocalDateTime.now());
sendRecordMapper.insert(record);
records.add(record);
@@ -641,7 +782,9 @@ public class SendService {
return records;
}
- private void dispatchRecord(SendRecord record) {
+ @Transactional
+ public void dispatchRecord(SendRecord record) {
+
ChannelType channelType = parseChannelType(record.getChannelType());
ChannelAdapter adapter = adapterRegistry.getAdapter(channelType);
if (adapter == null) {
@@ -652,26 +795,200 @@ public class SendService {
return;
}
- record.setSendStatus(SEND_STATUS_SENDING);
+ PolicyCheckResult policyResult = checkPolicy(record);
+ if (policyResult != null && !policyResult.isAllowed()) {
+ record.setSendStatus(SEND_STATUS_FAILED);
+ Map resultMap = new HashMap<>();
+ resultMap.put("policy_rejected", true);
+ resultMap.put("reason", policyResult.getReasonCode());
+ resultMap.put("message", policyResult.getReasonMessage());
+ record.setChannelResult(JsonUtils.toJson(resultMap));
+ record.setFinishTime(LocalDateTime.now());
+ sendRecordMapper.updateById(record);
+ return;
+ }
+
+ List candidateAccountIds = selectSendAccountCandidates(record, channelType);
+ if (candidateAccountIds.isEmpty()) {
+ record.setSendStatus(SEND_STATUS_FAILED);
+ record.setChannelResult(JsonUtils.toJson(Collections.singletonMap("error", "no_healthy_account")));
+ record.setFinishTime(LocalDateTime.now());
+ sendRecordMapper.updateById(record);
+ return;
+ }
+
+ List> attempts = new ArrayList<>();
+ boolean success = false;
+ SendMessageResult lastResult = null;
+
+ for (Long accountId : candidateAccountIds) {
+ record.setAccountId(accountId);
+ record.setSendStatus(SEND_STATUS_SENDING);
+ sendRecordMapper.updateById(record);
+
+ ChannelAccount account = accountMapper.selectById(accountId);
+ SendMessageResult result = doSendAttempt(record, adapter, account);
+ lastResult = result;
+
+ attempts.add(buildAttemptResult(accountId, result));
+
+ if (result != null && result.isSuccess()) {
+ success = true;
+ break;
+ }
+
+ markAccountUnhealthy(account);
+ }
+
+ record.setFinishTime(LocalDateTime.now());
+ if (success) {
+ record.setSendStatus(SEND_STATUS_SUCCESS);
+ record.setChannelResult(JsonUtils.toJson(buildSuccessResult(lastResult, record.getAccountId(), attempts)));
+ } else {
+ record.setSendStatus(SEND_STATUS_FAILED);
+ record.setChannelResult(JsonUtils.toJson(buildAllAttemptsFailureResult(attempts, policyResult)));
+ }
sendRecordMapper.updateById(record);
+ }
+
+ private List selectSendAccountCandidates(SendRecord record, ChannelType channelType) {
+ if (record.getConversationId() == null) {
+ return record.getAccountId() != null ? Collections.singletonList(record.getAccountId()) : Collections.emptyList();
+ }
+ String preferredStrategy = extractStrategy(parseChannelStrategy(record.getChannelStrategy()));
+ return accountSelector.selectSendAccounts(record.getTenantCode(), record.getConversationId(), channelType, preferredStrategy);
+ }
+
+ private Map