From d28c959494f09dd50cb667b855f3a6a9ecc24ac8 Mon Sep 17 00:00:00 2001 From: marsal Date: Tue, 7 Jul 2026 20:25:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E8=B4=A6=E5=8F=B7=E5=BF=83?= =?UTF-8?q?=E8=B7=B3=E4=B8=8E=E7=8A=B6=E6=80=81=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/sino/mci/MciServerApplication.java | 2 + .../mci/admin/security/AuthInterceptor.java | 3 +- .../controller/ChannelAccountController.java | 6 + .../mci/channel/dto/HeartbeatRequest.java | 19 +++ .../ChannelAccountHealthScheduler.java | 25 ++++ .../service/ChannelAccountService.java | 53 ++++++++ .../java/com/sino/mci/config/WebConfig.java | 3 +- .../channel/ChannelAccountHeartbeatTest.java | 114 ++++++++++++++++++ 8 files changed, 223 insertions(+), 2 deletions(-) create mode 100644 backend/mci-server/src/main/java/com/sino/mci/channel/dto/HeartbeatRequest.java create mode 100644 backend/mci-server/src/main/java/com/sino/mci/channel/scheduler/ChannelAccountHealthScheduler.java create mode 100644 backend/mci-server/src/test/java/com/sino/mci/channel/ChannelAccountHeartbeatTest.java diff --git a/backend/mci-server/src/main/java/com/sino/mci/MciServerApplication.java b/backend/mci-server/src/main/java/com/sino/mci/MciServerApplication.java index cc1efe0..881bafa 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/MciServerApplication.java +++ b/backend/mci-server/src/main/java/com/sino/mci/MciServerApplication.java @@ -3,8 +3,10 @@ package com.sino.mci; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.annotation.EnableScheduling; @EnableAsync +@EnableScheduling @SpringBootApplication public class MciServerApplication { diff --git a/backend/mci-server/src/main/java/com/sino/mci/admin/security/AuthInterceptor.java b/backend/mci-server/src/main/java/com/sino/mci/admin/security/AuthInterceptor.java index 913d43a..815e811 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/admin/security/AuthInterceptor.java +++ b/backend/mci-server/src/main/java/com/sino/mci/admin/security/AuthInterceptor.java @@ -17,7 +17,8 @@ public class AuthInterceptor implements HandlerInterceptor { if (path.contains("/api/v1/account/login") || ("POST".equals(method) && isExactPath(path, "/api/v1/account/tenant")) || ("POST".equals(method) && isExactPath(path, "/api/v1/account/user")) - || isExactPath(path, "/api/v1/health")) { + || isExactPath(path, "/api/v1/health") + || ("POST".equals(method) && isExactPath(path, "/api/v1/channel-account/heartbeat"))) { return true; } diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountController.java b/backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountController.java index 3ce7216..7bf640a 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountController.java +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountController.java @@ -3,6 +3,7 @@ package com.sino.mci.channel.controller; import com.sino.mci.admin.security.AuthContext; import com.sino.mci.channel.dto.ChannelAccountInitRequest; import com.sino.mci.channel.dto.CreateChannelAccountRequest; +import com.sino.mci.channel.dto.HeartbeatRequest; import com.sino.mci.channel.dto.SyncResult; import com.sino.mci.channel.entity.ChannelAccount; import com.sino.mci.channel.service.ChannelAccountService; @@ -27,6 +28,11 @@ public class ChannelAccountController { return ApiResponse.ok(accountService.createAccount(tenantCode, request)); } + @PostMapping("/heartbeat") + public ApiResponse heartbeat(@Valid @RequestBody HeartbeatRequest request) { + return ApiResponse.ok(accountService.heartbeat(request.getAccountId(), request)); + } + @PostMapping("/init") public ApiResponse initAccount( @RequestHeader("X-Tenant-Code") String tenantCode, diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/dto/HeartbeatRequest.java b/backend/mci-server/src/main/java/com/sino/mci/channel/dto/HeartbeatRequest.java new file mode 100644 index 0000000..05ead21 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/dto/HeartbeatRequest.java @@ -0,0 +1,19 @@ +package com.sino.mci.channel.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +@Data +public class HeartbeatRequest { + + @NotBlank + private String accountId; + + private String accountName; + + @NotNull + private Integer riskLevel; + + private Integer sentCountToday; +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/scheduler/ChannelAccountHealthScheduler.java b/backend/mci-server/src/main/java/com/sino/mci/channel/scheduler/ChannelAccountHealthScheduler.java new file mode 100644 index 0000000..c371d82 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/scheduler/ChannelAccountHealthScheduler.java @@ -0,0 +1,25 @@ +package com.sino.mci.channel.scheduler; + +import com.sino.mci.channel.service.ChannelAccountService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class ChannelAccountHealthScheduler { + + private static final int OFFLINE_TIMEOUT_SECONDS = 120; + + private final ChannelAccountService accountService; + + @Scheduled(fixedDelayString = "${channel.account.health.interval:60000}") + public void checkOfflineAccounts() { + int count = accountService.checkOfflineAccounts(OFFLINE_TIMEOUT_SECONDS); + if (count > 0) { + log.info("账号健康检查完成，共标记 {} 个账号为离线", count); + } + } +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountService.java b/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountService.java index 459e709..a66c5b1 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountService.java +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountService.java @@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.sino.mci.channel.common.model.ChannelType; import com.sino.mci.channel.dto.ChannelAccountInitRequest; import com.sino.mci.channel.dto.CreateChannelAccountRequest; +import com.sino.mci.channel.dto.HeartbeatRequest; import com.sino.mci.channel.dto.SyncResult; import com.sino.mci.channel.entity.ChannelAccount; import com.sino.mci.channel.entity.ChannelAccountConversation; @@ -28,12 +29,14 @@ import com.sino.mci.crm.repository.PersonMapper; import com.sino.mci.exception.BusinessException; import com.sino.mci.shared.util.JsonUtils; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.List; +@Slf4j @Service @RequiredArgsConstructor public class ChannelAccountService { @@ -50,6 +53,11 @@ public class ChannelAccountService { private static final int PARTICIPANT_TYPE_MEMBER = 1; private static final int PARTICIPANT_TYPE_OWNER = 2; + private static final int ONLINE_STATUS_OFFLINE = 0; + private static final int ONLINE_STATUS_ONLINE = 1; + private static final int ACCOUNT_STATUS_ENABLED = 1; + private static final int RISK_LEVEL_HIGH = 80; + private final ChannelAccountMapper accountMapper; private final ChannelSubjectMapper subjectMapper; private final PersonMapper personMapper; @@ -103,6 +111,51 @@ public class ChannelAccountService { return account; } + @Transactional + public ChannelAccount heartbeat(String accountCode, HeartbeatRequest request) { + ChannelAccount account = accountMapper.selectOne( + new LambdaQueryWrapper() + .eq(ChannelAccount::getAccountId, accountCode) + .eq(ChannelAccount::getStatus, ACCOUNT_STATUS_ENABLED)); + if (account == null) { + throw new BusinessException("渠道账号不存在: " + accountCode); + } + + account.setOnlineStatus(ONLINE_STATUS_ONLINE); + account.setLastHeartbeat(LocalDateTime.now()); + if (request != null && request.getRiskLevel() != null) { + account.setRiskLevel(request.getRiskLevel()); + } + accountMapper.updateById(account); + + if (account.getRiskLevel() >= RISK_LEVEL_HIGH) { + log.warn("账号高风险: accountId={}, riskLevel={}", account.getAccountId(), account.getRiskLevel()); + } + + return account; + } + + @Transactional + public int checkOfflineAccounts(int timeoutSeconds) { + LocalDateTime threshold = LocalDateTime.now().minusSeconds(timeoutSeconds); + List accounts = accountMapper.selectList( + new LambdaQueryWrapper() + .eq(ChannelAccount::getOnlineStatus, ONLINE_STATUS_ONLINE) + .eq(ChannelAccount::getStatus, ACCOUNT_STATUS_ENABLED) + .and(w -> w.isNull(ChannelAccount::getLastHeartbeat) + .or() + .lt(ChannelAccount::getLastHeartbeat, threshold))); + + int offlineCount = 0; + for (ChannelAccount account : accounts) { + log.warn("账号离线: accountId={}, lastHeartbeat={}", account.getAccountId(), account.getLastHeartbeat()); + account.setOnlineStatus(ONLINE_STATUS_OFFLINE); + accountMapper.updateById(account); + offlineCount++; + } + return offlineCount; + } + public ChannelAccount initAccount(String tenantCode, ChannelAccountInitRequest request) { // Phase 2 placeholder: actual adapter dispatch will be implemented later throw new BusinessException("渠道账号初始化适配器尚未实现: " + request.getChannelType()); diff --git a/backend/mci-server/src/main/java/com/sino/mci/config/WebConfig.java b/backend/mci-server/src/main/java/com/sino/mci/config/WebConfig.java index 9aa5198..5eaecc6 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/config/WebConfig.java +++ b/backend/mci-server/src/main/java/com/sino/mci/config/WebConfig.java @@ -20,6 +20,7 @@ public class WebConfig implements WebMvcConfigurer { "/msg-platform/api/v1/account/login", "/api/v1/account/login", "/msg-platform/api/v1/account/tenant", "/api/v1/account/tenant", "/msg-platform/api/v1/account/user", "/api/v1/account/user", - "/msg-platform/api/v1/health", "/api/v1/health"); + "/msg-platform/api/v1/health", "/api/v1/health", + "/msg-platform/api/v1/channel-account/heartbeat", "/api/v1/channel-account/heartbeat"); } } diff --git a/backend/mci-server/src/test/java/com/sino/mci/channel/ChannelAccountHeartbeatTest.java b/backend/mci-server/src/test/java/com/sino/mci/channel/ChannelAccountHeartbeatTest.java new file mode 100644 index 0000000..fd659bd --- /dev/null +++ b/backend/mci-server/src/test/java/com/sino/mci/channel/ChannelAccountHeartbeatTest.java @@ -0,0 +1,114 @@ +package com.sino.mci.channel; + +import com.sino.mci.channel.entity.ChannelAccount; +import com.sino.mci.channel.repository.ChannelAccountMapper; +import com.sino.mci.channel.service.ChannelAccountService; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; +import org.springframework.http.MediaType; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.transaction.annotation.Transactional; + +import java.time.LocalDateTime; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +@SpringBootTest +@AutoConfigureMockMvc +@TestPropertySource(properties = { + "spring.rabbitmq.listener.simple.auto-startup=false", + "spring.main.allow-bean-definition-overriding=true", + "channel.account.health.interval=31536000000" +}) +@Transactional +class ChannelAccountHeartbeatTest { + + private static final String TENANT_CODE = "heartbeat-" + UUID.randomUUID().toString().substring(0, 8); + private static final String ACCOUNT_ID = "heartbeat-account-" + UUID.randomUUID().toString().substring(0, 8); + + @Autowired + private MockMvc mockMvc; + + @Autowired + private ChannelAccountMapper accountMapper; + + @Autowired + private ChannelAccountService accountService; + + @Test + void shouldUpdateAccountStatusOnHeartbeat() throws Exception { + createAccount(); + + String payload = String.format("{\"accountId\":\"%s\",\"riskLevel\":30}", ACCOUNT_ID); + mockMvc.perform(post("/msg-platform/api/v1/channel-account/heartbeat") + .contextPath("/msg-platform") + .contentType(MediaType.APPLICATION_JSON) + .content(payload)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value(0)) + .andExpect(jsonPath("$.data.accountId").value(ACCOUNT_ID)) + .andExpect(jsonPath("$.data.onlineStatus").value(1)) + .andExpect(jsonPath("$.data.riskLevel").value(30)); + + ChannelAccount account = findAccount(); + assertThat(account.getOnlineStatus()).isEqualTo(1); + assertThat(account.getRiskLevel()).isEqualTo(30); + assertThat(account.getLastHeartbeat()).isNotNull(); + } + + @Test + void shouldMarkAccountOfflineWhenHeartbeatTimeout() { + ChannelAccount account = createAccount(); + account.setOnlineStatus(1); + account.setLastHeartbeat(LocalDateTime.now().minusSeconds(150)); + accountMapper.updateById(account); + + int offlineCount = accountService.checkOfflineAccounts(120); + + assertThat(offlineCount).isEqualTo(1); + ChannelAccount updated = accountMapper.selectById(account.getId()); + assertThat(updated.getOnlineStatus()).isEqualTo(0); + } + + @Test + void shouldKeepOnlineAccountWhenHeartbeatWithinTimeout() { + ChannelAccount account = createAccount(); + account.setOnlineStatus(1); + account.setLastHeartbeat(LocalDateTime.now().minusSeconds(60)); + accountMapper.updateById(account); + + int offlineCount = accountService.checkOfflineAccounts(120); + + assertThat(offlineCount).isEqualTo(0); + ChannelAccount updated = accountMapper.selectById(account.getId()); + assertThat(updated.getOnlineStatus()).isEqualTo(1); + } + + private ChannelAccount createAccount() { + ChannelAccount account = new ChannelAccount(); + account.setTenantCode(TENANT_CODE); + account.setAccountType(3); + account.setChannelType("wechat_personal"); + account.setAccountId(ACCOUNT_ID); + account.setAccountName("心跳测试账号"); + account.setOnlineStatus(0); + account.setRiskLevel(0); + account.setStatus(1); + accountMapper.insert(account); + return account; + } + + private ChannelAccount findAccount() { + return accountMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(ChannelAccount::getTenantCode, TENANT_CODE) + .eq(ChannelAccount::getAccountId, ACCOUNT_ID)); + } +}