feat(channel): add event handling service

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<String, Object> data;
+}

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<ChannelAccount>()
+                        .eq(ChannelAccount::getTenantCode, tenantCode)
+                        .eq(ChannelAccount::getAccountId, accountId));
+        if (account == null) {
+            throw new BusinessException("渠道账号不存在: " + accountId);
+        }
+        return account;
+    }
+
+    private void handleHeartbeat(ChannelAccount account, Map<String, Object> 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<String, Object> 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<String, Object> data) {
+        account.setOnlineStatus(parseOnlineStatus(getString(data, "current")));
+        accountMapper.updateById(account);
+    }
+
+    private void handleSendResult(Map<String, Object> 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<String, Object> map, String key) {
+        Object value = map.get(key);
+        return value == null ? null : value.toString();
+    }
+
+    private Integer getInteger(Map<String, Object> map, String key) {
+        Object value = map.get(key);
+        if (value instanceof Number n) {
+            return n.intValue();
+        }
+        return null;
+    }
+
+    private Long getLong(Map<String, Object> 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<String, Object> 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();
+        }
+    }
+}