server/sino-mci
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(channel): 账号心跳与状态监控

Browse Source
This commit is contained in:
王宇航
2026-07-07 20:25:23 +08:00
parent 79fed8b607
commit d28c959494
8 changed files with 223 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/mci-server/src/main/java/com/sino/mci/MciServerApplication.java
View File

@@ -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 {

3
backend/mci-server/src/main/java/com/sino/mci/admin/security/AuthInterceptor.java
View File

@@ -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;
}

6
backend/mci-server/src/main/java/com/sino/mci/channel/controller/ChannelAccountController.java
View File

@@ -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<ChannelAccount> heartbeat(@Valid @RequestBody HeartbeatRequest request) {
return ApiResponse.ok(accountService.heartbeat(request.getAccountId(), request));
}
@PostMapping("/init")
public ApiResponse<?> initAccount(
@RequestHeader("X-Tenant-Code") String tenantCode,

19
backend/mci-server/src/main/java/com/sino/mci/channel/dto/HeartbeatRequest.java Normal file
View File

@@ -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;
}

25
backend/mci-server/src/main/java/com/sino/mci/channel/scheduler/ChannelAccountHealthScheduler.java Normal file
View File

@@ -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);
}
}
}

53
backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountService.java
View File

@@ -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<ChannelAccount>()
.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<ChannelAccount> accounts = accountMapper.selectList(
new LambdaQueryWrapper<ChannelAccount>()
.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());

3
backend/mci-server/src/main/java/com/sino/mci/config/WebConfig.java
View File

@@ -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");
}
}

114
backend/mci-server/src/test/java/com/sino/mci/channel/ChannelAccountHeartbeatTest.java Normal file
View File

@@ -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<ChannelAccount>()
.eq(ChannelAccount::getTenantCode, TENANT_CODE)
.eq(ChannelAccount::getAccountId, ACCOUNT_ID));
}
}