|
|
@@ -0,0 +1,321 @@
|
|
|
+package com.usky.issue.service.client;
|
|
|
+
|
|
|
+import com.fasterxml.jackson.databind.JsonNode;
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
+import com.usky.issue.domain.IssueCloudConfig;
|
|
|
+import com.usky.issue.service.constant.CloudIntegrationConstants;
|
|
|
+import com.usky.issue.service.support.CloudEntitySupport;
|
|
|
+import com.usky.issue.service.util.AesGcmCipher;
|
|
|
+import com.usky.issue.service.util.CredentialMaskUtil;
|
|
|
+import com.usky.issue.service.util.HmacSignUtil;
|
|
|
+import com.usky.issue.service.vo.CloudConnectionTestResult;
|
|
|
+import com.usky.issue.service.vo.CloudPollResult;
|
|
|
+import com.usky.issue.service.vo.SyncPacket;
|
|
|
+import com.usky.issue.service.vo.SyncPollRequest;
|
|
|
+import com.usky.issue.service.vo.SyncResponse;
|
|
|
+import com.usky.issue.service.util.ApiResultParser;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.boot.web.client.RestTemplateBuilder;
|
|
|
+import org.springframework.http.HttpEntity;
|
|
|
+import org.springframework.http.HttpHeaders;
|
|
|
+import org.springframework.http.HttpMethod;
|
|
|
+import org.springframework.http.MediaType;
|
|
|
+import org.springframework.http.ResponseEntity;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+import org.springframework.util.StringUtils;
|
|
|
+import org.springframework.web.client.HttpClientErrorException;
|
|
|
+import org.springframework.web.client.HttpServerErrorException;
|
|
|
+import org.springframework.web.client.HttpStatusCodeException;
|
|
|
+import org.springframework.web.client.ResourceAccessException;
|
|
|
+import org.springframework.web.client.RestTemplate;
|
|
|
+import org.springframework.web.util.UriComponentsBuilder;
|
|
|
+
|
|
|
+import java.time.Duration;
|
|
|
+import java.util.List;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 云平台 HTTP 客户端(连通性测试 + 双向同步)
|
|
|
+ *
|
|
|
+ * @author fyc
|
|
|
+ * @date 2026-05-21
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Component
|
|
|
+public class CloudPlatformClient {
|
|
|
+
|
|
|
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
|
|
+
|
|
|
+ private final RestTemplate restTemplate;
|
|
|
+ private final AesGcmCipher aesGcmCipher;
|
|
|
+
|
|
|
+ private static final String URL_END = "/service-issue/cloud/integration/testConnection";
|
|
|
+ private static final String INTEGRATION_PATH = "/service-issue/cloud/integration";
|
|
|
+
|
|
|
+ public CloudPlatformClient(RestTemplateBuilder restTemplateBuilder, AesGcmCipher aesGcmCipher) {
|
|
|
+ this.restTemplate = restTemplateBuilder
|
|
|
+ .setConnectTimeout(Duration.ofMillis(CloudIntegrationConstants.CONNECTION_TEST_TIMEOUT_MS))
|
|
|
+ .setReadTimeout(Duration.ofMillis(CloudIntegrationConstants.CONNECTION_TEST_TIMEOUT_MS))
|
|
|
+ .build();
|
|
|
+ this.aesGcmCipher = aesGcmCipher;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 使用请求参数发起连通性测试:cloudAddress + URL_END
|
|
|
+ */
|
|
|
+ public CloudConnectionTestResult testConnection(String cloudAddress, Integer tenantId,
|
|
|
+ String credentialKey, String token) {
|
|
|
+ if (!StringUtils.hasText(cloudAddress) || tenantId == null || !StringUtils.hasText(credentialKey)) {
|
|
|
+ return CloudConnectionTestResult.failure(false, "测试参数不完整");
|
|
|
+ }
|
|
|
+
|
|
|
+ String testUrl = buildTestConnectionUrl(cloudAddress);
|
|
|
+ long start = System.currentTimeMillis();
|
|
|
+ log.info("[连接测试-HTTP] 准备请求 url={}, tenantId={}, credentialKey={}, hasToken={}",
|
|
|
+ testUrl, tenantId, CredentialMaskUtil.mask(credentialKey), StringUtils.hasText(token));
|
|
|
+
|
|
|
+ try {
|
|
|
+ HttpHeaders headers = buildAuthHeaders(tenantId, credentialKey, token);
|
|
|
+ ResponseEntity<String> response = restTemplate.exchange(
|
|
|
+ testUrl, HttpMethod.POST, new HttpEntity<>(headers), String.class);
|
|
|
+ log.info("[连接测试-HTTP] 响应 status={}, body={}", response.getStatusCodeValue(), response.getBody());
|
|
|
+ return parseConnectionTestResult(response.getBody(), System.currentTimeMillis() - start);
|
|
|
+ } catch (ResourceAccessException ex) {
|
|
|
+ log.error("云平台网络不可达: url={}, error={}", testUrl, ex.getMessage());
|
|
|
+ return CloudConnectionTestResult.networkFailure(CloudIntegrationConstants.MSG_NETWORK_UNREACHABLE);
|
|
|
+ } catch (HttpClientErrorException.Unauthorized ex) {
|
|
|
+ return CloudConnectionTestResult.authFailure(System.currentTimeMillis() - start, "认证失败:凭证无效");
|
|
|
+ } catch (HttpClientErrorException | HttpServerErrorException ex) {
|
|
|
+ log.error("云平台连接测试失败: url={}, status={}, body={}",
|
|
|
+ testUrl, ex.getStatusCode(), ex.getResponseBodyAsString());
|
|
|
+ return CloudConnectionTestResult.authFailure(System.currentTimeMillis() - start,
|
|
|
+ "云平台返回异常:" + ex.getStatusCode());
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("云平台连接测试失败: url={}, error={}", testUrl, ex.getMessage(), ex);
|
|
|
+ return CloudConnectionTestResult.failure(false, ex.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 使用已保存配置发起连通性测试(同步等场景)
|
|
|
+ */
|
|
|
+ public CloudConnectionTestResult testConnection(IssueCloudConfig config) {
|
|
|
+ if (config == null || config.getTenantId() == null || !StringUtils.hasText(config.getCloudAddress())
|
|
|
+ || config.getCredentialKey() == null) {
|
|
|
+ log.warn("云平台配置不完整");
|
|
|
+ return CloudConnectionTestResult.failure(false, "云平台配置不完整");
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ String credential = aesGcmCipher.decrypt(config.getCredentialKey());
|
|
|
+ String token = CloudEntitySupport.resolveAccessToken(config);
|
|
|
+ return testConnection(config.getCloudAddress(), config.getTenantId(), credential, token);
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("凭证解密失败 tenantId={}", config.getTenantId(), ex);
|
|
|
+ return CloudConnectionTestResult.failure(false, "凭证解密失败");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public SyncResponse pushToCloud(SyncPacket packet, IssueCloudConfig config) {
|
|
|
+ try {
|
|
|
+ String credential = aesGcmCipher.decrypt(config.getCredentialKey());
|
|
|
+ String body = OBJECT_MAPPER.writeValueAsString(packet);
|
|
|
+ HttpHeaders headers = buildAuthHeaders(config, credential);
|
|
|
+ headers.setContentType(MediaType.APPLICATION_JSON);
|
|
|
+ headers.add(CloudIntegrationConstants.HEADER_SIGN, HmacSignUtil.sign(body, credential));
|
|
|
+
|
|
|
+ ResponseEntity<String> response = restTemplate.exchange(
|
|
|
+ resolveIntegrationBaseUrl(config) + CloudIntegrationConstants.PATH_SYNC_RECEIVE,
|
|
|
+ HttpMethod.POST,
|
|
|
+ new HttpEntity<>(body, headers),
|
|
|
+ String.class);
|
|
|
+ return parseSyncResponse(response.getBody());
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("推送云端网络异常", e);
|
|
|
+ return SyncResponse.builder().success(false).message(e.getMessage()).build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public CloudPollResult pollCloud(Integer tenantId, String tableName, Long lastVersion,
|
|
|
+ IssueCloudConfig config, boolean forceBackfill) {
|
|
|
+ try {
|
|
|
+ String credential = aesGcmCipher.decrypt(config.getCredentialKey());
|
|
|
+ HttpHeaders headers = buildAuthHeaders(config, credential);
|
|
|
+ headers.setContentType(MediaType.APPLICATION_JSON);
|
|
|
+
|
|
|
+ SyncPollRequest pollRequest = new SyncPollRequest();
|
|
|
+ pollRequest.setTenantId(String.valueOf(tenantId));
|
|
|
+ pollRequest.setTableName(tableName);
|
|
|
+ pollRequest.setLastVersion(lastVersion == null ? 0L : lastVersion);
|
|
|
+ pollRequest.setForceBackfill(forceBackfill);
|
|
|
+ String requestBody = OBJECT_MAPPER.writeValueAsString(pollRequest);
|
|
|
+
|
|
|
+ String url = resolveIntegrationBaseUrl(config) + CloudIntegrationConstants.PATH_SYNC_POLL;
|
|
|
+ log.info("[pollCloud] POST url={}, tenantId={}, tableName={}, lastVersion={}, forceBackfill={}, hasToken={}",
|
|
|
+ url, tenantId, tableName, lastVersion, forceBackfill,
|
|
|
+ StringUtils.hasText(CloudEntitySupport.resolveAccessToken(config)));
|
|
|
+
|
|
|
+ ResponseEntity<String> response = restTemplate.exchange(
|
|
|
+ url, HttpMethod.POST, new HttpEntity<>(requestBody, headers), String.class);
|
|
|
+ return parsePollResponse(response.getBody());
|
|
|
+ } catch (HttpStatusCodeException ex) {
|
|
|
+ log.warn("[pollCloud] HTTP {} tenantId={}, body={}",
|
|
|
+ ex.getStatusCode(), tenantId, ex.getResponseBodyAsString());
|
|
|
+ return parsePollResponse(ex.getResponseBodyAsString());
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("[pollCloud] 轮询云端失败 tenantId={}, tableName={}", tenantId, tableName, e);
|
|
|
+ return CloudPollResult.builder().requestFailed(true).errorMessage(e.getMessage()).build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private CloudPollResult parsePollResponse(String body) {
|
|
|
+ if (!StringUtils.hasText(body)) {
|
|
|
+ return CloudPollResult.builder().requestFailed(true).errorMessage("云端返回空响应").build();
|
|
|
+ }
|
|
|
+ log.info("[pollCloud] 响应 body={}", body);
|
|
|
+ try {
|
|
|
+ JsonNode json = OBJECT_MAPPER.readTree(body);
|
|
|
+ if (!ApiResultParser.isSuccess(json)) {
|
|
|
+ String msg = ApiResultParser.resolveMessage(json);
|
|
|
+ log.warn("[pollCloud] 业务失败 msg={}", msg);
|
|
|
+ boolean authFailed = msg != null && (msg.contains("认证") || msg.contains("令牌"));
|
|
|
+ return CloudPollResult.builder()
|
|
|
+ .authFailed(authFailed)
|
|
|
+ .requestFailed(true)
|
|
|
+ .errorMessage(StringUtils.hasText(msg) ? msg : "poll 调用失败")
|
|
|
+ .build();
|
|
|
+ }
|
|
|
+ JsonNode dataNode = json.has("data") ? json.get("data") : null;
|
|
|
+ if (dataNode == null || dataNode.isNull()) {
|
|
|
+ return CloudPollResult.builder().packet(null).build();
|
|
|
+ }
|
|
|
+ SyncPacket packet = OBJECT_MAPPER.treeToValue(dataNode, SyncPacket.class);
|
|
|
+ return CloudPollResult.builder().packet(packet).build();
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("[pollCloud] 响应解析失败 body={}", body, e);
|
|
|
+ return CloudPollResult.builder().requestFailed(true).errorMessage("响应解析失败: " + e.getMessage()).build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public SyncResponse ackPoll(List<Long> queueIds, IssueCloudConfig config) {
|
|
|
+ try {
|
|
|
+ String credential = aesGcmCipher.decrypt(config.getCredentialKey());
|
|
|
+ HttpHeaders headers = buildAuthHeaders(config, credential);
|
|
|
+ headers.setContentType(MediaType.APPLICATION_JSON);
|
|
|
+ String body = OBJECT_MAPPER.writeValueAsString(queueIds);
|
|
|
+
|
|
|
+ ResponseEntity<String> response = restTemplate.exchange(
|
|
|
+ resolveIntegrationBaseUrl(config) + CloudIntegrationConstants.PATH_SYNC_POLL_ACK,
|
|
|
+ HttpMethod.POST,
|
|
|
+ new HttpEntity<>(body, headers),
|
|
|
+ String.class);
|
|
|
+ return parseSyncResponse(response.getBody());
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("ACK云端失败", e);
|
|
|
+ return SyncResponse.builder().success(false).message(e.getMessage()).build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private HttpHeaders buildAuthHeaders(IssueCloudConfig config, String credential) {
|
|
|
+ return buildAuthHeaders(config.getTenantId(), credential, CloudEntitySupport.resolveAccessToken(config));
|
|
|
+ }
|
|
|
+
|
|
|
+ private HttpHeaders buildAuthHeaders(Integer tenantId, String credential, String token) {
|
|
|
+ HttpHeaders headers = new HttpHeaders();
|
|
|
+ headers.add(CloudIntegrationConstants.HEADER_TENANT_ID, String.valueOf(tenantId));
|
|
|
+ headers.add(CloudIntegrationConstants.HEADER_API_KEY, credential);
|
|
|
+ if (StringUtils.hasText(token)) {
|
|
|
+ headers.add("Authorization", "Bearer " + token);
|
|
|
+ }
|
|
|
+ return headers;
|
|
|
+ }
|
|
|
+
|
|
|
+ private CloudConnectionTestResult parseConnectionTestResult(String body, long costTime) {
|
|
|
+ if (!StringUtils.hasText(body)) {
|
|
|
+ log.warn("[连接测试-HTTP] 云端返回空响应");
|
|
|
+ return CloudConnectionTestResult.failure(true, "云端返回空响应");
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ JsonNode json = OBJECT_MAPPER.readTree(body);
|
|
|
+ JsonNode dataNode = json.has("data") ? json.get("data") : json;
|
|
|
+ if (dataNode == null || dataNode.isNull()) {
|
|
|
+ log.warn("[连接测试-HTTP] 云端返回空 data, rawBody={}", body);
|
|
|
+ return CloudConnectionTestResult.failure(true, "云端返回空数据");
|
|
|
+ }
|
|
|
+ CloudConnectionTestResult result = new CloudConnectionTestResult();
|
|
|
+ result.setSuccess(dataNode.path("success").asBoolean(false));
|
|
|
+ result.setMessage(resolveResponseMessage(json, dataNode));
|
|
|
+ result.setNetworkReachable(dataNode.has("networkReachable")
|
|
|
+ ? dataNode.get("networkReachable").asBoolean() : true);
|
|
|
+ result.setAuthValid(dataNode.has("authValid")
|
|
|
+ ? dataNode.get("authValid").asBoolean() : result.isSuccess());
|
|
|
+ result.setTenantExists(dataNode.has("tenantExists")
|
|
|
+ ? dataNode.get("tenantExists").asBoolean() : result.isSuccess());
|
|
|
+ result.setCostTime(dataNode.has("costTime") ? dataNode.get("costTime").asLong() : costTime);
|
|
|
+ if (!StringUtils.hasText(result.getMessage())) {
|
|
|
+ result.setMessage(result.isSuccess()
|
|
|
+ ? CloudIntegrationConstants.MSG_CONNECTION_SUCCESS
|
|
|
+ : "连接测试未通过");
|
|
|
+ log.warn("[连接测试-HTTP] 响应缺少 message,已使用默认文案, parsedSuccess={}, rawBody={}",
|
|
|
+ result.isSuccess(), body);
|
|
|
+ }
|
|
|
+ log.info("[连接测试-HTTP] 解析结果 success={}, authValid={}, tenantExists={}, message={}",
|
|
|
+ result.isSuccess(), result.isAuthValid(), result.isTenantExists(), result.getMessage());
|
|
|
+ return result;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("[连接测试-HTTP] 响应解析失败, rawBody={}", body, e);
|
|
|
+ return CloudConnectionTestResult.failure(true, "响应解析失败");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private String resolveResponseMessage(JsonNode root, JsonNode dataNode) {
|
|
|
+ String message = dataNode.path("message").asText(null);
|
|
|
+ if (StringUtils.hasText(message)) {
|
|
|
+ return message;
|
|
|
+ }
|
|
|
+ message = root.path("msg").asText(null);
|
|
|
+ if (StringUtils.hasText(message)) {
|
|
|
+ return message;
|
|
|
+ }
|
|
|
+ return dataNode.path("errorMessage").asText(null);
|
|
|
+ }
|
|
|
+
|
|
|
+ private String buildTestConnectionUrl(String cloudAddress) {
|
|
|
+ return normalizeCloudBaseUrl(cloudAddress) + URL_END;
|
|
|
+ }
|
|
|
+
|
|
|
+ private String resolveIntegrationBaseUrl(IssueCloudConfig config) {
|
|
|
+ if (config == null || !StringUtils.hasText(config.getCloudAddress())) {
|
|
|
+ return "";
|
|
|
+ }
|
|
|
+ return normalizeCloudBaseUrl(config.getCloudAddress()) + INTEGRATION_PATH;
|
|
|
+ }
|
|
|
+
|
|
|
+ private String normalizeCloudBaseUrl(String cloudAddress) {
|
|
|
+ String url = cloudAddress.trim();
|
|
|
+ if (url.endsWith("/")) {
|
|
|
+ url = url.substring(0, url.length() - 1);
|
|
|
+ }
|
|
|
+ if (url.endsWith(URL_END)) {
|
|
|
+ return url.substring(0, url.length() - URL_END.length());
|
|
|
+ }
|
|
|
+ if (url.endsWith(INTEGRATION_PATH)) {
|
|
|
+ return url.substring(0, url.length() - INTEGRATION_PATH.length());
|
|
|
+ }
|
|
|
+ if (url.endsWith(CloudIntegrationConstants.PATH_TEST_CONNECTION)) {
|
|
|
+ return url.substring(0, url.length() - CloudIntegrationConstants.PATH_TEST_CONNECTION.length());
|
|
|
+ }
|
|
|
+ return url;
|
|
|
+ }
|
|
|
+
|
|
|
+ private SyncResponse parseSyncResponse(String body) {
|
|
|
+ if (!StringUtils.hasText(body)) {
|
|
|
+ return SyncResponse.builder().success(false).message("云端返回空响应").build();
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ JsonNode json = OBJECT_MAPPER.readTree(body);
|
|
|
+ JsonNode dataNode = json.has("data") ? json.get("data") : json;
|
|
|
+ return OBJECT_MAPPER.treeToValue(dataNode, SyncResponse.class);
|
|
|
+ } catch (Exception e) {
|
|
|
+ return SyncResponse.builder().success(false).message("响应解析失败").build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|