diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java b/backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java new file mode 100644 index 0000000..2fc1352 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/dto/ChannelAccountEventRequest.java @@ -0,0 +1,28 @@ +package com.sino.mci.channel.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +import java.util.Map; + +@Data +public class ChannelAccountEventRequest { + + @NotBlank + private String eventId; + + @NotBlank + private String eventType; + + @NotBlank + private String accountId; + + @NotBlank + private String tenantCode; + + private String timestamp; + + @NotNull + private Map data; +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java b/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java new file mode 100644 index 0000000..12dc7b5 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/channel/service/ChannelAccountEventService.java @@ -0,0 +1,217 @@ +package com.sino.mci.channel.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sino.mci.channel.dto.ChannelAccountEventRequest; +import com.sino.mci.channel.entity.ChannelAccount; +import com.sino.mci.channel.entity.ChannelEventRecord; +import com.sino.mci.channel.repository.ChannelAccountMapper; +import com.sino.mci.channel.repository.ChannelEventRecordMapper; +import com.sino.mci.exception.BusinessException; +import com.sino.mci.send.entity.SendRecord; +import com.sino.mci.send.repository.SendRecordMapper; +import com.sino.mci.send.service.SendCallbackService; +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.Map; + +@Slf4j +@Service +@RequiredArgsConstructor +public class ChannelAccountEventService { + + private static final int LOGIN_STATUS_PENDING = 0; + private static final int LOGIN_STATUS_SUCCESS = 1; + private static final int LOGIN_STATUS_EXPIRED = 2; + private static final int LOGIN_STATUS_FAILED = 3; + + private static final int ONLINE_STATUS_OFFLINE = 0; + private static final int ONLINE_STATUS_ONLINE = 1; + private static final int ONLINE_STATUS_ERROR = 2; + + private static final int SEND_STATUS_SUCCESS = 2; + private static final int SEND_STATUS_FAILED = 3; + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private final ChannelAccountMapper accountMapper; + private final ChannelEventRecordMapper eventRecordMapper; + private final SendRecordMapper sendRecordMapper; + private final SendCallbackService sendCallbackService; + + @Transactional + public void handleEvent(ChannelAccountEventRequest request) { + if (isDuplicate(request.getEventId())) { + log.debug("重复事件已忽略: eventId={}", request.getEventId()); + return; + } + + ChannelAccount account = findAccount(request.getTenantCode(), request.getAccountId()); + + switch (request.getEventType()) { + case "heartbeat": + handleHeartbeat(account, request.getData()); + break; + case "login_status_changed": + handleLoginStatusChanged(account, request.getData()); + break; + case "online_status_changed": + handleOnlineStatusChanged(account, request.getData()); + break; + case "send_result": + handleSendResult(request.getData()); + break; + case "error": + log.warn("channel-service 上报错误: accountId={}, data={}", request.getAccountId(), request.getData()); + break; + default: + log.warn("未知事件类型: {}", request.getEventType()); + } + + saveEventRecord(request); + } + + private boolean isDuplicate(String eventId) { + return eventRecordMapper.selectById(eventId) != null; + } + + private void saveEventRecord(ChannelAccountEventRequest request) { + ChannelEventRecord record = new ChannelEventRecord(); + record.setEventId(request.getEventId()); + record.setAccountId(request.getAccountId()); + record.setEventType(request.getEventType()); + record.setCreateTime(LocalDateTime.now()); + eventRecordMapper.insert(record); + } + + private ChannelAccount findAccount(String tenantCode, String accountId) { + ChannelAccount account = accountMapper.selectOne( + new LambdaQueryWrapper() + .eq(ChannelAccount::getTenantCode, tenantCode) + .eq(ChannelAccount::getAccountId, accountId)); + if (account == null) { + throw new BusinessException("渠道账号不存在: " + accountId); + } + return account; + } + + private void handleHeartbeat(ChannelAccount account, Map data) { + account.setLoginStatus(parseLoginStatus(getString(data, "login_status"))); + account.setOnlineStatus(parseOnlineStatus(getString(data, "online_status"))); + account.setLastHeartbeat(LocalDateTime.now()); + Integer riskLevel = getInteger(data, "risk_level"); + if (riskLevel != null) { + account.setRiskLevel(riskLevel); + } + accountMapper.updateById(account); + } + + private void handleLoginStatusChanged(ChannelAccount account, Map data) { + String current = getString(data, "current"); + account.setLoginStatus(parseLoginStatus(current)); + if ("success".equalsIgnoreCase(current)) { + account.setOnlineStatus(ONLINE_STATUS_ONLINE); + } + accountMapper.updateById(account); + } + + private void handleOnlineStatusChanged(ChannelAccount account, Map data) { + account.setOnlineStatus(parseOnlineStatus(getString(data, "current"))); + accountMapper.updateById(account); + } + + private void handleSendResult(Map data) { + Long recordId = getLong(data, "record_id"); + if (recordId == null) { + log.warn("send_result 事件缺少 record_id"); + return; + } + SendRecord record = sendRecordMapper.selectById(recordId); + if (record == null) { + log.warn("发送记录不存在: recordId={}", recordId); + return; + } + + Boolean success = getBoolean(data, "success"); + record.setSendStatus(Boolean.TRUE.equals(success) ? SEND_STATUS_SUCCESS : SEND_STATUS_FAILED); + record.setFinishTime(LocalDateTime.now()); + record.setChannelResult(toJson(data.get("channel_result"))); + sendRecordMapper.updateById(record); + + sendCallbackService.sendCallback(record); + } + + private int parseLoginStatus(String status) { + if ("success".equalsIgnoreCase(status)) { + return LOGIN_STATUS_SUCCESS; + } + if ("expired".equalsIgnoreCase(status)) { + return LOGIN_STATUS_EXPIRED; + } + if ("failed".equalsIgnoreCase(status)) { + return LOGIN_STATUS_FAILED; + } + return LOGIN_STATUS_PENDING; + } + + private int parseOnlineStatus(String status) { + if ("online".equalsIgnoreCase(status)) { + return ONLINE_STATUS_ONLINE; + } + if ("error".equalsIgnoreCase(status)) { + return ONLINE_STATUS_ERROR; + } + return ONLINE_STATUS_OFFLINE; + } + + private String getString(Map map, String key) { + Object value = map.get(key); + return value == null ? null : value.toString(); + } + + private Integer getInteger(Map map, String key) { + Object value = map.get(key); + if (value instanceof Number n) { + return n.intValue(); + } + return null; + } + + private Long getLong(Map map, String key) { + Object value = map.get(key); + if (value instanceof Number n) { + return n.longValue(); + } + if (value instanceof String s) { + try { + return Long.parseLong(s); + } catch (NumberFormatException e) { + return null; + } + } + return null; + } + + private Boolean getBoolean(Map map, String key) { + Object value = map.get(key); + if (value instanceof Boolean b) { + return b; + } + return null; + } + + private String toJson(Object value) { + if (value == null) { + return null; + } + try { + return OBJECT_MAPPER.writeValueAsString(value); + } catch (Exception e) { + return value.toString(); + } + } +}