集成中心代码提交

service-issue/service-issue-biz/src/main/java/com/usky/issue/IssueApplication.java

@@ -9,6 +9,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.core.env.Environment;
@@ -24,6 +25,7 @@ import java.net.UnknownHostException;
 
 //@EnableSwagger2
 @EnableCustomSwagger2
+@EnableScheduling
 @EnableFeignClients(basePackages = "com.usky")
 @MapperScan(value = "com.usky.issue.mapper")
 @ComponentScan("com.usky")

service-issue/service-issue-biz/src/main/java/com/usky/issue/controller/api/CloudIntegrationApiController.java

@@ -0,0 +1,163 @@
+package com.usky.issue.controller.api;
+
+import com.usky.common.core.bean.ApiResult;
+import com.usky.issue.service.CloudAuthService;
+import com.usky.issue.service.CloudConfigService;
+import com.usky.issue.service.CloudSyncReceiveService;
+import com.usky.issue.service.constant.CloudIntegrationConstants;
+import com.usky.issue.service.vo.CloudConnectionTestResponse;
+import com.usky.issue.service.vo.SyncPacket;
+import com.usky.issue.service.vo.SyncResponse;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.StringUtils;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * 云平台集成 API（供本地服务器调用：连通性测试 + 双向同步）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@RestController
+@RequestMapping("/cloud/integration")
+public class CloudIntegrationApiController {
+
+    @Autowired
+    private CloudConfigService cloudConfigService;
+    @Autowired
+    private CloudAuthService cloudAuthService;
+    @Autowired
+    private CloudSyncReceiveService cloudSyncReceiveService;
+
+    /**
+     * 网络层探测
+     */
+    @GetMapping("/ping")
+    public String ping() {
+        return "PONG";
+    }
+
+    /**
+     * 认证层 + 租户层检测
+     */
+    @GetMapping("/auth-check")
+    public ResponseEntity<String> authCheck(
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_API_KEY, required = false) String apiKey,
+            @RequestParam Integer tenantId) {
+        if (!cloudAuthService.validateToken(apiKey, tenantId)) {
+            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).body("AUTH_INVALID");
+        }
+        CloudConnectionTestResponse tenantResult = cloudConfigService.validateTenantConnection(tenantId);
+        if (!tenantResult.isSuccess()) {
+            return ResponseEntity.ok(CloudIntegrationConstants.AUTH_CHECK_TENANT_MISSING);
+        }
+        return ResponseEntity.ok(CloudIntegrationConstants.AUTH_CHECK_VALID);
+    }
+
+    /**
+     * 连通性测试入口（兼容旧调用方式）
+     */
+    @PostMapping("/testConnection")
+    public ApiResult<CloudConnectionTestResponse> testConnection(
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_TENANT_ID, required = false) String tenantIdHeader,
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_API_KEY, required = false) String apiKey) {
+        Integer tenantId = parseTenantId(tenantIdHeader);
+        if (tenantId != null && !cloudAuthService.validateToken(apiKey, tenantId)) {
+            CloudConnectionTestResponse response = new CloudConnectionTestResponse();
+            response.setSuccess(false);
+            response.setConnectionStatus(2);
+            response.setMessage(CloudIntegrationConstants.MSG_AUTH_INVALID);
+            response.setNetworkReachable(true);
+            response.setAuthValid(false);
+            response.setTenantExists(false);
+            return ApiResult.success(response);
+        }
+        CloudConnectionTestResponse response = cloudConfigService.validateTenantConnection(tenantId);
+        response.setNetworkReachable(true);
+        response.setAuthValid(tenantId == null || cloudAuthService.validateToken(apiKey, tenantId));
+        response.setTenantExists(response.isSuccess());
+        return ApiResult.success(response);
+    }
+
+    /**
+     * 接收本地推送数据（本地→云端）
+     */
+    @PostMapping("/sync/receive")
+    public SyncResponse receive(
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_API_KEY, required = false) String apiKey,
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_SIGN, required = false) String sign,
+            @RequestBody SyncPacket packet) {
+
+        if (!cloudAuthService.validate(apiKey, sign, packet)) {
+            return SyncResponse.builder().success(false).message("认证失败或请求已过期").build();
+        }
+        if (!cloudSyncReceiveService.validateTenant(packet.getTenantId())) {
+            return SyncResponse.builder().success(false).message("租户不存在或已停用").build();
+        }
+        try {
+            int count = cloudSyncReceiveService.batchUpsert(packet);
+            return SyncResponse.builder()
+                    .success(true)
+                    .message("接收成功")
+                    .acceptedCount(count)
+                    .confirmedVersion(packet.getBatchVersion())
+                    .build();
+        } catch (Exception e) {
+            return SyncResponse.builder().success(false).message("写入失败:" + e.getMessage()).build();
+        }
+    }
+
+    /**
+     * 下发数据轮询（云端→本地）
+     */
+    @GetMapping("/sync/poll")
+    public ApiResult<SyncPacket> poll(
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_API_KEY, required = false) String apiKey,
+            @RequestParam String tenantId,
+            @RequestParam String tableName,
+            @RequestParam(required = false) Long lastVersion) {
+
+        if (!cloudAuthService.validateToken(apiKey, Integer.valueOf(tenantId))) {
+            return ApiResult.success(null);
+        }
+        return ApiResult.success(cloudSyncReceiveService.pollDownQueue(tenantId, tableName, lastVersion));
+    }
+
+    /**
+     * 确认下发数据已消费
+     */
+    @PostMapping("/sync/poll/ack")
+    public SyncResponse ackPoll(
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_API_KEY, required = false) String apiKey,
+            @RequestHeader(value = CloudIntegrationConstants.HEADER_TENANT_ID, required = false) String tenantIdHeader,
+            @RequestBody List<Long> queueIds) {
+
+        Integer tenantId = parseTenantId(tenantIdHeader);
+        if (tenantId != null && !cloudAuthService.validateToken(apiKey, tenantId)) {
+            return SyncResponse.builder().success(false).message("认证失败").build();
+        }
+        cloudSyncReceiveService.ackDownQueue(queueIds);
+        return SyncResponse.builder().success(true).message("确认成功").build();
+    }
+
+    private Integer parseTenantId(String tenantIdHeader) {
+        if (!StringUtils.hasText(tenantIdHeader)) {
+            return null;
+        }
+        try {
+            return Integer.valueOf(tenantIdHeader.trim());
+        } catch (NumberFormatException ex) {
+            return null;
+        }
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/controller/web/CloudConfigController.java

@@ -4,6 +4,7 @@ import com.usky.common.core.bean.ApiResult;
 import com.usky.common.security.utils.SecurityUtils;
 import com.usky.issue.service.vo.CloudConfigResponse;
 import com.usky.issue.service.vo.CloudConfigSaveRequest;
+import com.usky.issue.service.vo.CloudConnectionTestRequest;
 import com.usky.issue.service.vo.CloudConnectionTestResponse;
 import com.usky.issue.service.vo.CloudDisableResponse;
 import com.usky.issue.service.CloudConfigService;
@@ -53,10 +54,21 @@ public class CloudConfigController {
     }
 
 
+    /**
+     * 分层连通性测试：网络层 → 认证层 → 租户层
+     */
     @PostMapping("/config/testConnection")
-    public ApiResult<CloudConnectionTestResponse> testConnection(HttpServletRequest httpRequest) {
+    public ApiResult<CloudConnectionTestResponse> testConnection(
+            @Valid @RequestBody(required = false) CloudConnectionTestRequest request,
+            HttpServletRequest httpRequest) {
+        if (request == null) {
+            request = new CloudConnectionTestRequest();
+        }
+        if (request.getTenantId() == null) {
+            request.setTenantId(SecurityUtils.getTenantId());
+        }
         return ApiResult.success(cloudConfigService.testConnection(
-                SecurityUtils.getUsername(), httpRequest.getRemoteAddr()));
+                request, SecurityUtils.getUsername(), httpRequest.getRemoteAddr()));
     }
 
     /**

service-issue/service-issue-biz/src/main/java/com/usky/issue/domain/IssueSyncCursor.java

@@ -0,0 +1,35 @@
+package com.usky.issue.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 同步游标
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("issue_sync_cursor")
+public class IssueSyncCursor extends CloudAuditEntity {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Long id;
+
+    private Integer tenantId;
+
+    private String tableName;
+
+    /** UP/DOWN */
+    private String direction;
+
+    private Long lastSyncVersion;
+
+    private LocalDateTime lastSyncTime;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/domain/IssueSyncData.java

@@ -0,0 +1,40 @@
+package com.usky.issue.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 本地同步业务数据
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("issue_sync_data")
+public class IssueSyncData extends CloudAuditEntity {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Long id;
+
+    private Integer tenantId;
+
+    private String tableName;
+
+    private String dataKey;
+
+    private String payload;
+
+    private Long syncVersion;
+
+    /** 0:未同步 1:已推送 2:推送失败 */
+    private Integer syncStatus;
+
+    /** LOCAL/CLOUD */
+    private String syncDirection;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/domain/IssueSyncDownQueue.java

@@ -0,0 +1,36 @@
+package com.usky.issue.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 云端下发队列（云端→本地）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("issue_sync_down_queue")
+public class IssueSyncDownQueue extends CloudAuditEntity {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Long id;
+
+    private Integer tenantId;
+
+    private String tableName;
+
+    /** INSERT/UPDATE/DELETE */
+    private String operation;
+
+    private String payload;
+
+    private Long syncVersion;
+
+    /** 0:待拉取 1:已确认 2:失败 */
+    private Integer status;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/mapper/IssueSyncCursorMapper.java

@@ -0,0 +1,17 @@
+package com.usky.issue.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.usky.common.mybatis.core.CrudMapper;
+import com.usky.issue.domain.IssueSyncCursor;
+import org.springframework.stereotype.Repository;
+
+/**
+ * 同步游标 Mapper
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@DS("usky-cloud")
+@Repository
+public interface IssueSyncCursorMapper extends CrudMapper<IssueSyncCursor> {
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/mapper/IssueSyncDataMapper.java

@@ -0,0 +1,17 @@
+package com.usky.issue.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.usky.common.mybatis.core.CrudMapper;
+import com.usky.issue.domain.IssueSyncData;
+import org.springframework.stereotype.Repository;
+
+/**
+ * 同步业务数据 Mapper
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@DS("usky-cloud")
+@Repository
+public interface IssueSyncDataMapper extends CrudMapper<IssueSyncData> {
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/mapper/IssueSyncDownQueueMapper.java

@@ -0,0 +1,17 @@
+package com.usky.issue.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.usky.common.mybatis.core.CrudMapper;
+import com.usky.issue.domain.IssueSyncDownQueue;
+import org.springframework.stereotype.Repository;
+
+/**
+ * 云端下发队列 Mapper
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@DS("usky-cloud")
+@Repository
+public interface IssueSyncDownQueueMapper extends CrudMapper<IssueSyncDownQueue> {
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/CloudAuthService.java

@@ -0,0 +1,16 @@
+package com.usky.issue.service;
+
+import com.usky.issue.service.vo.SyncPacket;
+
+/**
+ * 云平台认证服务
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public interface CloudAuthService {
+
+    boolean validateToken(String apiKey, Integer tenantId);
+
+    boolean validate(String apiKey, String sign, SyncPacket packet);
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/CloudConfigService.java

@@ -3,6 +3,7 @@ package com.usky.issue.service;
 import com.baomidou.dynamic.datasource.annotation.DS;
 import com.usky.issue.service.vo.CloudConfigResponse;
 import com.usky.issue.service.vo.CloudConfigSaveRequest;
+import com.usky.issue.service.vo.CloudConnectionTestRequest;
 import com.usky.issue.service.vo.CloudConnectionTestResponse;
 import com.usky.issue.service.vo.CloudDisableResponse;
 import com.usky.issue.domain.IssueCloudConfig;
@@ -20,7 +21,12 @@ public interface CloudConfigService {
 
     CloudConfigResponse saveConfig(CloudConfigSaveRequest request, String operator, String requestIp);
 
-    CloudConnectionTestResponse testConnection(String operator, String requestIp);
+    CloudConnectionTestResponse testConnection(CloudConnectionTestRequest request, String operator, String requestIp);
+
+    /**
+     * 云平台侧租户配置校验（供集成 API 调用，查询云库后返回结果，不再向外发起 HTTP）
+     */
+    CloudConnectionTestResponse validateTenantConnection(Integer tenantId);
 
     CloudDisableResponse disable(String operator, String requestIp);
 

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/CloudSyncReceiveService.java

@@ -0,0 +1,25 @@
+package com.usky.issue.service;
+
+import com.usky.issue.service.vo.SyncPacket;
+import com.usky.issue.service.vo.SyncResponse;
+
+import java.util.List;
+
+/**
+ * 云端同步接收服务（接收推送、下发轮询）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public interface CloudSyncReceiveService {
+
+    boolean validateTenant(String tenantId);
+
+    int batchUpsert(SyncPacket packet);
+
+    SyncPacket pollDownQueue(String tenantId, String tableName, Long lastVersion);
+
+    void ackDownQueue(List<Long> queueIds);
+
+    int countPendingDownQueue(Integer tenantId, String tableName);
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/LocalPullService.java

@@ -0,0 +1,17 @@
+package com.usky.issue.service;
+
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.service.vo.SyncResult;
+
+/**
+ * 本地拉取服务（云端→本地）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public interface LocalPullService {
+
+    SyncResult pullAndApply(String tableName, IssueCloudConfig config);
+
+    SyncResult pullByTenant(Integer tenantId, String tableName, IssueCloudConfig config);
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/LocalPushService.java

@@ -0,0 +1,19 @@
+package com.usky.issue.service;
+
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.service.vo.SyncResult;
+
+/**
+ * 本地推送服务（本地→云端）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public interface LocalPushService {
+
+    SyncResult pushAllPending(String tableName, IssueCloudConfig config);
+
+    SyncResult pushByTenant(Integer tenantId, String tableName, IssueCloudConfig config);
+
+    void resetCursor(Integer tenantId, String tableName);
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/LocalSyncAgent.java

@@ -0,0 +1,19 @@
+package com.usky.issue.service;
+
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.enums.SyncTypeEnum;
+import com.usky.issue.service.vo.SyncResult;
+
+/**
+ * 本地同步 Agent
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public interface LocalSyncAgent {
+
+    SyncResult executeSync(SyncTypeEnum syncType, IssueCloudConfig config, SyncDirection direction, boolean fullSync);
+
+    int countPending(SyncTypeEnum syncType, IssueCloudConfig config);
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/client/CloudPlatformClient.java

@@ -1,22 +1,35 @@
 package com.usky.issue.service.client;
 
-import com.usky.issue.service.constant.CloudIntegrationConstants;
+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.util.AesGcmCipher;
+import com.usky.issue.service.util.HmacSignUtil;
+import com.usky.issue.service.vo.CloudConnectionTestResult;
+import com.usky.issue.service.vo.SyncPacket;
+import com.usky.issue.service.vo.SyncResponse;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 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.ResourceAccessException;
 import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
 
 import java.time.Duration;
+import java.util.List;
 
 /**
- * 云平台 HTTP 客户端（测试连接与数据拉取模拟）
+ * 云平台 HTTP 客户端（连通性测试 + 双向同步）
  *
  * @author fyc
  * @date 2026-05-21
@@ -25,10 +38,12 @@ import java.time.Duration;
 @Component
 public class CloudPlatformClient {
 
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
     private final RestTemplate restTemplate;
     private final AesGcmCipher aesGcmCipher;
 
-    @Value("${cloud.integration.api-base-url:http://192.168.10.165:13200/offline-api/service-issue/cloud/config/testConnection}")
+    @Value("${cloud.integration.api-base-url:http://192.168.10.165:13200/offline-api/service-issue/cloud/integration/testConnection}")
     private String apiBaseUrl;
 
     public CloudPlatformClient(RestTemplateBuilder restTemplateBuilder, AesGcmCipher aesGcmCipher) {
@@ -40,124 +55,166 @@ public class CloudPlatformClient {
     }
 
     /**
-     * 测试云平台连接
+     * 分层连通性测试：网络层 → 认证层 → 租户层
      */
-    public boolean testConnection(IssueCloudConfig config) {
+    public CloudConnectionTestResult testConnection(IssueCloudConfig config) {
         if (config == null || config.getTenantId() == null || config.getCredentialKey() == null) {
             log.warn("云平台配置不完整");
-            return false;
+            return CloudConnectionTestResult.failure(false, "云平台配置不完整");
+        }
+
+        long start = System.currentTimeMillis();
+        String integrationBase = resolveIntegrationBaseUrl();
+
+        try {
+            String pingUrl = integrationBase + CloudIntegrationConstants.PATH_PING;
+            restTemplate.getForObject(pingUrl, String.class);
+        } catch (ResourceAccessException ex) {
+            log.error("云平台网络不可达: {}", ex.getMessage());
+            return CloudConnectionTestResult.networkFailure(CloudIntegrationConstants.MSG_NETWORK_UNREACHABLE);
+        } catch (Exception ex) {
+            log.error("云平台 Ping 失败: {}", ex.getMessage());
+            return CloudConnectionTestResult.networkFailure("网络层检测失败：" + ex.getMessage());
         }
 
         try {
             String credential = aesGcmCipher.decrypt(config.getCredentialKey());
-            HttpHeaders headers = new HttpHeaders();
-            headers.add("X-Tenant-Id", String.valueOf(config.getTenantId()));
-            headers.add("X-Api-Key", credential);
-            headers.add("Authorization", "Bearer " + config.getToken());
-            String url = apiBaseUrl;
-            log.info("测试云平台连接 tenantId={}, url={}", config.getTenantId(), url);
+            HttpHeaders headers = buildAuthHeaders(config, credential);
+            String authCheckUrl = UriComponentsBuilder
+                    .fromHttpUrl(integrationBase + CloudIntegrationConstants.PATH_AUTH_CHECK)
+                    .queryParam("tenantId", config.getTenantId())
+                    .toUriString();
+
+            ResponseEntity<String> authResp = restTemplate.exchange(
+                    authCheckUrl, HttpMethod.GET, new HttpEntity<>(headers), String.class);
+
+            long costTime = System.currentTimeMillis() - start;
+            String body = authResp.getBody();
+            if (CloudIntegrationConstants.AUTH_CHECK_VALID.equals(body)) {
+                return CloudConnectionTestResult.fullSuccess(costTime);
+            }
+            if (CloudIntegrationConstants.AUTH_CHECK_TENANT_MISSING.equals(body)) {
+                return CloudConnectionTestResult.tenantMissing(costTime);
+            }
+            return CloudConnectionTestResult.authFailure(costTime, "认证层检测未通过");
+        } catch (HttpClientErrorException.Unauthorized ex) {
+            return CloudConnectionTestResult.authFailure(System.currentTimeMillis() - start, "认证失败：Token无效");
+        } catch (HttpClientErrorException | HttpServerErrorException ex) {
+            log.error("云平台认证检测失败: {} {}", ex.getStatusCode(), ex.getResponseBodyAsString());
+            return CloudConnectionTestResult.authFailure(System.currentTimeMillis() - start,
+                    "云平台返回异常：" + ex.getStatusCode());
+        } catch (ResourceAccessException ex) {
+            return CloudConnectionTestResult.networkFailure(CloudIntegrationConstants.MSG_NETWORK_UNREACHABLE);
+        } catch (Exception ex) {
+            log.error("云平台连接测试失败: {}", ex.getMessage(), ex);
+            return CloudConnectionTestResult.failure(true, ex.getMessage());
+        }
+    }
 
-            ResponseEntity<String> response = restTemplate.exchange(
-                    url, HttpMethod.POST, new HttpEntity<>(headers), String.class);
+    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));
 
-            boolean is2xx = response.getStatusCode().is2xxSuccessful();
-            String body = response.getBody();
+            ResponseEntity<String> response = restTemplate.exchange(
+                    resolveIntegrationBaseUrl() + 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();
+        }
+    }
 
-            log.info("云平台连接测试响应: status={}, body={}", response.getStatusCode(), body);
+    public SyncPacket pollCloud(Integer tenantId, String tableName, Long lastVersion, IssueCloudConfig config) {
+        try {
+            String credential = aesGcmCipher.decrypt(config.getCredentialKey());
+            HttpHeaders headers = buildAuthHeaders(config, credential);
+            String url = UriComponentsBuilder
+                    .fromHttpUrl(resolveIntegrationBaseUrl() + CloudIntegrationConstants.PATH_SYNC_POLL)
+                    .queryParam("tenantId", tenantId)
+                    .queryParam("tableName", tableName)
+                    .queryParam("lastVersion", lastVersion == null ? 0L : lastVersion)
+                    .toUriString();
 
-            if (!is2xx) {
-                log.warn("云平台返回非2xx状态码: {}", response.getStatusCode());
-                return false;
+            ResponseEntity<String> response = restTemplate.exchange(
+                    url, HttpMethod.GET, new HttpEntity<>(headers), String.class);
+            if (!StringUtils.hasText(response.getBody())) {
+                return null;
             }
-
-            if (body == null || body.isEmpty()) {
-                log.warn("云平台返回空响应体");
-                return false;
+            JsonNode json = OBJECT_MAPPER.readTree(response.getBody());
+            JsonNode dataNode = json.has("data") ? json.get("data") : json;
+            if (dataNode == null || dataNode.isNull()) {
+                return null;
             }
+            return OBJECT_MAPPER.treeToValue(dataNode, SyncPacket.class);
+        } catch (Exception e) {
+            log.error("轮询云端失败", e);
+            return null;
+        }
+    }
 
-            try {
-                com.fasterxml.jackson.databind.JsonNode json =
-                        new com.fasterxml.jackson.databind.ObjectMapper().readTree(body);
-
-                // 1. 检查外层状态码
-                if (json.has("code")) {
-                    String codeStr = json.get("code").asText();
-                    if (!"0".equals(codeStr) && !"200".equals(codeStr)) {
-                        log.warn("云平台接口返回错误码: {}", codeStr);
-                        return false;
-                    }
-                }
-
-                // 2. 深度检查 data 字段
-                if (json.has("data")) {
-                    com.fasterxml.jackson.databind.JsonNode dataNode = json.get("data");
-                    
-                    // 如果 data 是对象（如连接测试结果）
-                    if (dataNode.isObject()) {
-                        // 检查 success 字段
-                        if (dataNode.has("success") && !dataNode.get("success").asBoolean(false)) {
-                            String msg = dataNode.has("message") ? dataNode.get("message").asText() : "业务失败";
-                            log.warn("云平台业务返回失败: {}", msg);
-                            return false;
-                        }
-                        
-                        // 检查 connectionStatus 字段 (2代表失败)
-                        if (dataNode.has("connectionStatus") && dataNode.get("connectionStatus").asInt(-1) == 2) {
-                            log.warn("云平台连接状态标识为失败");
-                            return false;
-                        }
-                    }
-                }
-            } catch (Exception e) {
-                log.error("解析云平台响应失败: {}", e.getMessage());
-                return false; // 解析失败视为连接异常
-            }
+    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);
 
-            log.info("云平台连接测试成功");
-            return true;
-
-        } catch (org.springframework.web.client.HttpClientErrorException ex) {
-            log.error("云平台连接失败 - HTTP客户端错误: {} {}", ex.getStatusCode(), ex.getResponseBodyAsString());
-            return false;
-        } catch (org.springframework.web.client.HttpServerErrorException ex) {
-            log.error("云平台连接失败 - HTTP服务端错误: {} {}", ex.getStatusCode(), ex.getResponseBodyAsString());
-            return false;
-        } catch (org.springframework.web.client.ResourceAccessException ex) {
-            log.error("云平台连接失败 - 网络异常: {}", ex.getMessage());
-            return false;
-        } catch (Exception ex) {
-            log.error("云平台连接测试失败: {}", ex.getMessage(), ex);
-            return false;
+            ResponseEntity<String> response = restTemplate.exchange(
+                    resolveIntegrationBaseUrl() + 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();
         }
     }
 
-    /**
-     * 模拟拉取待同步数据条数
-     */
-    public int fetchPendingCount(String syncType) {
-        switch (syncType) {
-            case "IOT_FACILITY":
-            case "IOT_MONITOR":
-                return 520;
-            case "BUILDING":
-                return 120;
-            case "ALARM":
-                return 15;
-            case "USER":
-                return 3;
-            case "TENANT":
-                return 1;
-            default:
-                return 10;
+    private HttpHeaders buildAuthHeaders(IssueCloudConfig config, String credential) {
+        HttpHeaders headers = new HttpHeaders();
+        headers.add(CloudIntegrationConstants.HEADER_TENANT_ID, String.valueOf(config.getTenantId()));
+        headers.add(CloudIntegrationConstants.HEADER_API_KEY, credential);
+        if (StringUtils.hasText(config.getToken())) {
+            headers.add("Authorization", "Bearer " + config.getToken());
         }
+        return headers;
     }
 
-    public java.util.List<String> fetchDataIds(String syncType, int batchSize) {
-        int count = Math.min(fetchPendingCount(syncType), batchSize);
-        java.util.List<String> ids = new java.util.ArrayList<>(count);
-        for (int i = 0; i < count; i++) {
-            ids.add(syncType + "-" + System.currentTimeMillis() + "-" + i);
+    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();
+        }
+    }
+
+    String resolveIntegrationBaseUrl() {
+        if (!StringUtils.hasText(apiBaseUrl)) {
+            return "";
+        }
+        String url = apiBaseUrl.trim();
+        if (url.endsWith(CloudIntegrationConstants.PATH_TEST_CONNECTION)) {
+            return url.substring(0, url.length() - CloudIntegrationConstants.PATH_TEST_CONNECTION.length());
+        }
+        if (url.endsWith("/cloud/integration")) {
+            return url;
+        }
+        if (url.endsWith("/")) {
+            return url.substring(0, url.length() - 1);
         }
-        return ids;
+        return url;
     }
 }

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/constant/CloudIntegrationConstants.java

@@ -19,6 +19,25 @@ public final class CloudIntegrationConstants {
 
     public static final int CONNECTION_TEST_TIMEOUT_MS = 10_000;
 
+    public static final String HEADER_TENANT_ID = "X-Tenant-Id";
+    public static final String HEADER_API_KEY = "X-Api-Key";
+    public static final String HEADER_SIGN = "X-Sign";
+
+    public static final String PATH_TEST_CONNECTION = "/testConnection";
+    public static final String PATH_PING = "/ping";
+    public static final String PATH_AUTH_CHECK = "/auth-check";
+    public static final String PATH_SYNC_RECEIVE = "/sync/receive";
+    public static final String PATH_SYNC_POLL = "/sync/poll";
+    public static final String PATH_SYNC_POLL_ACK = "/sync/poll/ack";
+
+    public static final String AUTH_CHECK_VALID = "VALID";
+    public static final String AUTH_CHECK_TENANT_MISSING = "TENANT_MISSING";
+
+    public static final String MSG_TENANT_CONFIG_MISSING = "租户配置缺失";
+    public static final String MSG_AUTH_INVALID = "认证失败：API Key无效";
+    public static final String MSG_CONNECTION_SUCCESS = "连接成功";
+    public static final String MSG_NETWORK_UNREACHABLE = "云平台网络不可达";
+
     public static final String OPERATION_SAVE_CONFIG = "SAVE_CONFIG";
     public static final String OPERATION_TEST_CONNECTION = "TEST_CONNECTION";
     public static final String OPERATION_DISABLE = "DISABLE_CONFIG";

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/enums/SyncDirection.java

@@ -0,0 +1,31 @@
+package com.usky.issue.service.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * 同步方向
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Getter
+@AllArgsConstructor
+public enum SyncDirection {
+
+    UP("UP", "本地→云端"),
+    DOWN("DOWN", "云端→本地"),
+    BOTH("BOTH", "双向");
+
+    private final String code;
+    private final String desc;
+
+    public static SyncDirection fromCode(String code) {
+        for (SyncDirection direction : values()) {
+            if (direction.code.equalsIgnoreCase(code)) {
+                return direction;
+            }
+        }
+        throw new IllegalArgumentException("未知同步方向: " + code);
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/CloudAuthServiceImpl.java

@@ -0,0 +1,91 @@
+package com.usky.issue.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.mapper.IssueCloudConfigMapper;
+import com.usky.issue.service.CloudAuthService;
+import com.usky.issue.service.util.AesGcmCipher;
+import com.usky.issue.service.util.HmacSignUtil;
+import com.usky.issue.service.vo.SyncPacket;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+/**
+ * 云平台认证服务实现
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Slf4j
+@Service
+public class CloudAuthServiceImpl implements CloudAuthService {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    @Autowired
+    private IssueCloudConfigMapper configMapper;
+    @Autowired
+    private AesGcmCipher aesGcmCipher;
+
+    @Override
+    public boolean validateToken(String apiKey, Integer tenantId) {
+        if (!StringUtils.hasText(apiKey) || tenantId == null) {
+            return false;
+        }
+        IssueCloudConfig config = findActiveConfig(tenantId);
+        if (config == null) {
+            return false;
+        }
+        return matchCredential(apiKey, config);
+    }
+
+    @Override
+    public boolean validate(String apiKey, String sign, SyncPacket packet) {
+        if (!validateToken(apiKey, parseTenantId(packet))) {
+            return false;
+        }
+        if (!StringUtils.hasText(sign) || packet == null) {
+            return false;
+        }
+        try {
+            String expected = HmacSignUtil.sign(OBJECT_MAPPER.writeValueAsString(packet), apiKey);
+            return expected.equals(sign);
+        } catch (Exception e) {
+            log.warn("签名校验失败: {}", e.getMessage());
+            return false;
+        }
+    }
+
+    private Integer parseTenantId(SyncPacket packet) {
+        if (packet == null || !StringUtils.hasText(packet.getTenantId())) {
+            return null;
+        }
+        try {
+            return Integer.valueOf(packet.getTenantId());
+        } catch (NumberFormatException ex) {
+            return null;
+        }
+    }
+
+    private IssueCloudConfig findActiveConfig(Integer tenantId) {
+        return configMapper.selectOne(new QueryWrapper<IssueCloudConfig>()
+                .eq("deleted", 0)
+                .eq("tenant_id", tenantId)
+                .eq("status", 1)
+                .orderByDesc("id")
+                .last("LIMIT 1"));
+    }
+
+    private boolean matchCredential(String apiKey, IssueCloudConfig config) {
+        try {
+            String decrypted = aesGcmCipher.decrypt(config.getCredentialKey());
+            return apiKey.equals(decrypted);
+        } catch (Exception e) {
+            log.warn("凭证解密失败 tenantId={}", config.getTenantId());
+            return false;
+        }
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/CloudConfigServiceImpl.java

@@ -23,7 +23,9 @@ import com.usky.issue.service.util.AesGcmCipher;
 import com.usky.issue.service.util.CredentialMaskUtil;
 import com.usky.issue.service.vo.CloudConfigResponse;
 import com.usky.issue.service.vo.CloudConfigSaveRequest;
+import com.usky.issue.service.vo.CloudConnectionTestRequest;
 import com.usky.issue.service.vo.CloudConnectionTestResponse;
+import com.usky.issue.service.vo.CloudConnectionTestResult;
 import com.usky.issue.service.vo.CloudDisableResponse;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -114,20 +116,72 @@ public class CloudConfigServiceImpl implements CloudConfigService {
 
     @Override
     @Transactional(rollbackFor = Exception.class)
-    public CloudConnectionTestResponse testConnection(String operator, String requestIp) {
-        IssueCloudConfig config = requireActiveConfig();
+    public CloudConnectionTestResponse testConnection(CloudConnectionTestRequest request, String operator, String requestIp) {
+        Integer tenantId = request.getTenantId();
+        Integer loginTenantId = SecurityUtils.getTenantId();
+        if (loginTenantId != null && !tenantId.equals(loginTenantId)) {
+            throw new BusinessException("输入租户ID与登录租户不一致，请确认后重试！");
+        }
+
+        IssueCloudConfig config = findConfigByTenantId(tenantId);
+        if (config == null) {
+            throw new BusinessException("云平台配置不存在，请先保存配置");
+        }
+        if (config.getStatus() != null && config.getStatus() == 0) {
+            throw new BusinessException("云平台集成已禁用");
+        }
+
         config.setToken(SecurityUtils.getToken());
-        boolean ok = cloudPlatformClient.testConnection(config);
-        config.setConnectionStatus(ok ? 1 : 2);
+        CloudConnectionTestResult result = cloudPlatformClient.testConnection(config);
+
+        config.setConnectionStatus(result.isSuccess() ? 1 : 2);
         config.setLastTestTime(LocalDateTime.now());
         CloudEntitySupport.fillOnUpdate(config);
         configMapper.updateById(config);
         operationLogService.log(config.getId(), CloudIntegrationConstants.OPERATION_TEST_CONNECTION,
-                "连接测试" + (ok ? "成功" : "失败"), operator, requestIp);
+                "连接测试" + (result.isSuccess() ? "成功" : "失败") + "：" + result.getMessage(), operator, requestIp);
+        return toConnectionTestResponse(result);
+    }
+
+    @Override
+    public CloudConnectionTestResponse validateTenantConnection(Integer tenantId) {
         CloudConnectionTestResponse response = new CloudConnectionTestResponse();
-        response.setSuccess(ok);
-        response.setConnectionStatus(config.getConnectionStatus());
-        response.setMessage(ok ? "连接成功" : "连接失败");
+        if (tenantId == null) {
+            response.setSuccess(false);
+            response.setConnectionStatus(2);
+            response.setMessage("租户ID不能为空");
+            response.setNetworkReachable(true);
+            response.setAuthValid(false);
+            response.setTenantExists(false);
+            return response;
+        }
+
+        IssueCloudConfig config = findConfigByTenantId(tenantId);
+        if (config == null) {
+            response.setSuccess(false);
+            response.setConnectionStatus(2);
+            response.setMessage(CloudIntegrationConstants.MSG_TENANT_CONFIG_MISSING);
+            response.setNetworkReachable(true);
+            response.setAuthValid(true);
+            response.setTenantExists(false);
+            return response;
+        }
+        if (config.getStatus() != null && config.getStatus() == 0) {
+            response.setSuccess(false);
+            response.setConnectionStatus(2);
+            response.setMessage("租户配置已禁用");
+            response.setNetworkReachable(true);
+            response.setAuthValid(true);
+            response.setTenantExists(false);
+            return response;
+        }
+
+        response.setSuccess(true);
+        response.setConnectionStatus(1);
+        response.setMessage(CloudIntegrationConstants.MSG_CONNECTION_SUCCESS);
+        response.setNetworkReachable(true);
+        response.setAuthValid(true);
+        response.setTenantExists(true);
         return response;
     }
 
@@ -160,13 +214,32 @@ public class CloudConfigServiceImpl implements CloudConfigService {
     }
 
     private IssueCloudConfig findActiveConfig() {
+        return findConfigByTenantId(SecurityUtils.getTenantId());
+    }
+
+    private IssueCloudConfig findConfigByTenantId(Integer tenantId) {
+        if (tenantId == null) {
+            return null;
+        }
         return configMapper.selectOne(new QueryWrapper<IssueCloudConfig>()
                 .eq("deleted", 0)
-                .eq("tenant_id", SecurityUtils.getTenantId())
+                .eq("tenant_id", tenantId)
                 .orderByDesc("id")
                 .last("LIMIT 1"));
     }
 
+    private CloudConnectionTestResponse toConnectionTestResponse(CloudConnectionTestResult result) {
+        CloudConnectionTestResponse response = new CloudConnectionTestResponse();
+        response.setSuccess(result.isSuccess());
+        response.setConnectionStatus(result.isSuccess() ? 1 : 2);
+        response.setMessage(result.getMessage());
+        response.setNetworkReachable(result.isNetworkReachable());
+        response.setAuthValid(result.isAuthValid());
+        response.setTenantExists(result.isTenantExists());
+        response.setCostTime(result.getCostTime());
+        return response;
+    }
+
     private int cancelRunningTasks(Long configId) {
         List<IssueSyncTask> running = syncTaskMapper.selectList(new QueryWrapper<IssueSyncTask>()
                 .eq("config_id", configId)

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/CloudSyncReceiveServiceImpl.java

@@ -0,0 +1,162 @@
+package com.usky.issue.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.usky.issue.domain.IssueSyncData;
+import com.usky.issue.domain.IssueSyncDownQueue;
+import com.usky.issue.mapper.IssueSyncDataMapper;
+import com.usky.issue.mapper.IssueSyncDownQueueMapper;
+import com.usky.issue.service.CloudConfigService;
+import com.usky.issue.service.CloudSyncReceiveService;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.support.CloudEntitySupport;
+import com.usky.issue.service.vo.SyncPacket;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 云端同步接收服务实现
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Slf4j
+@Service
+public class CloudSyncReceiveServiceImpl implements CloudSyncReceiveService {
+
+    private static final int POLL_BATCH_SIZE = 100;
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    @Autowired
+    private IssueSyncDataMapper syncDataMapper;
+    @Autowired
+    private IssueSyncDownQueueMapper downQueueMapper;
+    @Autowired
+    private CloudConfigService cloudConfigService;
+
+    @Override
+    public boolean validateTenant(String tenantId) {
+        if (!StringUtils.hasText(tenantId)) {
+            return false;
+        }
+        return cloudConfigService.validateTenantConnection(Integer.valueOf(tenantId)).isSuccess();
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public int batchUpsert(SyncPacket packet) {
+        if (packet == null || packet.getDataList() == null || packet.getDataList().isEmpty()) {
+            return 0;
+        }
+        int count = 0;
+        for (Object item : packet.getDataList()) {
+            try {
+                IssueSyncData incoming = convertToSyncData(item, packet.getTableName());
+                if (incoming == null) {
+                    continue;
+                }
+                IssueSyncData exist = syncDataMapper.selectOne(new LambdaQueryWrapper<IssueSyncData>()
+                        .eq(IssueSyncData::getTenantId, incoming.getTenantId())
+                        .eq(IssueSyncData::getTableName, incoming.getTableName())
+                        .eq(IssueSyncData::getDataKey, incoming.getDataKey())
+                        .eq(IssueSyncData::getDeleted, 0));
+                if (exist == null) {
+                    incoming.setSyncStatus(1);
+                    incoming.setSyncDirection("LOCAL");
+                    CloudEntitySupport.fillOnInsert(incoming);
+                    syncDataMapper.insert(incoming);
+                    count++;
+                } else if (incoming.getSyncVersion() >= exist.getSyncVersion()) {
+                    incoming.setId(exist.getId());
+                    incoming.setSyncStatus(1);
+                    incoming.setSyncDirection("LOCAL");
+                    CloudEntitySupport.fillOnUpdate(incoming);
+                    syncDataMapper.updateById(incoming);
+                    count++;
+                }
+            } catch (Exception e) {
+                log.error("云端写入同步数据失败", e);
+            }
+        }
+        return count;
+    }
+
+    @Override
+    public SyncPacket pollDownQueue(String tenantId, String tableName, Long lastVersion) {
+        if (lastVersion == null) {
+            lastVersion = 0L;
+        }
+        Integer tenant = Integer.valueOf(tenantId);
+        List<IssueSyncDownQueue> queueList = downQueueMapper.selectList(new LambdaQueryWrapper<IssueSyncDownQueue>()
+                .eq(IssueSyncDownQueue::getTenantId, tenant)
+                .eq(IssueSyncDownQueue::getTableName, tableName)
+                .eq(IssueSyncDownQueue::getStatus, 0)
+                .gt(IssueSyncDownQueue::getSyncVersion, lastVersion)
+                .eq(IssueSyncDownQueue::getDeleted, 0)
+                .orderByAsc(IssueSyncDownQueue::getSyncVersion)
+                .last("LIMIT " + POLL_BATCH_SIZE));
+
+        if (queueList.isEmpty()) {
+            return null;
+        }
+
+        Long maxVersion = queueList.stream()
+                .mapToLong(IssueSyncDownQueue::getSyncVersion)
+                .max()
+                .orElse(lastVersion);
+
+        return SyncPacket.builder()
+                .tenantId(tenantId)
+                .tableName(tableName)
+                .direction(SyncDirection.DOWN)
+                .dataList(Collections.unmodifiableList(queueList))
+                .batchVersion(maxVersion)
+                .hasMore(queueList.size() >= POLL_BATCH_SIZE)
+                .build();
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public void ackDownQueue(List<Long> queueIds) {
+        if (queueIds == null || queueIds.isEmpty()) {
+            return;
+        }
+        downQueueMapper.update(null, new LambdaUpdateWrapper<IssueSyncDownQueue>()
+                .in(IssueSyncDownQueue::getId, queueIds)
+                .set(IssueSyncDownQueue::getStatus, 1));
+    }
+
+    @Override
+    public int countPendingDownQueue(Integer tenantId, String tableName) {
+        Long count = downQueueMapper.selectCount(new LambdaQueryWrapper<IssueSyncDownQueue>()
+                .eq(IssueSyncDownQueue::getTenantId, tenantId)
+                .eq(IssueSyncDownQueue::getTableName, tableName)
+                .eq(IssueSyncDownQueue::getStatus, 0)
+                .eq(IssueSyncDownQueue::getDeleted, 0));
+        return count == null ? 0 : count.intValue();
+    }
+
+    @SuppressWarnings("unchecked")
+    private IssueSyncData convertToSyncData(Object item, String tableName) throws Exception {
+        if (item instanceof IssueSyncData) {
+            return (IssueSyncData) item;
+        }
+        Map<String, Object> map = OBJECT_MAPPER.convertValue(item, Map.class);
+        IssueSyncData data = new IssueSyncData();
+        data.setTenantId(Integer.valueOf(String.valueOf(map.get("tenantId"))));
+        data.setTableName(tableName);
+        data.setDataKey(String.valueOf(map.get("dataKey")));
+        data.setPayload(map.containsKey("payload") ? String.valueOf(map.get("payload")) : OBJECT_MAPPER.writeValueAsString(map));
+        Object version = map.get("syncVersion");
+        data.setSyncVersion(version == null ? System.currentTimeMillis() : Long.valueOf(String.valueOf(version)));
+        return data;
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/CloudSyncTaskRunnerImpl.java

@@ -1,26 +1,28 @@
 package com.usky.issue.service.impl;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
-import com.usky.issue.service.client.CloudPlatformClient;
-import com.usky.issue.service.constant.CloudIntegrationConstants;
 import com.usky.issue.domain.IssueSyncDetail;
 import com.usky.issue.domain.IssueSyncStatus;
 import com.usky.issue.domain.IssueSyncTask;
-import com.usky.issue.service.enums.SyncTypeEnum;
-import com.usky.issue.service.enums.TaskStatusEnum;
 import com.usky.issue.mapper.IssueSyncDetailMapper;
 import com.usky.issue.mapper.IssueSyncStatusMapper;
 import com.usky.issue.mapper.IssueSyncTaskMapper;
 import com.usky.issue.service.CloudSyncTaskRunner;
+import com.usky.issue.service.LocalSyncAgent;
+import com.usky.issue.service.constant.CloudIntegrationConstants;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.enums.SyncTypeEnum;
+import com.usky.issue.service.enums.TaskStatusEnum;
 import com.usky.issue.service.support.CloudEntitySupport;
 import com.usky.common.redis.core.RedisHelper;
+import com.usky.issue.service.vo.SyncResult;
 import org.springframework.beans.factory.annotation.Autowired;
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.mapper.IssueCloudConfigMapper;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
 
 /**
  * 同步任务执行器实现
@@ -38,14 +40,17 @@ public class CloudSyncTaskRunnerImpl implements CloudSyncTaskRunner {
     @Autowired
     private IssueSyncStatusMapper syncStatusMapper;
     @Autowired
-    private CloudPlatformClient cloudPlatformClient;
+    private IssueCloudConfigMapper configMapper;
+    @Autowired
+    private LocalSyncAgent localSyncAgent;
     @Autowired
     private RedisHelper redisHelper;
 
     @Override
     @Transactional(rollbackFor = Exception.class)
     public IssueSyncTask createTask(Long configId, SyncTypeEnum type, String triggerMode) {
-        int pending = cloudPlatformClient.fetchPendingCount(type.getCode());
+        IssueCloudConfig config = configMapper.selectById(configId);
+        int pending = localSyncAgent.countPending(type, config);
         IssueSyncTask task = new IssueSyncTask();
         task.setConfigId(configId);
         task.setSyncType(type.getCode());
@@ -75,59 +80,51 @@ public class CloudSyncTaskRunnerImpl implements CloudSyncTaskRunner {
         CloudEntitySupport.fillOnUpdate(task);
         syncTaskMapper.updateById(task);
 
-        int batchSize = type.isAsync() ? 100 : task.getTotalCount();
-        List<String> dataIds = cloudPlatformClient.fetchDataIds(type.getCode(), batchSize);
-        int success = 0;
-        int failure = 0;
-        for (String dataId : dataIds) {
-            IssueSyncTask current = syncTaskMapper.selectById(taskId);
-            if (current != null && TaskStatusEnum.CANCELLED.getCode().equals(current.getTaskStatus())) {
-                break;
-            }
-            IssueSyncDetail existing = syncDetailMapper.selectOne(new LambdaQueryWrapper<IssueSyncDetail>()
-                    .eq(IssueSyncDetail::getTaskId, taskId)
-                    .eq(IssueSyncDetail::getDataId, dataId)
-                    .eq(IssueSyncDetail::getDeleted, 0));
-            if (existing != null) {
-                continue;
-            }
-            boolean itemOk = !dataId.contains("-fail-");
-            IssueSyncDetail detail = new IssueSyncDetail();
-            detail.setTaskId(taskId);
-            detail.setDataId(dataId);
-            detail.setRetryCount(0);
-            detail.setDetailStatus(itemOk ? TaskStatusEnum.SUCCESS.getCode() : TaskStatusEnum.FAILED.getCode());
-            if (!itemOk) {
-                detail.setErrorMessage("模拟同步失败");
-                failure++;
-            } else {
-                success++;
-            }
-            CloudEntitySupport.fillOnInsert(detail);
-            syncDetailMapper.insert(detail);
-            task.setProcessedCount(task.getProcessedCount() + 1);
-            task.setSuccessCount(success);
-            task.setFailureCount(failure);
-            CloudEntitySupport.fillOnUpdate(task);
-            syncTaskMapper.updateById(task);
-        }
+        IssueCloudConfig config = configMapper.selectById(task.getConfigId());
+        SyncResult syncResult = localSyncAgent.executeSync(type, config, SyncDirection.BOTH, false);
 
-        task = syncTaskMapper.selectById(taskId);
-        if (TaskStatusEnum.CANCELLED.getCode().equals(task.getTaskStatus())) {
+        IssueSyncTask current = syncTaskMapper.selectById(taskId);
+        if (current != null && TaskStatusEnum.CANCELLED.getCode().equals(current.getTaskStatus())) {
             return;
         }
-        task.setTaskStatus(failure > 0 && success == 0
+
+        task.setTotalCount(Math.max(task.getTotalCount(), syncResult.getTotalCount()));
+        task.setProcessedCount(syncResult.getTotalCount());
+        task.setSuccessCount(syncResult.getSuccessCount());
+        task.setFailureCount(syncResult.getFailureCount());
+        task.setTaskStatus(syncResult.getFailureCount() > 0 && syncResult.getSuccessCount() == 0
                 ? TaskStatusEnum.FAILED.getCode() : TaskStatusEnum.SUCCESS.getCode());
-        if (failure > 0) {
-            task.setErrorSummary("部分失败: " + failure + " 条");
+        if (syncResult.getFailureCount() > 0) {
+            task.setErrorSummary(syncResult.getErrorMessage() != null
+                    ? syncResult.getErrorMessage()
+                    : "部分失败: " + syncResult.getFailureCount() + " 条");
         }
         task.setFinishTime(LocalDateTime.now());
         CloudEntitySupport.fillOnUpdate(task);
         syncTaskMapper.updateById(task);
+
+        recordSyncDetails(taskId, syncResult);
         refreshSyncStatus(task.getConfigId(), type, task);
         redisHelper.delete(CloudIntegrationConstants.CACHE_STATUS_PREFIX + task.getConfigId());
     }
 
+    private void recordSyncDetails(Long taskId, SyncResult syncResult) {
+        if (syncResult.getTotalCount() <= 0) {
+            return;
+        }
+        IssueSyncDetail detail = new IssueSyncDetail();
+        detail.setTaskId(taskId);
+        detail.setDataId("batch-" + taskId);
+        detail.setRetryCount(0);
+        detail.setDetailStatus(syncResult.getFailureCount() > 0
+                ? TaskStatusEnum.FAILED.getCode() : TaskStatusEnum.SUCCESS.getCode());
+        if (syncResult.getFailureCount() > 0) {
+            detail.setErrorMessage(syncResult.getErrorMessage());
+        }
+        CloudEntitySupport.fillOnInsert(detail);
+        syncDetailMapper.insert(detail);
+    }
+
     private void refreshSyncStatus(Long configId, SyncTypeEnum type, IssueSyncTask task) {
         IssueSyncStatus status = syncStatusMapper.selectOne(new LambdaQueryWrapper<IssueSyncStatus>()
                 .eq(IssueSyncStatus::getConfigId, configId)

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/LocalPullServiceImpl.java

@@ -0,0 +1,185 @@
+package com.usky.issue.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.domain.IssueSyncCursor;
+import com.usky.issue.domain.IssueSyncData;
+import com.usky.issue.domain.IssueSyncDownQueue;
+import com.usky.issue.mapper.IssueSyncCursorMapper;
+import com.usky.issue.mapper.IssueSyncDataMapper;
+import com.usky.issue.service.LocalPullService;
+import com.usky.issue.service.client.CloudPlatformClient;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.support.CloudEntitySupport;
+import com.usky.issue.service.vo.SyncPacket;
+import com.usky.issue.service.vo.SyncResponse;
+import com.usky.issue.service.vo.SyncResult;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 本地拉取服务实现
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Slf4j
+@Service
+public class LocalPullServiceImpl implements LocalPullService {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    @Autowired
+    private IssueSyncDataMapper syncDataMapper;
+    @Autowired
+    private IssueSyncCursorMapper cursorMapper;
+    @Autowired
+    private CloudPlatformClient cloudPlatformClient;
+
+    @Override
+    public SyncResult pullAndApply(String tableName, IssueCloudConfig config) {
+        SyncResult result = SyncResult.empty();
+        List<IssueSyncCursor> cursors = cursorMapper.selectList(new LambdaQueryWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getDirection, SyncDirection.DOWN.getCode())
+                .eq(IssueSyncCursor::getDeleted, 0));
+        if (cursors.isEmpty() && config.getTenantId() != null) {
+            result.merge(pullByTenant(config.getTenantId(), tableName, config));
+            return result;
+        }
+        for (IssueSyncCursor cursor : cursors) {
+            try {
+                result.merge(pullByTenant(cursor.getTenantId(), tableName, config));
+            } catch (Exception e) {
+                log.error("拉取租户{}数据失败", cursor.getTenantId(), e);
+                result.setFailureCount(result.getFailureCount() + 1);
+                result.setErrorMessage(e.getMessage());
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public SyncResult pullByTenant(Integer tenantId, String tableName, IssueCloudConfig config) {
+        SyncResult result = SyncResult.empty();
+        IssueSyncCursor cursor = getOrInitCursor(tenantId, tableName, SyncDirection.DOWN.getCode());
+        Long lastVersion = cursor.getLastSyncVersion() == null ? 0L : cursor.getLastSyncVersion();
+
+        while (true) {
+            SyncPacket packet = cloudPlatformClient.pollCloud(tenantId, tableName, lastVersion, config);
+            if (packet == null || packet.getDataList() == null || packet.getDataList().isEmpty()) {
+                break;
+            }
+
+            List<IssueSyncDownQueue> queueList = convertQueueList(packet.getDataList());
+            List<Long> ackIds = new ArrayList<>();
+            int success = 0;
+            int failure = 0;
+
+            for (IssueSyncDownQueue queue : queueList) {
+                try {
+                    applyToLocal(queue);
+                    ackIds.add(queue.getId());
+                    success++;
+                } catch (Exception e) {
+                    log.error("应用云端数据到本地失败, queueId={}", queue.getId(), e);
+                    failure++;
+                }
+            }
+
+            result.setTotalCount(result.getTotalCount() + queueList.size());
+            result.setSuccessCount(result.getSuccessCount() + success);
+            result.setFailureCount(result.getFailureCount() + failure);
+
+            lastVersion = packet.getBatchVersion();
+            updateCursor(tenantId, tableName, SyncDirection.DOWN.getCode(), lastVersion);
+            ackCloud(ackIds, config);
+
+            if (Boolean.FALSE.equals(packet.getHasMore())) {
+                break;
+            }
+        }
+        return result;
+    }
+
+    private void applyToLocal(IssueSyncDownQueue queue) throws Exception {
+        IssueSyncData data = OBJECT_MAPPER.readValue(queue.getPayload(), IssueSyncData.class);
+        if (data.getDataKey() == null) {
+            data.setDataKey(queue.getTableName() + "-" + queue.getId());
+        }
+        data.setTenantId(queue.getTenantId());
+        data.setTableName(queue.getTableName());
+        data.setSyncVersion(queue.getSyncVersion());
+        data.setSyncDirection("CLOUD");
+        data.setSyncStatus(1);
+
+        IssueSyncData exist = syncDataMapper.selectOne(new LambdaQueryWrapper<IssueSyncData>()
+                .eq(IssueSyncData::getTenantId, data.getTenantId())
+                .eq(IssueSyncData::getTableName, data.getTableName())
+                .eq(IssueSyncData::getDataKey, data.getDataKey())
+                .eq(IssueSyncData::getDeleted, 0));
+        if (exist == null) {
+            CloudEntitySupport.fillOnInsert(data);
+            syncDataMapper.insert(data);
+        } else if (queue.getSyncVersion() >= exist.getSyncVersion()) {
+            data.setId(exist.getId());
+            CloudEntitySupport.fillOnUpdate(data);
+            syncDataMapper.updateById(data);
+        }
+    }
+
+    private void ackCloud(List<Long> queueIds, IssueCloudConfig config) {
+        if (queueIds == null || queueIds.isEmpty()) {
+            return;
+        }
+        SyncResponse response = cloudPlatformClient.ackPoll(queueIds, config);
+        if (response == null || !response.isSuccess()) {
+            log.warn("ACK云端失败 queueIds={}", queueIds);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private List<IssueSyncDownQueue> convertQueueList(List<?> dataList) {
+        return dataList.stream().map(item -> {
+            if (item instanceof IssueSyncDownQueue) {
+                return (IssueSyncDownQueue) item;
+            }
+            return OBJECT_MAPPER.convertValue(item, IssueSyncDownQueue.class);
+        }).collect(Collectors.toList());
+    }
+
+    private IssueSyncCursor getOrInitCursor(Integer tenantId, String tableName, String direction) {
+        IssueSyncCursor cursor = cursorMapper.selectOne(new LambdaQueryWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getTenantId, tenantId)
+                .eq(IssueSyncCursor::getTableName, tableName)
+                .eq(IssueSyncCursor::getDirection, direction)
+                .eq(IssueSyncCursor::getDeleted, 0));
+        if (cursor == null) {
+            cursor = new IssueSyncCursor();
+            cursor.setTenantId(tenantId);
+            cursor.setTableName(tableName);
+            cursor.setDirection(direction);
+            cursor.setLastSyncVersion(0L);
+            cursor.setLastSyncTime(LocalDateTime.now());
+            CloudEntitySupport.fillOnInsert(cursor);
+            cursorMapper.insert(cursor);
+        }
+        return cursor;
+    }
+
+    private void updateCursor(Integer tenantId, String tableName, String direction, Long version) {
+        cursorMapper.update(null, new LambdaUpdateWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getTenantId, tenantId)
+                .eq(IssueSyncCursor::getTableName, tableName)
+                .eq(IssueSyncCursor::getDirection, direction)
+                .set(IssueSyncCursor::getLastSyncVersion, version)
+                .set(IssueSyncCursor::getLastSyncTime, LocalDateTime.now()));
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/LocalPushServiceImpl.java

@@ -0,0 +1,176 @@
+package com.usky.issue.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.domain.IssueSyncCursor;
+import com.usky.issue.domain.IssueSyncData;
+import com.usky.issue.mapper.IssueSyncCursorMapper;
+import com.usky.issue.mapper.IssueSyncDataMapper;
+import com.usky.issue.service.LocalPushService;
+import com.usky.issue.service.client.CloudPlatformClient;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.support.CloudEntitySupport;
+import com.usky.issue.service.vo.SyncPacket;
+import com.usky.issue.service.vo.SyncResponse;
+import com.usky.issue.service.vo.SyncResult;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * 本地推送服务实现
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Slf4j
+@Service
+public class LocalPushServiceImpl implements LocalPushService {
+
+    @Value("${sync.push.batch-size:100}")
+    private int batchSize;
+
+    @Autowired
+    private IssueSyncDataMapper syncDataMapper;
+    @Autowired
+    private IssueSyncCursorMapper cursorMapper;
+    @Autowired
+    private CloudPlatformClient cloudPlatformClient;
+
+    @Override
+    public SyncResult pushAllPending(String tableName, IssueCloudConfig config) {
+        if (config == null || config.getTenantId() == null) {
+            return SyncResult.empty();
+        }
+        try {
+            return pushByTenant(config.getTenantId(), tableName, config);
+        } catch (Exception e) {
+            log.error("推送租户{}失败", config.getTenantId(), e);
+            SyncResult result = SyncResult.empty();
+            result.setFailureCount(1);
+            result.setErrorMessage(e.getMessage());
+            return result;
+        }
+    }
+
+    @Override
+    public SyncResult pushByTenant(Integer tenantId, String tableName, IssueCloudConfig config) {
+        SyncResult result = SyncResult.empty();
+        IssueSyncCursor cursor = getOrInitCursor(tenantId, tableName, SyncDirection.UP.getCode());
+        Long lastVersion = cursor.getLastSyncVersion() == null ? 0L : cursor.getLastSyncVersion();
+
+        while (true) {
+            List<IssueSyncData> dataList = syncDataMapper.selectList(new LambdaQueryWrapper<IssueSyncData>()
+                    .eq(IssueSyncData::getTenantId, tenantId)
+                    .eq(IssueSyncData::getTableName, tableName)
+                    .in(IssueSyncData::getSyncStatus, 0, 2)
+                    .gt(IssueSyncData::getSyncVersion, lastVersion)
+                    .eq(IssueSyncData::getDeleted, 0)
+                    .orderByAsc(IssueSyncData::getSyncVersion)
+                    .last("LIMIT " + batchSize));
+
+            if (dataList.isEmpty()) {
+                break;
+            }
+
+            result.setTotalCount(result.getTotalCount() + dataList.size());
+            Long batchVersion = dataList.stream()
+                    .mapToLong(IssueSyncData::getSyncVersion)
+                    .max()
+                    .orElse(lastVersion);
+
+            SyncPacket packet = SyncPacket.builder()
+                    .tenantId(String.valueOf(tenantId))
+                    .tableName(tableName)
+                    .direction(SyncDirection.UP)
+                    .dataList(dataList)
+                    .batchVersion(batchVersion)
+                    .hasMore(dataList.size() >= batchSize)
+                    .nonce(UUID.randomUUID().toString())
+                    .build();
+
+            SyncResponse response = cloudPlatformClient.pushToCloud(packet, config);
+            if (response != null && response.isSuccess()) {
+                lastVersion = response.getConfirmedVersion() != null ? response.getConfirmedVersion() : batchVersion;
+                updateCursor(tenantId, tableName, SyncDirection.UP.getCode(), lastVersion);
+                List<Long> ids = dataList.stream().map(IssueSyncData::getId).collect(Collectors.toList());
+                markSynced(ids);
+                result.setSuccessCount(result.getSuccessCount() + dataList.size());
+                if (Boolean.FALSE.equals(packet.getHasMore())) {
+                    break;
+                }
+            } else {
+                String message = response != null ? response.getMessage() : "云端无响应";
+                log.error("云端接收失败: {}", message);
+                result.setFailureCount(result.getFailureCount() + dataList.size());
+                result.setErrorMessage(message);
+                markFailed(dataList.stream().map(IssueSyncData::getId).collect(Collectors.toList()));
+                break;
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public void resetCursor(Integer tenantId, String tableName) {
+        cursorMapper.update(null, new LambdaUpdateWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getTenantId, tenantId)
+                .eq(IssueSyncCursor::getTableName, tableName)
+                .eq(IssueSyncCursor::getDirection, SyncDirection.UP.getCode())
+                .set(IssueSyncCursor::getLastSyncVersion, 0L)
+                .set(IssueSyncCursor::getLastSyncTime, LocalDateTime.now()));
+    }
+
+    private IssueSyncCursor getOrInitCursor(Integer tenantId, String tableName, String direction) {
+        IssueSyncCursor cursor = cursorMapper.selectOne(new LambdaQueryWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getTenantId, tenantId)
+                .eq(IssueSyncCursor::getTableName, tableName)
+                .eq(IssueSyncCursor::getDirection, direction)
+                .eq(IssueSyncCursor::getDeleted, 0));
+        if (cursor == null) {
+            cursor = new IssueSyncCursor();
+            cursor.setTenantId(tenantId);
+            cursor.setTableName(tableName);
+            cursor.setDirection(direction);
+            cursor.setLastSyncVersion(0L);
+            cursor.setLastSyncTime(LocalDateTime.now());
+            CloudEntitySupport.fillOnInsert(cursor);
+            cursorMapper.insert(cursor);
+        }
+        return cursor;
+    }
+
+    private void updateCursor(Integer tenantId, String tableName, String direction, Long version) {
+        cursorMapper.update(null, new LambdaUpdateWrapper<IssueSyncCursor>()
+                .eq(IssueSyncCursor::getTenantId, tenantId)
+                .eq(IssueSyncCursor::getTableName, tableName)
+                .eq(IssueSyncCursor::getDirection, direction)
+                .set(IssueSyncCursor::getLastSyncVersion, version)
+                .set(IssueSyncCursor::getLastSyncTime, LocalDateTime.now()));
+    }
+
+    private void markSynced(List<Long> ids) {
+        if (ids == null || ids.isEmpty()) {
+            return;
+        }
+        syncDataMapper.update(null, new LambdaUpdateWrapper<IssueSyncData>()
+                .in(IssueSyncData::getId, ids)
+                .set(IssueSyncData::getSyncStatus, 1));
+    }
+
+    private void markFailed(List<Long> ids) {
+        if (ids == null || ids.isEmpty()) {
+            return;
+        }
+        syncDataMapper.update(null, new LambdaUpdateWrapper<IssueSyncData>()
+                .in(IssueSyncData::getId, ids)
+                .set(IssueSyncData::getSyncStatus, 2));
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/impl/LocalSyncAgentImpl.java

@@ -0,0 +1,82 @@
+package com.usky.issue.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.domain.IssueSyncData;
+import com.usky.issue.mapper.IssueSyncDataMapper;
+import com.usky.issue.service.CloudConfigService;
+import com.usky.issue.service.CloudSyncReceiveService;
+import com.usky.issue.service.LocalPullService;
+import com.usky.issue.service.LocalPushService;
+import com.usky.issue.service.LocalSyncAgent;
+import com.usky.issue.service.enums.SyncDirection;
+import com.usky.issue.service.enums.SyncTypeEnum;
+import com.usky.issue.service.vo.SyncResult;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 本地同步 Agent 实现
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Slf4j
+@Component
+public class LocalSyncAgentImpl implements LocalSyncAgent {
+
+    @Autowired
+    private LocalPushService pushService;
+    @Autowired
+    private LocalPullService pullService;
+    @Autowired
+    private IssueSyncDataMapper syncDataMapper;
+    @Autowired
+    private CloudSyncReceiveService cloudSyncReceiveService;
+    @Autowired
+    private CloudConfigService cloudConfigService;
+
+    @Scheduled(fixedDelayString = "${sync.push.fixed-delay-ms:5000}")
+    public void autoPush() {
+        try {
+            IssueCloudConfig config = cloudConfigService.requireActiveConfig();
+            for (SyncTypeEnum type : SyncTypeEnum.values()) {
+                pushService.pushAllPending(type.getCode(), config);
+            }
+        } catch (Exception e) {
+            log.debug("自动推送跳过: {}", e.getMessage());
+        }
+    }
+
+    @Override
+    public SyncResult executeSync(SyncTypeEnum syncType, IssueCloudConfig config,
+                                  SyncDirection direction, boolean fullSync) {
+        String tableName = syncType.getCode();
+        SyncResult result = SyncResult.empty();
+
+        if (direction == SyncDirection.UP || direction == SyncDirection.BOTH) {
+            if (fullSync) {
+                pushService.resetCursor(config.getTenantId(), tableName);
+            }
+            result.merge(pushService.pushByTenant(config.getTenantId(), tableName, config));
+        }
+        if (direction == SyncDirection.DOWN || direction == SyncDirection.BOTH) {
+            result.merge(pullService.pullByTenant(config.getTenantId(), tableName, config));
+        }
+        return result;
+    }
+
+    @Override
+    public int countPending(SyncTypeEnum syncType, IssueCloudConfig config) {
+        String tableName = syncType.getCode();
+        Long upCount = syncDataMapper.selectCount(new LambdaQueryWrapper<IssueSyncData>()
+                .eq(IssueSyncData::getTenantId, config.getTenantId())
+                .eq(IssueSyncData::getTableName, tableName)
+                .in(IssueSyncData::getSyncStatus, 0, 2)
+                .eq(IssueSyncData::getDeleted, 0));
+        int downCount = cloudSyncReceiveService.countPendingDownQueue(config.getTenantId(), tableName);
+        return (upCount == null ? 0 : upCount.intValue()) + downCount;
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/util/HmacSignUtil.java

@@ -0,0 +1,31 @@
+package com.usky.issue.service.util;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+/**
+ * HMAC 签名工具
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+public final class HmacSignUtil {
+
+    private static final String HMAC_SHA256 = "HmacSHA256";
+
+    private HmacSignUtil() {
+    }
+
+    public static String sign(String content, String secret) {
+        try {
+            Mac mac = Mac.getInstance(HMAC_SHA256);
+            mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), HMAC_SHA256));
+            byte[] hash = mac.doFinal(content.getBytes(StandardCharsets.UTF_8));
+            return Base64.getEncoder().encodeToString(hash);
+        } catch (Exception e) {
+            throw new IllegalStateException("HMAC签名失败", e);
+        }
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/CloudConnectionTestRequest.java

@@ -0,0 +1,18 @@
+package com.usky.issue.service.vo;
+
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+
+/**
+ * 连接测试请求
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+public class CloudConnectionTestRequest {
+
+    @NotNull(message = "租户ID不能为空")
+    private Integer tenantId;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/CloudConnectionTestResponse.java

@@ -13,5 +13,14 @@ public class CloudConnectionTestResponse {
 
     private boolean success;
     private String message;
+    /** 0未知 1成功 2失败 */
     private Integer connectionStatus;
+    /** 云平台 HTTP/TCP 层是否可达 */
+    private Boolean networkReachable;
+    /** 认证层是否通过 */
+    private Boolean authValid;
+    /** 租户配置是否存在且启用 */
+    private Boolean tenantExists;
+    /** 耗时（毫秒） */
+    private Long costTime;
 }

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/CloudConnectionTestResult.java

@@ -0,0 +1,81 @@
+package com.usky.issue.service.vo;
+
+import com.usky.issue.service.constant.CloudIntegrationConstants;
+import lombok.Data;
+
+/**
+ * 连接测试内部结果（HTTP 客户端层）
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+public class CloudConnectionTestResult {
+
+    private boolean success;
+    private String message;
+    /** 是否已到达云平台（HTTP/TCP 层可达） */
+    private boolean networkReachable;
+    /** 认证层是否通过 */
+    private boolean authValid;
+    /** 租户配置是否存在且启用 */
+    private boolean tenantExists;
+    /** 耗时（毫秒） */
+    private Long costTime;
+
+    public static CloudConnectionTestResult success(String message) {
+        CloudConnectionTestResult result = new CloudConnectionTestResult();
+        result.setSuccess(true);
+        result.setMessage(message);
+        result.setNetworkReachable(true);
+        result.setAuthValid(true);
+        result.setTenantExists(true);
+        return result;
+    }
+
+    public static CloudConnectionTestResult fullSuccess(long costTime) {
+        CloudConnectionTestResult result = success(CloudIntegrationConstants.MSG_CONNECTION_SUCCESS);
+        result.setCostTime(costTime);
+        return result;
+    }
+
+    public static CloudConnectionTestResult networkFailure(String message) {
+        CloudConnectionTestResult result = new CloudConnectionTestResult();
+        result.setSuccess(false);
+        result.setMessage(message);
+        result.setNetworkReachable(false);
+        result.setAuthValid(false);
+        result.setTenantExists(false);
+        return result;
+    }
+
+    public static CloudConnectionTestResult authFailure(long costTime, String message) {
+        CloudConnectionTestResult result = new CloudConnectionTestResult();
+        result.setSuccess(false);
+        result.setMessage(message);
+        result.setNetworkReachable(true);
+        result.setAuthValid(false);
+        result.setTenantExists(false);
+        result.setCostTime(costTime);
+        return result;
+    }
+
+    public static CloudConnectionTestResult tenantMissing(long costTime) {
+        CloudConnectionTestResult result = new CloudConnectionTestResult();
+        result.setSuccess(false);
+        result.setMessage(CloudIntegrationConstants.MSG_TENANT_CONFIG_MISSING);
+        result.setNetworkReachable(true);
+        result.setAuthValid(true);
+        result.setTenantExists(false);
+        result.setCostTime(costTime);
+        return result;
+    }
+
+    public static CloudConnectionTestResult failure(boolean networkReachable, String message) {
+        CloudConnectionTestResult result = new CloudConnectionTestResult();
+        result.setSuccess(false);
+        result.setMessage(message);
+        result.setNetworkReachable(networkReachable);
+        return result;
+    }
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/SyncPacket.java

@@ -0,0 +1,30 @@
+package com.usky.issue.service.vo;
+
+import com.usky.issue.service.enums.SyncDirection;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/**
+ * 同步数据包
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SyncPacket {
+
+    private String tenantId;
+    private String tableName;
+    private SyncDirection direction;
+    private List<?> dataList;
+    private Long batchVersion;
+    private Boolean hasMore;
+    private String nonce;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/SyncResponse.java

@@ -0,0 +1,24 @@
+package com.usky.issue.service.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * 同步响应
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SyncResponse {
+
+    private boolean success;
+    private String message;
+    private Long confirmedVersion;
+    private Integer acceptedCount;
+}

service-issue/service-issue-biz/src/main/java/com/usky/issue/service/vo/SyncResult.java

@@ -0,0 +1,34 @@
+package com.usky.issue.service.vo;
+
+import lombok.Data;
+
+/**
+ * 同步执行结果
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@Data
+public class SyncResult {
+
+    private int totalCount;
+    private int successCount;
+    private int failureCount;
+    private String errorMessage;
+
+    public static SyncResult empty() {
+        return new SyncResult();
+    }
+
+    public void merge(SyncResult other) {
+        if (other == null) {
+            return;
+        }
+        this.totalCount += other.totalCount;
+        this.successCount += other.successCount;
+        this.failureCount += other.failureCount;
+        if (other.errorMessage != null) {
+            this.errorMessage = other.errorMessage;
+        }
+    }
+}

service-issue/service-issue-biz/src/main/resources/sql/issue_cloud_sync_tables.sql

@@ -0,0 +1,51 @@
+-- 本地同步游标表
+CREATE TABLE IF NOT EXISTS issue_sync_cursor (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id INT NOT NULL,
+    table_name VARCHAR(50) NOT NULL COMMENT '同步类型/表名，如 TENANT、USER',
+    direction VARCHAR(10) NOT NULL COMMENT 'UP/DOWN',
+    last_sync_version BIGINT DEFAULT 0 COMMENT '最后同步的版本号',
+    last_sync_time DATETIME,
+    create_by VARCHAR(64),
+    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
+    update_by VARCHAR(64),
+    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    deleted TINYINT DEFAULT 0,
+    UNIQUE KEY uk_tenant_table_dir (tenant_id, table_name, direction)
+) COMMENT '同步游标表';
+
+-- 云端下发队列（云端→本地）
+CREATE TABLE IF NOT EXISTS issue_sync_down_queue (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id INT NOT NULL,
+    table_name VARCHAR(50) NOT NULL,
+    operation VARCHAR(20) COMMENT 'INSERT/UPDATE/DELETE',
+    payload JSON NOT NULL COMMENT '完整数据JSON',
+    sync_version BIGINT NOT NULL,
+    status TINYINT DEFAULT 0 COMMENT '0:待拉取 1:已确认 2:失败',
+    create_by VARCHAR(64),
+    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
+    update_by VARCHAR(64),
+    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    deleted TINYINT DEFAULT 0,
+    INDEX idx_tenant_status (tenant_id, status)
+) COMMENT '云端下发队列';
+
+-- 本地同步业务数据表
+CREATE TABLE IF NOT EXISTS issue_sync_data (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id INT NOT NULL,
+    table_name VARCHAR(50) NOT NULL COMMENT '同步类型，如 TENANT、USER',
+    data_key VARCHAR(100) NOT NULL COMMENT '业务唯一键',
+    payload JSON NOT NULL COMMENT '业务数据JSON',
+    sync_version BIGINT NOT NULL COMMENT '数据版本时间戳(毫秒)',
+    sync_status TINYINT DEFAULT 0 COMMENT '0:未同步 1:已推送 2:推送失败',
+    sync_direction VARCHAR(10) DEFAULT 'LOCAL' COMMENT 'LOCAL/CLOUD 数据来源',
+    create_by VARCHAR(64),
+    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
+    update_by VARCHAR(64),
+    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    deleted TINYINT DEFAULT 0,
+    UNIQUE KEY uk_tenant_table_key (tenant_id, table_name, data_key),
+    INDEX idx_tenant_table_status (tenant_id, table_name, sync_status)
+) COMMENT '同步业务数据表';

service-issue/service-issue-biz/src/main/resources/本地云端双向数据同步与连通性测试方案.md

@@ -0,0 +1,796 @@
+# 本地-云端双向数据同步与连通性测试方案
+
+## 一、方案概述
+
+### 1.1 背景与约束
+- **本地服务器A**：不可信区，不受管控，不能暴露云端数据库密码
+- **云服务器**：可信区，持有数据库连接凭证
+- **同步需求**：双向同步（本地↔云端），支持手动触发与自动同步
+- **连通性测试**：验证本地Agent与云端API通道的可用性
+
+### 1.2 核心设计原则
+| 原则 | 说明 |
+|------|------|
+| **单向出站** | 所有连接由本地主动发起（HTTPS出站），云端零暴露 |
+| **密码隔离** | 云端DB密码仅存在于云端服务，本地仅持有API Token |
+| **游标断点续传** | 基于`sync_version`游标，支持断网恢复后增量同步 |
+| **冲突解决** | 默认`last-write-wins`，云端优先；可配置为本地优先 |
+| **分层检测** | 连通性测试覆盖网络层、认证层、租户层 |
+
+---
+
+## 二、架构设计
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│                         云服务器（可信区）                          │
+│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐   │
+│  │  SyncReceive   │  │ SyncCommand    │  │   Cloud Data Service   │   │
+│  │  Controller    │  │  Controller    │  │   (直连云端DB)          │   │
+│  │  (接收推送)     │  │  (下发指令/数据) │  │                        │   │
+│  └──────┬─────────┘  └──────┬───────┘  └──────────────────────┘   │
+│         ▲                   │                                      │
+│         │                   │                                      │
+│    HTTPS│出站上推            │HTTPS出站长轮询拉取                    │
+│         │                   │                                      │
+│  ┌──────┴───────────────────┴──────────────────────────────────┐   │
+│  │                    本地服务器A（不可信区）                      │   │
+│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │   │
+│  │  │  LocalPush     │  │  LocalPull     │  │  Local Data    │       │   │
+│  │  │  Service       │  │  Service       │  │  Service       │       │   │
+│  │  │  (读取本地变更)  │  │  (拉云端指令)   │  │  (操作本地DB)   │       │   │
+│  │  └──────┬─────────┘  └──────┬───────┘  └──────┬───────┘       │   │
+│  │         │                   │                  │               │   │
+│  │  ┌──────┴───────────────────┴──────────────────┘               │   │
+│  │  │              Local Sync Agent (调度核心)                       │   │
+│  │  │   • Push线程：定时/实时扫描本地变更 → 推云端                    │   │
+│  │  │   • Pull线程：长轮询云端指令队列 → 写入本地                     │   │
+│  │  │   • 手动触发：REST接口接收任务 → 提交线程池                     │   │
+│  │  └────────────────────────────────────────────────────────────┘   │
+│  └─────────────────────────────────────────────────────────────────┘
+```
+
+---
+
+## 三、数据库表设计
+
+### 3.1 本地业务表（示例：device_data）
+```sql
+CREATE TABLE local_device_data (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id VARCHAR(50) NOT NULL,
+    device_code VARCHAR(50),
+    sensor_value DECIMAL(10,2),
+    -- 同步控制字段
+    sync_version BIGINT COMMENT '数据版本时间戳(毫秒)',
+    sync_status TINYINT DEFAULT 0 COMMENT '0:未同步 1:已推送 2:推送失败',
+    sync_direction VARCHAR(10) DEFAULT 'LOCAL' COMMENT 'LOCAL/CLOUD 数据来源',
+    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    create_time DATETIME DEFAULT CURRENT_TIMESTAMP
+);
+```
+
+### 3.2 本地同步游标表
+```sql
+CREATE TABLE local_sync_cursor (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id VARCHAR(50) NOT NULL,
+    table_name VARCHAR(50) NOT NULL,
+    direction VARCHAR(10) NOT NULL COMMENT 'UP/DOWN',
+    last_sync_version BIGINT DEFAULT 0 COMMENT '最后同步的版本号',
+    last_sync_time DATETIME,
+    UNIQUE KEY uk_tenant_table_dir (tenant_id, table_name, direction)
+);
+```
+
+### 3.3 云端下发队列（云端→本地）
+```sql
+CREATE TABLE cloud_sync_down_queue (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT,
+    tenant_id VARCHAR(50) NOT NULL,
+    table_name VARCHAR(50) NOT NULL,
+    operation VARCHAR(20) COMMENT 'INSERT/UPDATE/DELETE',
+    payload JSON NOT NULL COMMENT '完整数据JSON',
+    sync_version BIGINT NOT NULL,
+    status TINYINT DEFAULT 0 COMMENT '0:待拉取 1:已确认 2:失败',
+    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
+    INDEX idx_tenant_status (tenant_id, status)
+);
+```
+
+---
+
+## 四、公共实体与枚举
+
+### 4.1 同步方向枚举
+```java
+@Getter
+@AllArgsConstructor
+public enum SyncDirection {
+    UP("本地→云端"),
+    DOWN("云端→本地"),
+    BOTH("双向");
+    private final String desc;
+}
+```
+
+### 4.2 数据包
+```java
+@Data
+@Builder
+public class SyncPacket<T> {
+    private String tenantId;
+    private String tableName;
+    private SyncDirection direction;
+    private List<T> dataList;
+    private Long batchVersion;
+    private Boolean hasMore;
+    private String nonce;
+}
+```
+
+### 4.3 同步响应
+```java
+@Data
+@Builder
+public class SyncResponse {
+    private boolean success;
+    private String message;
+    private Long confirmedVersion;
+    private Integer acceptedCount;
+}
+```
+
+### 4.4 手动触发请求
+```java
+@Data
+public class SyncTriggerRequest {
+    private String tenantId;
+    private String tableName;
+    private SyncDirection direction;
+    private Boolean fullSync; // 是否全量同步（重置游标）
+}
+```
+
+---
+
+## 五、云端服务实现
+
+### 5.1 云端接收控制器（本地→云端）
+```java
+@Slf4j
+@RestController
+@RequestMapping("/api/cloud/sync")
+@RequiredArgsConstructor
+public class CloudSyncReceiveController {
+
+    private final CloudSyncService cloudSyncService;
+    private final CloudAuthService cloudAuthService;
+
+    @PostMapping("/receive")
+    public SyncResponse receive(
+            @RequestHeader("X-App-Token") String token,
+            @RequestHeader("X-Sign") String sign,
+            @RequestBody @Valid SyncPacket packet) {
+
+        if (!cloudAuthService.validate(token, sign, packet)) {
+            return SyncResponse.builder().success(false).message("认证失败或请求已过期").build();
+        }
+
+        if (!cloudSyncService.validateTenant(packet.getTenantId())) {
+            return SyncResponse.builder().success(false).message("租户不存在或已停用").build();
+        }
+
+        try {
+            int count = cloudSyncService.batchUpsert(packet);
+            return SyncResponse.builder()
+                    .success(true)
+                    .message("接收成功")
+                    .acceptedCount(count)
+                    .confirmedVersion(packet.getBatchVersion())
+                    .build();
+        } catch (Exception e) {
+            log.error("云端接收数据失败, tenantId={}", packet.getTenantId(), e);
+            return SyncResponse.builder().success(false).message("写入失败:" + e.getMessage()).build();
+        }
+    }
+}
+```
+
+### 5.2 云端下发控制器（云端→本地）
+```java
+@Slf4j
+@RestController
+@RequestMapping("/api/cloud/sync")
+@RequiredArgsConstructor
+public class CloudSyncCommandController {
+
+    private final CloudSyncService cloudSyncService;
+    private final CloudAuthService cloudAuthService;
+
+    @GetMapping("/poll")
+    public SyncPacket poll(
+            @RequestHeader("X-App-Token") String token,
+            @RequestParam String tenantId,
+            @RequestParam String tableName,
+            @RequestParam(required = false) Long lastVersion) {
+
+        if (!cloudAuthService.validateToken(token)) {
+            return null;
+        }
+        return cloudSyncService.pollDownQueue(tenantId, tableName, lastVersion);
+    }
+
+    @PostMapping("/poll/ack")
+    public SyncResponse ackPoll(
+            @RequestHeader("X-App-Token") String token,
+            @RequestBody List<Long> queueIds) {
+
+        if (!cloudAuthService.validateToken(token)) {
+            return SyncResponse.builder().success(false).message("认证失败").build();
+        }
+        cloudSyncService.ackDownQueue(queueIds);
+        return SyncResponse.builder().success(true).build();
+    }
+}
+```
+
+### 5.3 云端同步服务（冲突解决：云端优先）
+```java
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class CloudSyncService {
+
+    private final CloudDeviceDataMapper deviceDataMapper;
+    private final CloudSyncDownQueueMapper downQueueMapper;
+    private final TenantConfigMapper tenantConfigMapper;
+
+    public boolean validateTenant(String tenantId) {
+        TenantConfig config = tenantConfigMapper.selectByTenantId(tenantId);
+        return config != null && config.getStatus() == 1;
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    public int batchUpsert(SyncPacket packet) {
+        List<CloudDeviceData> list = convertToCloudEntity(packet);
+        int count = 0;
+        for (CloudDeviceData data : list) {
+            CloudDeviceData exist = deviceDataMapper.selectByUk(data.getTenantId(), data.getDeviceCode());
+            if (exist == null) {
+                deviceDataMapper.insert(data);
+                count++;
+            } else if (data.getSyncVersion() >= exist.getSyncVersion()) {
+                data.setId(exist.getId());
+                deviceDataMapper.updateById(data);
+                count++;
+            }
+        }
+        return count;
+    }
+
+    public SyncPacket pollDownQueue(String tenantId, String tableName, Long lastVersion) {
+        if (lastVersion == null) lastVersion = 0L;
+        List<CloudSyncDownQueue> queueList = downQueueMapper.selectPending(
+                tenantId, tableName, lastVersion, 100);
+
+        if (queueList.isEmpty()) return null;
+
+        Long maxVersion = queueList.stream()
+                .mapToLong(CloudSyncDownQueue::getSyncVersion)
+                .max().orElse(lastVersion);
+
+        return SyncPacket.builder()
+                .tenantId(tenantId)
+                .tableName(tableName)
+                .direction(SyncDirection.DOWN)
+                .dataList(queueList)
+                .batchVersion(maxVersion)
+                .hasMore(queueList.size() >= 100)
+                .build();
+    }
+
+    public void ackDownQueue(List<Long> queueIds) {
+        if (queueIds == null || queueIds.isEmpty()) return;
+        downQueueMapper.updateStatus(queueIds, 1);
+    }
+}
+```
+
+---
+
+## 六、本地服务实现
+
+### 6.1 手动触发接口
+```java
+@RestController
+@RequestMapping("/api/local/sync")
+@RequiredArgsConstructor
+public class LocalSyncTriggerController {
+
+    private final LocalSyncAgent syncAgent;
+
+    @PostMapping("/trigger")
+    public String trigger(@RequestBody @Valid SyncTriggerRequest request) {
+        syncAgent.submitTask(request);
+        return "同步任务已提交: " + request.getDirection() + 
+               ", tenant=" + request.getTenantId() + 
+               ", table=" + request.getTableName();
+    }
+
+    @GetMapping("/status")
+    public String status() {
+        return syncAgent.getStatus();
+    }
+}
+```
+
+### 6.2 本地Agent核心调度器
+```java
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class LocalSyncAgent {
+
+    private final LocalPushService pushService;
+    private final LocalPullService pullService;
+    private final ExecutorService executor = new ThreadPoolExecutor(
+            4, 8, 60, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(100),
+            new ThreadFactoryBuilder().setNameFormat("sync-agent-%d").build(),
+            new ThreadPoolExecutor.CallerRunsPolicy()
+    );
+
+    @PostConstruct
+    public void init() {
+        log.info("本地同步Agent启动");
+        executor.submit(this::pullLoop);
+    }
+
+    @PreDestroy
+    public void destroy() {
+        executor.shutdown();
+    }
+
+    @Scheduled(fixedDelay = 5000)
+    public void autoPush() {
+        try {
+            pushService.pushAllPending("device_data");
+        } catch (Exception e) {
+            log.error("自动推送失败", e);
+        }
+    }
+
+    private void pullLoop() {
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                pullService.pullAndApply("device_data");
+                Thread.sleep(3000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                log.error("拉取云端数据失败", e);
+                try {
+                    Thread.sleep(10000);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+        }
+    }
+
+    public void submitTask(SyncTriggerRequest request) {
+        executor.submit(() -> {
+            try {
+                if (request.getDirection() == SyncDirection.UP || request.getDirection() == SyncDirection.BOTH) {
+                    if (Boolean.TRUE.equals(request.getFullSync())) {
+                        pushService.resetCursor(request.getTenantId(), "device_data");
+                    }
+                    pushService.pushByTenant(request.getTenantId(), "device_data");
+                }
+                if (request.getDirection() == SyncDirection.DOWN || request.getDirection() == SyncDirection.BOTH) {
+                    pullService.pullByTenant(request.getTenantId(), "device_data");
+                }
+            } catch (Exception e) {
+                log.error("手动同步任务失败", e);
+            }
+        });
+    }
+
+    public String getStatus() {
+        return "运行中, 活跃线程:" + ((ThreadPoolExecutor) executor).getActiveCount();
+    }
+}
+```
+
+### 6.3 本地推送服务（本地→云端）
+```java
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class LocalPushService {
+
+    private final LocalDeviceDataMapper deviceDataMapper;
+    private final LocalSyncCursorMapper cursorMapper;
+    private final RestTemplate restTemplate;
+    private final ObjectMapper objectMapper;
+
+    @Value("${cloud.api.endpoint}")
+    private String cloudEndpoint;
+    @Value("${cloud.api.token}")
+    private String apiToken;
+
+    public void pushAllPending(String tableName) {
+        List<String> tenantIds = deviceDataMapper.selectPendingTenantIds(tableName);
+        for (String tenantId : tenantIds) {
+            try {
+                pushByTenant(tenantId, tableName);
+            } catch (Exception e) {
+                log.error("推送租户{}失败", tenantId, e);
+            }
+        }
+    }
+
+    public void pushByTenant(String tenantId, String tableName) {
+        LocalSyncCursor cursor = getOrInitCursor(tenantId, tableName, "UP");
+        Long lastVersion = cursor.getLastSyncVersion();
+
+        while (true) {
+            List<LocalDeviceData> dataList = deviceDataMapper.selectByVersionGt(
+                    tenantId, lastVersion, 100);
+            if (dataList.isEmpty()) break;
+
+            Long batchVersion = dataList.stream()
+                    .mapToLong(LocalDeviceData::getSyncVersion)
+                    .max().orElse(lastVersion);
+
+            SyncPacket packet = SyncPacket.builder()
+                    .tenantId(tenantId)
+                    .tableName(tableName)
+                    .direction(SyncDirection.UP)
+                    .dataList(dataList)
+                    .batchVersion(batchVersion)
+                    .hasMore(dataList.size() >= 100)
+                    .nonce(UUID.randomUUID().toString())
+                    .build();
+
+            SyncResponse response = pushToCloud(packet);
+            if (response != null && response.isSuccess()) {
+                lastVersion = response.getConfirmedVersion();
+                updateCursor(tenantId, tableName, "UP", lastVersion);
+                List<Long> ids = dataList.stream().map(LocalDeviceData::getId).toList();
+                deviceDataMapper.markSynced(ids);
+                if (Boolean.FALSE.equals(packet.getHasMore())) break;
+            } else {
+                log.error("云端接收失败: {}", response != null ? response.getMessage() : "无响应");
+                break;
+            }
+        }
+    }
+
+    public void resetCursor(String tenantId, String tableName) {
+        cursorMapper.updateLastVersion(tenantId, tableName, "UP", 0L);
+    }
+
+    private SyncResponse pushToCloud(SyncPacket packet) {
+        try {
+            HttpHeaders headers = new HttpHeaders();
+            headers.setContentType(MediaType.APPLICATION_JSON);
+            headers.set("X-App-Token", apiToken);
+            headers.set("X-Sign", HmacUtils.sign(objectMapper.writeValueAsString(packet), apiToken));
+
+            ResponseEntity<SyncResponse> response = restTemplate.exchange(
+                    cloudEndpoint + "/api/cloud/sync/receive",
+                    HttpMethod.POST,
+                    new HttpEntity<>(packet, headers),
+                    SyncResponse.class
+            );
+            return response.getBody();
+        } catch (Exception e) {
+            log.error("推送云端网络异常", e);
+            return null;
+        }
+    }
+
+    private LocalSyncCursor getOrInitCursor(String tenantId, String tableName, String direction) {
+        LocalSyncCursor cursor = cursorMapper.selectByUk(tenantId, tableName, direction);
+        if (cursor == null) {
+            cursor = new LocalSyncCursor();
+            cursor.setTenantId(tenantId);
+            cursor.setTableName(tableName);
+            cursor.setDirection(direction);
+            cursor.setLastSyncVersion(0L);
+            cursorMapper.insert(cursor);
+        }
+        return cursor;
+    }
+
+    private void updateCursor(String tenantId, String tableName, String direction, Long version) {
+        cursorMapper.updateLastVersion(tenantId, tableName, direction, version);
+    }
+}
+```
+
+### 6.4 本地拉取服务（云端→本地）
+```java
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class LocalPullService {
+
+    private final LocalDeviceDataMapper deviceDataMapper;
+    private final LocalSyncCursorMapper cursorMapper;
+    private final RestTemplate restTemplate;
+    private final ObjectMapper objectMapper;
+
+    @Value("${cloud.api.endpoint}")
+    private String cloudEndpoint;
+    @Value("${cloud.api.token}")
+    private String apiToken;
+
+    public void pullAndApply(String tableName) {
+        List<String> tenantIds = cursorMapper.selectAllTenantIds();
+        for (String tenantId : tenantIds) {
+            try {
+                pullByTenant(tenantId, tableName);
+            } catch (Exception e) {
+                log.error("拉取租户{}数据失败", tenantId, e);
+            }
+        }
+    }
+
+    public void pullByTenant(String tenantId, String tableName) {
+        LocalSyncCursor cursor = getOrInitCursor(tenantId, tableName, "DOWN");
+        Long lastVersion = cursor.getLastSyncVersion();
+
+        while (true) {
+            SyncPacket packet = pollCloud(tenantId, tableName, lastVersion);
+            if (packet == null || packet.getDataList() == null || packet.getDataList().isEmpty()) break;
+
+            List<CloudSyncDownQueue> queueList = (List<CloudSyncDownQueue>) packet.getDataList();
+            List<Long> ackIds = queueList.stream().map(CloudSyncDownQueue::getId).collect(Collectors.toList());
+
+            for (CloudSyncDownQueue queue : queueList) {
+                applyToLocal(queue);
+            }
+
+            lastVersion = packet.getBatchVersion();
+            updateCursor(tenantId, tableName, "DOWN", lastVersion);
+            ackCloud(ackIds);
+
+            if (Boolean.FALSE.equals(packet.getHasMore())) break;
+        }
+    }
+
+    private void applyToLocal(CloudSyncDownQueue queue) {
+        try {
+            LocalDeviceData data = objectMapper.readValue(queue.getPayload().toString(), LocalDeviceData.class);
+            data.setSyncVersion(queue.getSyncVersion());
+            data.setSyncDirection("CLOUD");
+            data.setSyncStatus(1);
+
+            LocalDeviceData exist = deviceDataMapper.selectByUk(data.getTenantId(), data.getDeviceCode());
+            if (exist == null) {
+                deviceDataMapper.insert(data);
+            } else if (queue.getSyncVersion() >= exist.getSyncVersion()) {
+                data.setId(exist.getId());
+                deviceDataMapper.updateById(data);
+            }
+        } catch (Exception e) {
+            log.error("应用云端数据到本地失败, queueId={}", queue.getId(), e);
+        }
+    }
+
+    private SyncPacket pollCloud(String tenantId, String tableName, Long lastVersion) {
+        try {
+            String url = UriComponentsBuilder.fromHttpUrl(cloudEndpoint + "/api/cloud/sync/poll")
+                    .queryParam("tenantId", tenantId)
+                    .queryParam("tableName", tableName)
+                    .queryParam("lastVersion", lastVersion)
+                    .toUriString();
+            return restTemplate.getForObject(url, SyncPacket.class);
+        } catch (Exception e) {
+            log.error("轮询云端失败", e);
+            return null;
+        }
+    }
+
+    private void ackCloud(List<Long> queueIds) {
+        try {
+            restTemplate.postForObject(
+                    cloudEndpoint + "/api/cloud/sync/poll/ack",
+                    queueIds,
+                    SyncResponse.class
+            );
+        } catch (Exception e) {
+            log.error("ACK云端失败", e);
+        }
+    }
+
+    private LocalSyncCursor getOrInitCursor(String tenantId, String tableName, String direction) {
+        LocalSyncCursor cursor = cursorMapper.selectByUk(tenantId, tableName, direction);
+        if (cursor == null) {
+            cursor = new LocalSyncCursor();
+            cursor.setTenantId(tenantId);
+            cursor.setTableName(tableName);
+            cursor.setDirection(direction);
+            cursor.setLastSyncVersion(0L);
+            cursorMapper.insert(cursor);
+        }
+        return cursor;
+    }
+
+    private void updateCursor(String tenantId, String tableName, String direction, Long version) {
+        cursorMapper.updateLastVersion(tenantId, tableName, direction, version);
+    }
+}
+```
+
+---
+
+## 七、连通性测试接口
+
+### 7.1 测试目标转变
+| 原目标 | 新目标 |
+|--------|--------|
+| 本地能否连云端DB | 本地Agent能否正常推送/拉取 |
+| 云端租户配置是否存在 | 云端API返回的认证结果 |
+
+### 7.2 本地连通性测试接口
+```java
+@RestController
+@RequestMapping("/api/local/health")
+@RequiredArgsConstructor
+public class LocalHealthCheckController {
+
+    private final RestTemplate restTemplate;
+    @Value("${cloud.api.endpoint}")
+    private String cloudEndpoint;
+    @Value("${cloud.api.token}")
+    private String apiToken;
+
+    @GetMapping("/check")
+    public HealthCheckResponse check(@RequestParam String tenantId) {
+        long start = System.currentTimeMillis();
+
+        try {
+            String pingUrl = cloudEndpoint + "/api/cloud/sync/ping";
+            restTemplate.getForObject(pingUrl, String.class);
+
+            HttpHeaders headers = new HttpHeaders();
+            headers.set("X-App-Token", apiToken);
+            HttpEntity entity = new HttpEntity<>(headers);
+            ResponseEntity<String> authResp = restTemplate.exchange(
+                    cloudEndpoint + "/api/cloud/sync/auth-check?tenantId=" + tenantId,
+                    HttpMethod.GET, entity, String.class);
+
+            boolean tenantExists = "VALID".equals(authResp.getBody());
+
+            return HealthCheckResponse.builder()
+                    .success(true)
+                    .networkReachable(true)
+                    .authValid(true)
+                    .tenantExists(tenantExists)
+                    .costTime(System.currentTimeMillis() - start)
+                    .build();
+
+        } catch (ResourceAccessException e) {
+            return HealthCheckResponse.builder()
+                    .success(false)
+                    .message("网络不可达：" + e.getMessage())
+                    .build();
+        } catch (HttpClientErrorException.Unauthorized e) {
+            return HealthCheckResponse.builder()
+                    .success(false)
+                    .message("认证失败：Token无效")
+                    .build();
+        }
+    }
+}
+```
+
+### 7.3 响应示例
+```json
+{
+  "success": true,
+  "networkReachable": true,
+  "authValid": true,
+  "tenantExists": true,
+  "costTime": 245
+}
+```
+
+---
+
+## 八、配置说明
+
+### 8.1 本地配置（application.yml）
+```yaml
+server:
+  port: 9887
+
+cloud:
+  api:
+    endpoint: https://cloud.example.com
+    token: ${CLOUD_API_TOKEN}  # 环境变量注入，非DB密码
+
+sync:
+  push:
+    batch-size: 100
+    fixed-delay-ms: 5000       # 准实时：每5秒扫描
+  pull:
+    interval-ms: 3000          # 长轮询间隔
+    fail-backoff-ms: 10000     # 失败退避
+```
+
+### 8.2 云端配置
+```yaml
+server:
+  port: 8080
+
+spring:
+  datasource:
+    url: jdbc:mysql://localhost:3306/cloud_db
+    username: cloud_root
+    password: ${DB_PASSWORD}   # 仅云端持有
+```
+
+---
+
+## 九、API调用示例
+
+### 9.1 手动触发本地→云端
+```bash
+curl -X POST http://localhost:9887/api/local/sync/trigger   -H "Content-Type: application/json"   -d '{
+    "tenantId": "TENANT_001",
+    "tableName": "device_data",
+    "direction": "UP"
+  }'
+```
+
+### 9.2 手动触发双向全量同步
+```bash
+curl -X POST http://localhost:9887/api/local/sync/trigger   -H "Content-Type: application/json"   -d '{
+    "tenantId": "TENANT_001",
+    "tableName": "device_data",
+    "direction": "BOTH",
+    "fullSync": true
+  }'
+```
+
+### 9.3 查询Agent状态
+```bash
+curl http://localhost:9887/api/local/sync/status
+```
+
+### 9.4 连通性测试
+```bash
+curl "http://localhost:9887/api/local/health/check?tenantId=TENANT_001"
+```
+
+---
+
+## 十、方案特性总结
+
+| 能力 | 实现方式 | 说明 |
+|------|---------|------|
+| **本地→云端** | PushService 定时扫描 + 批量推送 | 准实时（5秒级） |
+| **云端→本地** | PullService 长轮询 + 写入本地 | 准实时（3秒级） |
+| **手动触发** | REST接口提交任务到线程池 | 异步执行，不阻塞HTTP |
+| **自动实时** | 可接入Canal/Debezium替换轮询 | 当前用定时轮询兜底 |
+| **冲突解决** | sync_version 时间戳，云端优先 | 可配置为本地优先 |
+| **密码安全** | 本地只有API Token，无DB密码 | 云端Token可吊销、可限流 |
+| **断点续传** | sync_cursor 游标表 | 重启后从断点继续 |
+| **单向出站** | 所有连接由本地发起 | 云端零端口暴露 |
+
+---
+
+## 十一、扩展建议
+
+1. **CDC实时接入**：使用Canal或Debezium监听本地Binlog，替换轮询扫描，实现毫秒级同步
+2. **消息队列缓冲**：云端引入Kafka/RabbitMQ，削峰填谷，避免突发流量压垮云端DB
+3. **数据压缩**：大数据量时启用Gzip压缩，减少带宽占用
+4. **监控告警**：对同步延迟、失败率、队列堆积进行监控，超过阈值触发告警
+

service-issue/service-issue-biz/src/test/java/com/usky/issue/cloud/service/CloudConnectionTestServiceTest.java

@@ -0,0 +1,98 @@
+package com.usky.issue.cloud.service;
+
+import com.usky.issue.domain.IssueCloudConfig;
+import com.usky.issue.mapper.IssueCloudConfigMapper;
+import com.usky.issue.mapper.IssueSyncStatusMapper;
+import com.usky.issue.mapper.IssueSyncTaskMapper;
+import com.usky.issue.mapper.IssueTriggerEventMapper;
+import com.usky.issue.service.CloudOperationLogService;
+import com.usky.issue.service.client.CloudPlatformClient;
+import com.usky.issue.service.constant.CloudIntegrationConstants;
+import com.usky.issue.service.impl.CloudConfigServiceImpl;
+import com.usky.issue.service.util.AesGcmCipher;
+import com.usky.issue.service.vo.CloudConnectionTestRequest;
+import com.usky.issue.service.vo.CloudConnectionTestResponse;
+import com.usky.issue.service.vo.CloudConnectionTestResult;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 连接测试服务单元测试
+ *
+ * @author fyc
+ * @date 2026-06-08
+ */
+@ExtendWith(MockitoExtension.class)
+class CloudConnectionTestServiceTest {
+
+    @InjectMocks
+    private CloudConfigServiceImpl cloudConfigService;
+
+    @Mock
+    private IssueCloudConfigMapper configMapper;
+    @Mock
+    private IssueSyncStatusMapper syncStatusMapper;
+    @Mock
+    private IssueTriggerEventMapper triggerEventMapper;
+    @Mock
+    private IssueSyncTaskMapper syncTaskMapper;
+    @Mock
+    private AesGcmCipher aesGcmCipher;
+    @Mock
+    private CloudPlatformClient cloudPlatformClient;
+    @Mock
+    private CloudOperationLogService operationLogService;
+
+    @Test
+    void validateTenantConnectionReturnsMissingWhenConfigNotFound() {
+        when(configMapper.selectOne(any())).thenReturn(null);
+
+        CloudConnectionTestResponse response = cloudConfigService.validateTenantConnection(1001);
+
+        Assertions.assertFalse(response.isSuccess());
+        Assertions.assertEquals(CloudIntegrationConstants.MSG_TENANT_CONFIG_MISSING, response.getMessage());
+        Assertions.assertTrue(response.getNetworkReachable());
+    }
+
+    @Test
+    void validateTenantConnectionReturnsSuccessWhenConfigExists() {
+        IssueCloudConfig config = new IssueCloudConfig();
+        config.setTenantId(1001);
+        config.setStatus(1);
+        when(configMapper.selectOne(any())).thenReturn(config);
+
+        CloudConnectionTestResponse response = cloudConfigService.validateTenantConnection(1001);
+
+        Assertions.assertTrue(response.isSuccess());
+        Assertions.assertEquals(CloudIntegrationConstants.MSG_CONNECTION_SUCCESS, response.getMessage());
+    }
+
+    @Test
+    void testConnectionPropagatesTenantMissingFromCloud() {
+        IssueCloudConfig config = new IssueCloudConfig();
+        config.setId(1L);
+        config.setTenantId(1001);
+        config.setStatus(1);
+        when(configMapper.selectOne(any())).thenReturn(config);
+        when(cloudPlatformClient.testConnection(config)).thenReturn(
+                CloudConnectionTestResult.failure(true, CloudIntegrationConstants.MSG_TENANT_CONFIG_MISSING));
+
+        CloudConnectionTestRequest request = new CloudConnectionTestRequest();
+        request.setTenantId(1001);
+
+        CloudConnectionTestResponse response = cloudConfigService.testConnection(request, "admin", "127.0.0.1");
+
+        Assertions.assertFalse(response.isSuccess());
+        Assertions.assertEquals(CloudIntegrationConstants.MSG_TENANT_CONFIG_MISSING, response.getMessage());
+        Assertions.assertTrue(response.getNetworkReachable());
+        verify(configMapper).updateById(any(IssueCloudConfig.class));
+    }
+}