This commit is contained in:
2026-08-04 09:05:28 +08:00
parent fdb1e8af1c
commit 3afcff320a
116 changed files with 6749 additions and 749 deletions
+5
View File
@@ -16,6 +16,10 @@
| 平台资质上传、替换与查看 | `controller/PlatformQualificationFileController.java`;`POST /{tenantId}` 支持 `qualificationType`、`certificateNo`、`validTo`、`replace`,下载为 `GET /{tenantId}/{fileName}` |
| 服务状态 | `controller/FileHealthController.java` |
| 当前本地磁盘实现 | `service/LocalQualificationStorage.java` |
| 公共导出 API | `controller/PlatformExportTaskController.java`、`service/ExportTaskService.java` |
| 可靠调度 | `service/ExportStateService.java`、`mq/ExportOutboxPublisher.java`、`mq/ExportCommandConsumer.java`、`service/ExportReconciler.java` |
| 通用 Provider/XLSX | `client/ExportProviderClient.java`、`service/GenericXlsxExportRunner.java`、公共契约 `../common-core/.../export/ExportContracts.java` |
| 毒消息隔离 | `mq/ExportDlqHandler.java`;定位必须使用四个 User Properties,不解析损坏 body |
| 未被当前 Controller 使用的旧实现 | `service/QualificationFileStorage.java` |
| IAM 权限协作 | `client/IamQualificationClient.java`;商户登记 `/qualifications`,平台登记/替换 `/platform-qualifications` |
| 目标表结构(尚未接入) | `src/main/resources/db/file.sql` |
@@ -26,6 +30,7 @@
## 修改与验证
- 严格校验租户、文件名/路径、大小、类型和读取权限,防止目录穿越、越权访问和任意文件上传。
- 企业资质文件只接受 PDF、JPG 和 PNG,单个文件上限 5MB;Client、Admin 与 `LocalQualificationStorage` 的限制和提示必须保持一致。
- 生产存储应使用对象存储或共享文件存储;容器本地目录仅适合当前本地实现,不得硬编码环境路径。
- 接入元数据持久化时需同时实现 Entity、Mapper、Repository、事务和迁移验证,并同步检查 IAM 资质流程及平台/商户 Controller。
- 从父目录运行 `mvn -pl file -am test`,并按接口影响验证商户/平台上传、替换、预览、无权限、签名伪造和磁盘故障场景。
+1
View File
@@ -17,6 +17,7 @@
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId></dependency>
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-loadbalancer</artifactId></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency>
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId></dependency>
<dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><scope>runtime</scope></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
</dependencies>
@@ -0,0 +1,47 @@
package com.baiyue.file.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.export.ExportContracts.ExportBatchRequest;
import com.baiyue.common.core.export.ExportContracts.ExportBatchResponse;
import com.baiyue.file.config.ExportDefinitionRegistry.Definition;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@Component
public class ExportProviderClient {
private final Map<String, ExportProviderFeignClient> clients;
private final MeterRegistry meters;
public ExportProviderClient(IamExportProviderClient iam, OrderExportProviderClient order,
WarehouseExportProviderClient warehouse, FirstlegExportProviderClient firstleg,
MeterRegistry meters) {
this.clients = Map.of("iam-service", iam, "order-service", order,
"warehouse-service", warehouse, "firstleg-service", firstleg);
this.meters = meters;
}
public ApiResponse<ExportBatchResponse> batch(String internalToken, String exportType,
Definition definition, ExportBatchRequest request) {
ExportProviderFeignClient client = clients.get(definition.getProviderService());
if (client == null) throw new IllegalStateException("Export provider is not allowlisted: " + definition.getProviderService());
long started = System.nanoTime();
try {
ApiResponse<ExportBatchResponse> response = client.batch(internalToken, exportType, request);
if (response == null || !response.successful()) failure(definition.getProviderService(), exportType);
return response;
} catch (RuntimeException exception) {
failure(definition.getProviderService(), exportType);
throw exception;
} finally {
meters.timer("export_provider_request_seconds", "service", definition.getProviderService(), "type", exportType)
.record(System.nanoTime() - started, TimeUnit.NANOSECONDS);
}
}
private void failure(String service, String type) {
meters.counter("export_provider_failure_total", "service", service, "type", type).increment();
}
}
@@ -0,0 +1,17 @@
package com.baiyue.file.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.export.ExportContracts.ExportBatchRequest;
import com.baiyue.common.core.export.ExportContracts.ExportBatchResponse;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
interface ExportProviderFeignClient {
@PostMapping("/internal/exports/v1/{exportType}/batches")
ApiResponse<ExportBatchResponse> batch(
@RequestHeader("X-Internal-Service-Token") String internalToken,
@PathVariable String exportType,
@RequestBody ExportBatchRequest request);
}
@@ -0,0 +1,31 @@
package com.baiyue.file.client;
import feign.Response;
import feign.codec.Decoder;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.cloud.openfeign.support.SpringDecoder;
import org.springframework.context.annotation.Bean;
import java.io.IOException;
import java.io.InputStream;
public class ExportProviderFeignConfiguration {
static final int MAX_RESPONSE_BYTES = 4 * 1024 * 1024;
@Bean
Decoder exportProviderDecoder(ObjectFactory<HttpMessageConverters> converters) {
Decoder delegate = new SpringDecoder(converters);
return (response, type) -> delegate.decode(withLimitedBody(response), type);
}
private Response withLimitedBody(Response response) throws IOException {
if (response.body() == null) return response;
byte[] bytes;
try (InputStream input = response.body().asInputStream()) {
bytes = input.readNBytes(MAX_RESPONSE_BYTES + 1);
}
if (bytes.length > MAX_RESPONSE_BYTES) throw new ExportProviderResponseTooLargeException(MAX_RESPONSE_BYTES);
return response.toBuilder().body(bytes).build();
}
}
@@ -0,0 +1,7 @@
package com.baiyue.file.client;
public class ExportProviderResponseTooLargeException extends RuntimeException {
public ExportProviderResponseTooLargeException(long limit) {
super("Provider 单批响应超过 " + limit + " 字节");
}
}
@@ -0,0 +1,6 @@
package com.baiyue.file.client;
import org.springframework.cloud.openfeign.FeignClient;
@FeignClient(name="firstleg-service", contextId="firstlegExportProviderClient", url="${baiyue.clients.firstleg.url:}", configuration=ExportProviderFeignConfiguration.class)
interface FirstlegExportProviderClient extends ExportProviderFeignClient { }
@@ -1,34 +0,0 @@
package com.baiyue.file.client;
import com.baiyue.common.core.api.ApiResponse;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestParam;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
@FeignClient(name = "iam-service", contextId = "iamExportClient",
url = "${baiyue.clients.iam.url:}", path = "/internal/iam")
public interface IamExportClient {
@GetMapping("/exports/customers")
ApiResponse<CustomerExportBatch> customers(
@RequestHeader("X-Internal-Service-Token") String internalToken,
@RequestParam(required = false) Long cursor,
@RequestParam(defaultValue = "500") int limit,
@RequestParam(required = false) String keyword,
@RequestParam(required = false) String status);
record CustomerExportBatch(List<CustomerExportRow> records, Long nextCursor, boolean hasMore) { }
record CustomerExportRow(
String id, String tenantCode, String tenantName, String legalName,
String unifiedSocialCreditCode, String enterpriseType, LocalDate establishedDate,
BigDecimal registeredCapital, String registeredCapitalCurrency, String registeredAddress,
String contactName, String contactPhone, String contactEmail, String contactWechat,
String salesOwnerName, String servicePackage, Integer paymentTermDays,
String status, LocalDateTime createdAt, LocalDateTime approvedAt) { }
}
@@ -0,0 +1,6 @@
package com.baiyue.file.client;
import org.springframework.cloud.openfeign.FeignClient;
@FeignClient(name="iam-service", contextId="iamExportProviderClient", url="${baiyue.clients.iam.url:}", configuration=ExportProviderFeignConfiguration.class)
interface IamExportProviderClient extends ExportProviderFeignClient { }
@@ -0,0 +1,6 @@
package com.baiyue.file.client;
import org.springframework.cloud.openfeign.FeignClient;
@FeignClient(name="order-service", contextId="orderExportProviderClient", url="${baiyue.clients.order.url:}", configuration=ExportProviderFeignConfiguration.class)
interface OrderExportProviderClient extends ExportProviderFeignClient { }
@@ -0,0 +1,6 @@
package com.baiyue.file.client;
import org.springframework.cloud.openfeign.FeignClient;
@FeignClient(name="warehouse-service", contextId="warehouseExportProviderClient", url="${baiyue.clients.warehouse.url:}", configuration=ExportProviderFeignConfiguration.class)
interface WarehouseExportProviderClient extends ExportProviderFeignClient { }
@@ -0,0 +1,78 @@
package com.baiyue.file.config;
import jakarta.annotation.PostConstruct;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
@Component
@ConfigurationProperties(prefix = "baiyue.file.export")
public class ExportDefinitionRegistry {
private static final Set<String> SERVICES = Set.of("iam-service", "order-service", "warehouse-service", "firstleg-service");
private static final Pattern TYPE = Pattern.compile("[A-Z0-9_]+");
private static final Pattern PATH = Pattern.compile("/internal/exports/v1/([A-Z0-9_]+)/batches");
private static final Pattern PERMISSION = Pattern.compile("[a-z]+:[a-z-]+:[a-z-]+");
private Map<String, Definition> definitions = Map.of();
@PostConstruct
public void validate() {
if (definitions.isEmpty()) throw new IllegalStateException("At least one export definition is required");
definitions.forEach(this::validateDefinition);
}
public Definition require(String exportType) {
Definition definition = definitions.get(exportType);
if (definition == null) throw new IllegalArgumentException("Unsupported export type: " + exportType);
return definition;
}
private void validateDefinition(String type, Definition d) {
if (!TYPE.matcher(type).matches() || d == null) throw new IllegalStateException("Invalid export type: " + type);
if (!SERVICES.contains(d.providerService)) throw new IllegalStateException("Export provider is not allowlisted: " + d.providerService);
var path = PATH.matcher(d.providerPath == null ? "" : d.providerPath);
if (!path.matches() || !type.equals(path.group(1))) throw new IllegalStateException("Invalid provider path for " + type);
if (d.requiredPermission == null || !PERMISSION.matcher(d.requiredPermission).matches()) throw new IllegalStateException("Invalid export permission for " + type);
if (!Set.of("PLATFORM", "MERCHANT").contains(d.appScope)) throw new IllegalStateException("Invalid app scope for " + type);
if (d.maxRows < 1 || d.maxRows > 500_000 || d.batchSize < 50 || d.batchSize > 1_000 || d.maxAttempts < 1 || d.maxAttempts > 5) {
throw new IllegalStateException("Invalid export limits for " + type);
}
}
public Map<String, Definition> getDefinitions() { return definitions; }
public void setDefinitions(Map<String, Definition> definitions) {
Map<String, Definition> candidate = definitions == null ? Map.of() : Map.copyOf(definitions);
if (candidate.isEmpty()) throw new IllegalStateException("At least one export definition is required");
candidate.forEach(this::validateDefinition);
this.definitions = candidate;
}
public static class Definition {
private String providerService;
private String providerPath;
private String requiredPermission;
private String appScope;
private boolean enabled = true;
private int maxRows = 200_000;
private int maxAttempts = 3;
private int batchSize = 500;
public String getProviderService() { return providerService; }
public void setProviderService(String value) { providerService = value; }
public String getProviderPath() { return providerPath; }
public void setProviderPath(String value) { providerPath = value; }
public String getRequiredPermission() { return requiredPermission; }
public void setRequiredPermission(String value) { requiredPermission = value; }
public String getAppScope() { return appScope; }
public void setAppScope(String value) { appScope = value; }
public boolean isEnabled() { return enabled; }
public void setEnabled(boolean value) { enabled = value; }
public int getMaxRows() { return maxRows; }
public void setMaxRows(int value) { maxRows = value; }
public int getMaxAttempts() { return maxAttempts; }
public void setMaxAttempts(int value) { maxAttempts = value; }
public int getBatchSize() { return batchSize; }
public void setBatchSize(int value) { batchSize = value; }
}
}
@@ -45,6 +45,7 @@ public class PlatformQualificationFileController {
@RequestParam(required = false) LocalDate validTo,
@RequestParam(defaultValue = "false") boolean replace) {
requirePlatform(user);
qualificationType = QualificationTypes.requireSupported(qualificationType);
var stored = storage.store(tenantId, file);
try {
var response = iamClient.submitPlatform(internalToken,
@@ -33,11 +33,12 @@ public class QualificationFileController {
public ApiResponse<UploadResponse> upload(
@AuthenticationPrincipal AuthenticatedUser user,
@RequestPart("file") MultipartFile file,
@RequestParam(defaultValue = "BUSINESS_LICENSE") String qualificationType,
@RequestParam String qualificationType,
@RequestParam(value = "certificateNo", required = false) String certificateNo,
@RequestParam(required = false) LocalDate validTo,
@RequestParam(defaultValue = "false") boolean replace) {
long tenantId = merchantTenant(user);
qualificationType = QualificationTypes.requireSupported(qualificationType);
var stored = storage.store(tenantId, file);
try {
var response = iamClient.submit(internalToken,
@@ -0,0 +1,22 @@
package com.baiyue.file.controller;
import com.baiyue.common.core.exception.BizException;
import java.util.Set;
final class QualificationTypes {
private static final Set<String> SUPPORTED = Set.of(
"BUSINESS_LICENSE",
"LEGAL_PERSON_ID",
"FREIGHT_FORWARDER_CERTIFICATE",
"BANK_ACCOUNT_LICENSE");
private QualificationTypes() { }
static String requireSupported(String qualificationType) {
if (qualificationType == null || !SUPPORTED.contains(qualificationType)) {
throw new BizException("Unsupported qualification type / 不支持的资质类型");
}
return qualificationType;
}
}
@@ -0,0 +1,31 @@
package com.baiyue.file.entity;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Getter;
import lombok.Setter;
import java.time.LocalDateTime;
@Getter
@Setter
@TableName("file_export_outbox")
public class FileExportOutbox {
@TableId private Long id;
private String eventId;
private Long taskId;
private Integer dispatchVersion;
private String topic;
private String tag;
private String messageKey;
private String payloadJson;
private String status;
private Integer attemptCount;
private LocalDateTime nextRetryAt;
private String leaseOwner;
private LocalDateTime leaseUntil;
private LocalDateTime sentAt;
private String lastError;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}
@@ -28,6 +28,14 @@ public class FileExportTask extends BaseEntity {
private String contentType;
private Long sizeBytes;
private String failureMessage;
private Integer dispatchVersion;
private Integer attemptCount;
private Integer maxAttempts;
private LocalDateTime nextRetryAt;
private String leaseOwner;
private LocalDateTime leaseUntil;
private LocalDateTime lastHeartbeatAt;
private LocalDateTime executionDeadlineAt;
private Long requestedBy;
private String requestedUsername;
private LocalDateTime requestedAt;
@@ -0,0 +1,55 @@
package com.baiyue.file.mapper;
import com.baiyue.file.entity.FileExportOutbox;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import java.util.List;
@Mapper
public interface FileExportOutboxMapper extends BaseMapper<FileExportOutbox> {
@Insert("""
INSERT INTO file_export_outbox(id,event_id,task_id,dispatch_version,topic,tag,message_key,payload_json,
status,attempt_count,next_retry_at,created_at,updated_at)
VALUES(#{id},#{eventId},#{taskId},#{dispatchVersion},#{topic},#{tag},#{messageKey},CAST(#{payloadJson} AS JSON),
'PENDING',0,(SELECT next_retry_at FROM file_export_task WHERE id=#{taskId}),CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3))
""")
int insertCommand(FileExportOutbox row);
@Select("SELECT id FROM file_export_outbox WHERE status='PENDING' AND next_retry_at<=CURRENT_TIMESTAMP(3) ORDER BY next_retry_at,id LIMIT #{limit}")
List<Long> publishableIds(@Param("limit") int limit);
@Update("UPDATE file_export_outbox SET status='CLAIMED',lease_owner=#{owner},lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND),attempt_count=attempt_count+1 WHERE id=#{id} AND status='PENDING'")
int claim(@Param("id") long id, @Param("owner") String owner, @Param("leaseSeconds") int leaseSeconds);
@Update("UPDATE file_export_outbox SET status='SENT',sent_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL,last_error=NULL WHERE id=#{id} AND status='CLAIMED' AND lease_owner=#{owner}")
int sent(@Param("id") long id, @Param("owner") String owner);
@Update("UPDATE file_export_outbox SET status='PENDING',next_retry_at=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{backoffSeconds} SECOND),lease_owner=NULL,lease_until=NULL,last_error=#{error} WHERE id=#{id} AND status='CLAIMED' AND lease_owner=#{owner}")
int sendFailed(@Param("id") long id, @Param("owner") String owner,
@Param("backoffSeconds") int backoffSeconds, @Param("error") String error);
@Update("UPDATE file_export_outbox SET status='PENDING',next_retry_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL WHERE status='CLAIMED' AND lease_until<=CURRENT_TIMESTAMP(3) LIMIT 50")
int releaseExpiredClaims();
@Update("""
UPDATE file_export_outbox o JOIN file_export_task t
ON t.id=o.task_id AND t.dispatch_version=o.dispatch_version
SET o.status='PENDING',o.next_retry_at=CURRENT_TIMESTAMP(3),o.lease_owner=NULL,o.lease_until=NULL
WHERE o.status='SENT' AND o.sent_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{staleSeconds} SECOND)
AND t.status='PENDING' AND t.next_retry_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{graceSeconds} SECOND)
LIMIT 20
""")
int redispatchSent(@Param("staleSeconds") int staleSeconds, @Param("graceSeconds") int graceSeconds);
@Update("UPDATE file_export_outbox SET status='QUARANTINED',last_error=#{reason},lease_owner=NULL,lease_until=NULL WHERE event_id=#{eventId} AND task_id=#{taskId} AND dispatch_version=#{dispatchVersion} AND status<>'QUARANTINED'")
int quarantine(@Param("eventId") String eventId, @Param("taskId") long taskId,
@Param("dispatchVersion") int dispatchVersion, @Param("reason") String reason);
@Delete("DELETE FROM file_export_outbox WHERE status='SENT' AND sent_at<DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{retentionDays} DAY) ORDER BY sent_at LIMIT #{limit}")
int deleteRetainedSent(@Param("retentionDays") int retentionDays, @Param("limit") int limit);
}
@@ -3,6 +3,136 @@ package com.baiyue.file.mapper;
import com.baiyue.file.entity.FileExportTask;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import java.util.List;
@Mapper
public interface FileExportTaskMapper extends BaseMapper<FileExportTask> { }
public interface FileExportTaskMapper extends BaseMapper<FileExportTask> {
@Select("SELECT guard_key FROM file_export_task_guard WHERE guard_key='GLOBAL_CREATE' FOR UPDATE")
String lockCreationGuard();
@Select("SELECT guard_key FROM file_export_task_guard WHERE guard_key='GLOBAL_CREATE'")
String findCreationGuard();
@Update("UPDATE file_export_task SET next_retry_at=CURRENT_TIMESTAMP(3) WHERE id=#{id} AND status='PENDING' AND dispatch_version=1")
int initializeDispatchTime(@Param("id") long id);
@Update("""
UPDATE file_export_task SET status='RUNNING', attempt_count=attempt_count+1,
started_at=CURRENT_TIMESTAMP(3), lease_owner=#{owner},
lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND),
last_heartbeat_at=CURRENT_TIMESTAMP(3),
execution_deadline_at=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{durationMinutes} MINUTE),
processed_rows=0,total_rows=NULL,failure_message=NULL
WHERE id=#{id} AND dispatch_version=#{dispatchVersion} AND status='PENDING'
AND next_retry_at<=CURRENT_TIMESTAMP(3)
""")
int claim(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("leaseSeconds") int leaseSeconds,
@Param("durationMinutes") int durationMinutes);
@Update("""
UPDATE file_export_task SET lease_until=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{leaseSeconds} SECOND),
last_heartbeat_at=CURRENT_TIMESTAMP(3) WHERE id=#{id} AND status='RUNNING'
AND dispatch_version=#{dispatchVersion} AND lease_owner=#{owner}
AND execution_deadline_at>CURRENT_TIMESTAMP(3)
""")
int heartbeat(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("leaseSeconds") int leaseSeconds);
@Update("UPDATE file_export_task SET processed_rows=#{rows} WHERE id=#{id} AND status='RUNNING' AND dispatch_version=#{dispatchVersion} AND lease_owner=#{owner}")
int progress(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("rows") long rows);
@Update("""
UPDATE file_export_task SET status='SUCCEEDED',file_name=#{fileName},object_key=#{objectKey},
content_type=#{contentType},size_bytes=#{sizeBytes},total_rows=#{rows},processed_rows=#{rows},
completed_at=CURRENT_TIMESTAMP(3),expires_at=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{retentionDays} DAY),
lease_owner=NULL,lease_until=NULL,failure_message=NULL
WHERE id=#{id} AND status='RUNNING' AND dispatch_version=#{dispatchVersion}
AND lease_owner=#{owner} AND execution_deadline_at>CURRENT_TIMESTAMP(3)
""")
int succeed(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("fileName") String fileName,
@Param("objectKey") String objectKey, @Param("contentType") String contentType,
@Param("sizeBytes") long sizeBytes, @Param("rows") long rows,
@Param("retentionDays") int retentionDays);
@Update("""
UPDATE file_export_task SET status='PENDING',dispatch_version=dispatch_version+1,
next_retry_at=DATE_ADD(CURRENT_TIMESTAMP(3), INTERVAL #{backoffSeconds} SECOND),
lease_owner=NULL,lease_until=NULL,last_heartbeat_at=NULL,execution_deadline_at=NULL,
failure_message=#{message} WHERE id=#{id} AND status='RUNNING'
AND dispatch_version=#{dispatchVersion} AND lease_owner=#{owner} AND attempt_count<max_attempts
""")
int scheduleRetry(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("backoffSeconds") int backoffSeconds,
@Param("message") String message);
@Update("""
UPDATE file_export_task SET status='FAILED',completed_at=CURRENT_TIMESTAMP(3),
lease_owner=NULL,lease_until=NULL,last_heartbeat_at=NULL,failure_message=#{message}
WHERE id=#{id} AND status='RUNNING' AND dispatch_version=#{dispatchVersion} AND lease_owner=#{owner}
""")
int fail(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("owner") String owner, @Param("message") String message);
@Update("""
UPDATE file_export_task SET status='PENDING',dispatch_version=dispatch_version+1,
next_retry_at=CURRENT_TIMESTAMP(3),processed_rows=0,total_rows=NULL,file_name=NULL,object_key=NULL,
content_type=NULL,size_bytes=NULL,failure_message=NULL,started_at=NULL,completed_at=NULL,expires_at=NULL,
lease_owner=NULL,lease_until=NULL,last_heartbeat_at=NULL,execution_deadline_at=NULL
WHERE id=#{id} AND status='FAILED'
""")
int manualRetry(@Param("id") long id);
@Update("UPDATE file_export_task SET status='FAILED',completed_at=CURRENT_TIMESTAMP(3),failure_message=#{reason},lease_owner=NULL,lease_until=NULL WHERE id=#{id} AND dispatch_version=#{dispatchVersion} AND status='PENDING'")
int failQuarantined(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
@Param("reason") String reason);
@Update("""
UPDATE file_export_task SET status='PENDING',dispatch_version=dispatch_version+1,
next_retry_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL,last_heartbeat_at=NULL,
execution_deadline_at=NULL,failure_message='执行实例失联,任务已重新派发'
WHERE id=#{id} AND status='RUNNING' AND dispatch_version=#{dispatchVersion}
AND lease_until<=CURRENT_TIMESTAMP(3) AND attempt_count<max_attempts
""")
int recoverStale(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion);
@Update("""
UPDATE file_export_task SET status='FAILED',completed_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,
lease_until=NULL,last_heartbeat_at=NULL,failure_message='执行实例失联且已达最大尝试次数'
WHERE id=#{id} AND status='RUNNING' AND dispatch_version=#{dispatchVersion}
AND lease_until<=CURRENT_TIMESTAMP(3) AND attempt_count>=max_attempts
""")
int failStale(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion);
@Select("SELECT * FROM file_export_task WHERE status='SUCCEEDED' AND expires_at<=CURRENT_TIMESTAMP(3) AND object_key IS NOT NULL ORDER BY expires_at LIMIT 50")
List<FileExportTask> expiredArtifactsDb();
@Select("SELECT COUNT(*) FROM file_export_task WHERE tenant_id=#{tenantId} AND app_code=#{appCode} AND requested_by=#{requestedBy} AND status IN ('PENDING','RUNNING')")
long activeForUser(@Param("tenantId") long tenantId,@Param("appCode") String appCode,@Param("requestedBy") long requestedBy);
@Select("SELECT COUNT(*) FROM file_export_task WHERE export_type=#{exportType} AND status IN ('PENDING','RUNNING')")
long activeForType(@Param("exportType") String exportType);
@Select("SELECT * FROM file_export_task WHERE tenant_id=#{tenantId} AND app_code=#{appCode} AND requested_by=#{requestedBy} AND export_type=#{exportType} AND parameters_json=CAST(#{parametersJson} AS JSON) AND status IN ('PENDING','RUNNING') AND requested_at>=DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 60 SECOND) ORDER BY requested_at DESC LIMIT 1")
FileExportTask recentDuplicate(@Param("tenantId") long tenantId,@Param("appCode") String appCode,@Param("requestedBy") long requestedBy,@Param("exportType") String exportType,@Param("parametersJson") String parametersJson);
@Select("SELECT COUNT(*) FROM file_export_task WHERE id=#{id} AND status='SUCCEEDED' AND expires_at>CURRENT_TIMESTAMP(3)")
int downloadable(@Param("id") long id);
@Select("SELECT COUNT(*) FROM file_export_task WHERE status='SUCCEEDED' AND object_key=#{objectKey}")
int succeededObjectExists(@Param("objectKey") String objectKey);
@Select("SELECT * FROM file_export_task WHERE status='RUNNING' AND lease_until<=CURRENT_TIMESTAMP(3) ORDER BY lease_until LIMIT 20")
List<FileExportTask> staleRunning();
@Select("SELECT * FROM file_export_task WHERE status='PENDING' AND next_retry_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{graceSeconds} SECOND) AND NOT EXISTS (SELECT 1 FROM file_export_outbox o WHERE o.task_id=file_export_task.id AND o.dispatch_version=file_export_task.dispatch_version) ORDER BY id LIMIT 50")
List<FileExportTask> missingOutbox(@Param("graceSeconds") int graceSeconds);
@Update("UPDATE file_export_task SET status='EXPIRED' WHERE id=#{id} AND status='SUCCEEDED'")
int markExpiredDb(@Param("id") long id);
}
@@ -0,0 +1,52 @@
package com.baiyue.file.mq;
import com.baiyue.common.core.export.ExportContracts.ExportCommand;
import com.baiyue.file.service.ExportTaskExecutor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
@Service
@ConditionalOnProperty(prefix="baiyue.file.export.consumer", name="enabled", havingValue="true")
@RocketMQMessageListener(topic="${baiyue.file.export.rocketmq.topic}", consumerGroup="${baiyue.file.export.rocketmq.consumer-group}",
selectorExpression="EXPORT_TASK", messageModel= MessageModel.CLUSTERING, consumeThreadNumber=4,
consumeThreadMax=4, consumeTimeout=90, maxReconsumeTimes=2)
public class ExportCommandConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
private final ObjectMapper objectMapper;
private final ExportTaskExecutor executor;
private final String instance;
private final int concurrency;
private final long consumeTimeoutMinutes;
private final int maxReconsumeTimes;
public ExportCommandConsumer(ObjectMapper objectMapper, ExportTaskExecutor executor,
@Value("${baiyue.file.export.instance-id:${HOSTNAME:${random.uuid}}}") String instance,
@Value("${baiyue.file.export.consumer.concurrency:4}") int concurrency,
@Value("${baiyue.file.export.consumer.consume-timeout-minutes:90}") long consumeTimeoutMinutes,
@Value("${baiyue.file.export.consumer.max-reconsume-times:2}") int maxReconsumeTimes) {
this.objectMapper=objectMapper;this.executor=executor;this.instance=instance;
if (concurrency < 1 || concurrency > 64) throw new IllegalArgumentException("Export consumer concurrency must be between 1 and 64");
this.concurrency=concurrency;this.consumeTimeoutMinutes=consumeTimeoutMinutes;this.maxReconsumeTimes=maxReconsumeTimes;
}
@Override public void prepareStart(DefaultMQPushConsumer consumer) {
consumer.setConsumeThreadMin(concurrency);
consumer.setConsumeThreadMax(concurrency);
consumer.setConsumeTimeout(consumeTimeoutMinutes);
consumer.setMaxReconsumeTimes(maxReconsumeTimes);
}
@Override public void onMessage(MessageExt message) {
final ExportCommand command;
try { command = objectMapper.readValue(new String(message.getBody(), StandardCharsets.UTF_8), ExportCommand.class); }
catch (Exception exception) { throw new IllegalArgumentException("Malformed export command", exception); }
if (command.schemaVersion()!=1 || command.taskId()==null || command.eventId()==null) throw new IllegalArgumentException("Unsupported export command");
executor.execute(Long.parseLong(command.taskId()), command.dispatchVersion(), instance);
}
}
@@ -0,0 +1,36 @@
package com.baiyue.file.mq;
import com.baiyue.file.service.ExportStateService;
import io.micrometer.core.instrument.MeterRegistry;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
@Service
@ConditionalOnProperty(prefix="baiyue.file.export.consumer", name="enabled", havingValue="true")
@RocketMQMessageListener(topic="${baiyue.file.export.rocketmq.dlq-topic}",
consumerGroup="${baiyue.file.export.rocketmq.dlq-consumer-group}", messageModel=MessageModel.CLUSTERING,
consumeThreadNumber=1, consumeThreadMax=1, maxReconsumeTimes=Integer.MAX_VALUE)
public class ExportDlqHandler implements RocketMQListener<MessageExt> {
private static final Logger LOGGER = LoggerFactory.getLogger(ExportDlqHandler.class);
private final ExportStateService states;
private final MeterRegistry meters;
public ExportDlqHandler(ExportStateService states, MeterRegistry meters) { this.states=states; this.meters=meters; }
@Override public void onMessage(MessageExt message) {
meters.counter("export_dlq_messages").increment();
String eventId=message.getUserProperty("eventId"), taskId=message.getUserProperty("taskId");
String version=message.getUserProperty("dispatchVersion"), schema=message.getUserProperty("schemaVersion");
if (blank(eventId)||blank(taskId)||blank(version)||blank(schema)) {
LOGGER.error("CRITICAL: 畸形 DLQ 消息缺少定位属性 messageId={}", message.getMsgId());
return;
}
try { states.quarantine(eventId, Long.parseLong(taskId), Integer.parseInt(version), "消息契约损坏,已自动隔离(schemaVersion="+schema+")"); }
catch (NumberFormatException exception) { LOGGER.error("CRITICAL: 畸形 DLQ 消息定位属性非法 messageId={}", message.getMsgId(), exception); }
}
private boolean blank(String value) { return value==null||value.isBlank(); }
}
@@ -0,0 +1,54 @@
package com.baiyue.file.mq;
import com.baiyue.file.entity.FileExportOutbox;
import com.baiyue.file.repository.ExportOutboxRepository;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
@Service
@ConditionalOnProperty(prefix="baiyue.file.export.publisher", name="enabled", havingValue="true")
public class ExportOutboxPublisher {
private static final Logger LOGGER = LoggerFactory.getLogger(ExportOutboxPublisher.class);
private static final int[] BACKOFF = {5, 15, 60, 300, 900};
private final ExportOutboxRepository outbox;
private final RocketMQTemplate rocketMQ;
private final String owner;
private final int batchSize;
private final int leaseSeconds;
public ExportOutboxPublisher(ExportOutboxRepository outbox, RocketMQTemplate rocketMQ,
@Value("${baiyue.file.export.instance-id:${HOSTNAME:${random.uuid}}}") String owner,
@Value("${baiyue.file.export.publisher.batch-size:50}") int batchSize,
@Value("${baiyue.file.export.publisher.lease-seconds:60}") int leaseSeconds) {
this.outbox=outbox;this.rocketMQ=rocketMQ;this.owner=owner;this.batchSize=batchSize;this.leaseSeconds=leaseSeconds;
}
@Scheduled(fixedDelayString="${baiyue.file.export.publisher.poll-interval-ms:5000}")
public void publish() {
for (Long id : outbox.publishableIds(batchSize)) {
if (!outbox.claim(id, owner, leaseSeconds)) continue;
FileExportOutbox row = outbox.find(id);
try {
var message = MessageBuilder.withPayload(row.getPayloadJson())
.setHeader(RocketMQHeaders.KEYS, row.getMessageKey())
.setHeader("eventId", row.getEventId())
.setHeader("taskId", String.valueOf(row.getTaskId()))
.setHeader("dispatchVersion", String.valueOf(row.getDispatchVersion()))
.setHeader("schemaVersion", "1").build();
rocketMQ.syncSend(row.getTopic() + ":" + row.getTag(), message, 3000);
outbox.sent(id, owner);
} catch (RuntimeException exception) {
int attempt = Math.max(1, row.getAttemptCount());
int seconds = BACKOFF[Math.min(attempt - 1, BACKOFF.length - 1)];
outbox.sendFailed(id, owner, seconds, safe(exception.getMessage()));
LOGGER.error("导出命令发送失败 eventId={} taskId={}", row.getEventId(), row.getTaskId(), exception);
}
}
}
private String safe(String value) { String message=value==null?"RocketMQ send failed":value; return message.length()>1000?message.substring(0,1000):message; }
}
@@ -0,0 +1,27 @@
package com.baiyue.file.repository;
import com.baiyue.file.entity.FileExportOutbox;
import com.baiyue.file.mapper.FileExportOutboxMapper;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public class ExportOutboxRepository {
private final FileExportOutboxMapper mapper;
public ExportOutboxRepository(FileExportOutboxMapper mapper) { this.mapper = mapper; }
public boolean insert(FileExportOutbox row) {
try { return mapper.insertCommand(row) == 1; }
catch (DuplicateKeyException ignored) { return false; }
}
public FileExportOutbox find(long id) { return mapper.selectById(id); }
public List<Long> publishableIds(int limit) { return mapper.publishableIds(limit); }
public boolean claim(long id, String owner, int leaseSeconds) { return mapper.claim(id, owner, leaseSeconds) == 1; }
public boolean sent(long id, String owner) { return mapper.sent(id, owner) == 1; }
public boolean sendFailed(long id, String owner, int seconds, String error) { return mapper.sendFailed(id, owner, seconds, error) == 1; }
public int releaseExpiredClaims() { return mapper.releaseExpiredClaims(); }
public int redispatchSent(int staleSeconds, int graceSeconds) { return mapper.redispatchSent(staleSeconds, graceSeconds); }
public boolean quarantine(String eventId, long taskId, int version, String reason) { return mapper.quarantine(eventId, taskId, version, reason) == 1; }
public int deleteRetainedSent(int retentionDays, int limit) { return mapper.deleteRetainedSent(retentionDays, limit); }
}
@@ -6,6 +6,7 @@ import com.baiyue.file.mapper.FileExportTaskMapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import jakarta.annotation.PostConstruct;
import org.springframework.stereotype.Repository;
import org.springframework.util.StringUtils;
@@ -21,7 +22,60 @@ public class ExportTaskRepository {
this.mapper = mapper;
}
@PostConstruct
void validateCreationGuard() {
if (!"GLOBAL_CREATE".equals(mapper.findCreationGuard())) {
throw new IllegalStateException("Missing GLOBAL_CREATE export task guard; apply file.sql before starting File Service");
}
}
public void lockCreationGuard() {
if (!"GLOBAL_CREATE".equals(mapper.lockCreationGuard())) {
throw new IllegalStateException("Unable to acquire export task creation guard");
}
}
public void insert(FileExportTask task) { mapper.insert(task); }
public void initializeDispatchTime(long id) { mapper.initializeDispatchTime(id); }
public boolean claim(long id, int dispatchVersion, String owner, int leaseSeconds, int durationMinutes) {
return mapper.claim(id, dispatchVersion, owner, leaseSeconds, durationMinutes) == 1;
}
public boolean heartbeat(long id, int dispatchVersion, String owner, int leaseSeconds) {
return mapper.heartbeat(id, dispatchVersion, owner, leaseSeconds) == 1;
}
public boolean progress(long id, int dispatchVersion, String owner, long rows) {
return mapper.progress(id, dispatchVersion, owner, rows) == 1;
}
public boolean succeed(long id, int dispatchVersion, String owner, String fileName, String objectKey,
String contentType, long sizeBytes, long rows, int retentionDays) {
return mapper.succeed(id, dispatchVersion, owner, fileName, objectKey, contentType,
sizeBytes, rows, retentionDays) == 1;
}
public boolean scheduleRetry(long id, int dispatchVersion, String owner, int backoffSeconds, String message) {
return mapper.scheduleRetry(id, dispatchVersion, owner, backoffSeconds, message) == 1;
}
public boolean fail(long id, int dispatchVersion, String owner, String message) {
return mapper.fail(id, dispatchVersion, owner, message) == 1;
}
public boolean manualRetry(long id) { return mapper.manualRetry(id) == 1; }
public boolean failQuarantined(long id, int dispatchVersion, String reason) { return mapper.failQuarantined(id, dispatchVersion, reason) == 1; }
public boolean recoverStale(long id, int dispatchVersion) { return mapper.recoverStale(id, dispatchVersion) == 1; }
public boolean failStale(long id, int dispatchVersion) { return mapper.failStale(id, dispatchVersion) == 1; }
public List<FileExportTask> expiredArtifactsDb() { return mapper.expiredArtifactsDb(); }
public long activeForUser(long tenantId,String appCode,long requestedBy){return mapper.activeForUser(tenantId,appCode,requestedBy);}
public long activeForType(String exportType){return mapper.activeForType(exportType);}
public FileExportTask recentDuplicate(long tenantId,String appCode,long requestedBy,String exportType,String json){return mapper.recentDuplicate(tenantId,appCode,requestedBy,exportType,json);}
public boolean downloadable(long id){return mapper.downloadable(id)>0;}
public boolean succeededObjectExists(String objectKey){return mapper.succeededObjectExists(objectKey)>0;}
public List<FileExportTask> staleRunning() { return mapper.staleRunning(); }
public List<FileExportTask> missingOutbox(int graceSeconds) { return mapper.missingOutbox(graceSeconds); }
public FileExportTask find(long id) { return mapper.selectById(id); }
@@ -43,86 +97,6 @@ public class ExportTaskRepository {
return PageResult.from(mapper.selectPage(new Page<>(page, size), query));
}
public Long firstPendingId() {
FileExportTask task = mapper.selectOne(new LambdaQueryWrapper<FileExportTask>()
.select(FileExportTask::getId)
.eq(FileExportTask::getStatus, "PENDING")
.orderByAsc(FileExportTask::getRequestedAt)
.orderByAsc(FileExportTask::getId)
.last("LIMIT 1"));
return task == null ? null : task.getId();
}
public boolean claim(long id, LocalDateTime now) {
return mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "PENDING")
.set(FileExportTask::getStatus, "RUNNING")
.set(FileExportTask::getStartedAt, now)
.set(FileExportTask::getFailureMessage, null)) == 1;
}
public void progress(long id, long rows) {
mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "RUNNING")
.set(FileExportTask::getProcessedRows, rows));
}
public boolean succeed(long id, String fileName, String objectKey, String contentType,
long sizeBytes, long rows, LocalDateTime completedAt, LocalDateTime expiresAt) {
return mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "RUNNING")
.set(FileExportTask::getStatus, "SUCCEEDED")
.set(FileExportTask::getFileName, fileName)
.set(FileExportTask::getObjectKey, objectKey)
.set(FileExportTask::getContentType, contentType)
.set(FileExportTask::getSizeBytes, sizeBytes)
.set(FileExportTask::getTotalRows, rows)
.set(FileExportTask::getProcessedRows, rows)
.set(FileExportTask::getCompletedAt, completedAt)
.set(FileExportTask::getExpiresAt, expiresAt)) == 1;
}
public void fail(long id, String message, LocalDateTime completedAt) {
mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "RUNNING")
.set(FileExportTask::getStatus, "FAILED")
.set(FileExportTask::getFailureMessage, message)
.set(FileExportTask::getCompletedAt, completedAt));
}
public boolean retry(long id) {
return mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "FAILED")
.set(FileExportTask::getStatus, "PENDING")
.set(FileExportTask::getProcessedRows, 0L)
.set(FileExportTask::getTotalRows, null)
.set(FileExportTask::getFailureMessage, null)
.set(FileExportTask::getStartedAt, null)
.set(FileExportTask::getCompletedAt, null)) == 1;
}
public int requeueStale(LocalDateTime cutoff) {
return mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getStatus, "RUNNING")
.lt(FileExportTask::getUpdatedAt, cutoff)
.set(FileExportTask::getStatus, "PENDING")
.set(FileExportTask::getStartedAt, null)
.set(FileExportTask::getFailureMessage, "Previous worker stopped before completion; task requeued"));
}
public List<FileExportTask> expiredArtifacts(LocalDateTime now) {
return mapper.selectList(new LambdaQueryWrapper<FileExportTask>()
.eq(FileExportTask::getStatus, "SUCCEEDED")
.le(FileExportTask::getExpiresAt, now)
.isNotNull(FileExportTask::getObjectKey)
.last("LIMIT 50"));
}
public void markExpired(long id) {
mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
@@ -1,168 +0,0 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.file.client.IamExportClient;
import com.baiyue.file.entity.FileExportTask;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.poi.ss.usermodel.CellStyle;
import org.apache.poi.ss.usermodel.FillPatternType;
import org.apache.poi.ss.usermodel.IndexedColors;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.xssf.streaming.SXSSFWorkbook;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.Set;
@Service
public class CustomerExportTaskHandler implements ExportTaskHandler {
static final String CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final List<String> HEADERS = List.of(
"客户ID", "客户编码", "公司名称", "营业执照企业名称", "统一社会信用代码", "企业类型",
"成立日期", "注册资本", "币种", "注册地址", "联系人", "联系电话", "联系邮箱", "联系微信",
"销售人员", "客户套餐", "账期(天)", "状态", "入驻时间", "审批时间");
private final IamExportClient iamClient;
private final LocalExportStorage storage;
private final ObjectMapper objectMapper;
private final String internalToken;
public CustomerExportTaskHandler(IamExportClient iamClient, LocalExportStorage storage,
ObjectMapper objectMapper,
@Value("${baiyue.security.internal-service-token}") String internalToken) {
this.iamClient = iamClient;
this.storage = storage;
this.objectMapper = objectMapper;
this.internalToken = internalToken;
}
@Override public String exportType() { return "CUSTOMER_LIST"; }
@Override public String requiredPermission() { return "platform:customer:export"; }
@Override
public void validateParameters(Map<String, Object> parameters) {
if (parameters == null) return;
Object keyword = parameters.get("keyword");
if (keyword != null && String.valueOf(keyword).length() > 128) {
throw new BizException(ResultCode.VALIDATION_FAILED, "客户导出关键词不能超过 128 个字符");
}
Object status = parameters.get("status");
if (status != null && !String.valueOf(status).isBlank()
&& !Set.of("PROFILE_REQUIRED", "PENDING", "ACTIVE", "REJECTED", "DISABLED")
.contains(String.valueOf(status))) {
throw new BizException(ResultCode.VALIDATION_FAILED, "客户导出状态无效");
}
}
@Override
public ExportResult export(FileExportTask task, ProgressReporter progress) {
CustomerParameters parameters = parameters(task.getParametersJson());
LocalExportStorage.PendingFile target = storage.create(task.getId());
long rows = 0;
try (target; SXSSFWorkbook workbook = new SXSSFWorkbook(100)) {
workbook.setCompressTempFiles(true);
Sheet sheet = workbook.createSheet("客户列表");
writeHeader(workbook, sheet);
Long cursor = null;
boolean hasMore;
do {
var response = iamClient.customers(internalToken, cursor, 500,
blankToNull(parameters.keyword()), blankToNull(parameters.status()));
if (response == null || !response.successful() || response.data() == null) {
throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "IAM customer export is unavailable");
}
var batch = response.data();
for (var item : batch.records()) writeRow(sheet.createRow(Math.toIntExact(++rows)), item);
progress.update(rows);
hasMore = batch.hasMore();
if (hasMore && (batch.nextCursor() == null || batch.nextCursor().equals(cursor))) {
throw new IllegalStateException("IAM customer export cursor did not advance");
}
cursor = batch.nextCursor();
} while (hasMore);
for (int index = 0; index < HEADERS.size(); index++) sheet.setColumnWidth(index, columnWidth(index));
workbook.write(target.output());
} catch (IOException exception) {
storage.delete(target.objectKey());
throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "客户导出文件生成失败");
} catch (RuntimeException exception) {
storage.delete(target.objectKey());
throw exception;
}
String fileName = "客户列表-" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss")) + ".xlsx";
return new ExportResult(fileName, target.objectKey(), CONTENT_TYPE,
storage.size(target.objectKey()), rows);
}
private CustomerParameters parameters(String json) {
try { return objectMapper.readValue(json, CustomerParameters.class); }
catch (Exception exception) { throw new BizException(ResultCode.BAD_REQUEST, "客户导出筛选条件无效"); }
}
private void writeHeader(Workbook workbook, Sheet sheet) {
CellStyle style = workbook.createCellStyle();
style.setFillForegroundColor(IndexedColors.GREY_25_PERCENT.getIndex());
style.setFillPattern(FillPatternType.SOLID_FOREGROUND);
var font = workbook.createFont();
font.setBold(true);
style.setFont(font);
Row row = sheet.createRow(0);
for (int index = 0; index < HEADERS.size(); index++) {
var cell = row.createCell(index);
cell.setCellValue(HEADERS.get(index));
cell.setCellStyle(style);
}
sheet.createFreezePane(0, 1);
}
private void writeRow(Row row, IamExportClient.CustomerExportRow item) {
int column = 0;
cell(row, column++, item.id()); cell(row, column++, item.tenantCode());
cell(row, column++, item.tenantName()); cell(row, column++, item.legalName());
cell(row, column++, item.unifiedSocialCreditCode()); cell(row, column++, item.enterpriseType());
cell(row, column++, item.establishedDate()); cell(row, column++, item.registeredCapital());
cell(row, column++, item.registeredCapitalCurrency()); cell(row, column++, item.registeredAddress());
cell(row, column++, item.contactName()); cell(row, column++, item.contactPhone());
cell(row, column++, item.contactEmail()); cell(row, column++, item.contactWechat());
cell(row, column++, item.salesOwnerName()); cell(row, column++, item.servicePackage());
cell(row, column++, item.paymentTermDays()); cell(row, column++, statusLabel(item.status()));
cell(row, column++, item.createdAt()); cell(row, column, item.approvedAt());
}
private void cell(Row row, int column, Object value) {
row.createCell(column).setCellValue(value == null ? "" : String.valueOf(value));
}
private int columnWidth(int index) {
return switch (index) {
case 2, 3, 9 -> 28 * 256;
case 4, 12 -> 24 * 256;
case 18, 19 -> 21 * 256;
default -> 16 * 256;
};
}
private String statusLabel(String status) {
if (status == null) return "";
return switch (status) {
case "PROFILE_REQUIRED" -> "待完善";
case "PENDING" -> "待审核";
case "ACTIVE" -> "启用";
case "REJECTED" -> "已驳回";
case "DISABLED" -> "已停用";
default -> status;
};
}
private String blankToNull(String value) { return value == null || value.isBlank() ? null : value.trim(); }
public record CustomerParameters(String keyword, String status) { }
}
@@ -0,0 +1,40 @@
package com.baiyue.file.service;
import com.baiyue.common.core.export.ExportContracts.ExportCommand;
import com.baiyue.file.entity.FileExportOutbox;
import com.baiyue.file.entity.FileExportTask;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.time.OffsetDateTime;
import java.util.UUID;
@Component
public class ExportCommandFactory {
private final ObjectMapper objectMapper;
private final String topic;
public ExportCommandFactory(ObjectMapper objectMapper,
@Value("${baiyue.file.export.rocketmq.topic:test-baiyue-file-export-command-v1}") String topic) {
this.objectMapper = objectMapper;
this.topic = topic;
}
public FileExportOutbox create(FileExportTask task, int dispatchVersion) {
String eventId = UUID.randomUUID().toString();
ExportCommand command = new ExportCommand(1, eventId, String.valueOf(task.getId()), task.getExportType(),
dispatchVersion, OffsetDateTime.now(), MDC.get("traceId"));
FileExportOutbox row = new FileExportOutbox();
row.setId(IdWorker.getId()); row.setEventId(eventId); row.setTaskId(task.getId());
row.setDispatchVersion(dispatchVersion); row.setTopic(topic); row.setTag("EXPORT_TASK");
row.setMessageKey(String.valueOf(task.getId())); row.setPayloadJson(json(command));
row.setStatus("PENDING"); row.setAttemptCount(0);
return row;
}
private String json(ExportCommand command) {
try { return objectMapper.writeValueAsString(command); }
catch (JsonProcessingException exception) { throw new IllegalStateException("Unable to serialize export command", exception); }
}
}
@@ -0,0 +1,10 @@
package com.baiyue.file.service;
public class ExportFailure extends RuntimeException {
private final boolean retryable;
public ExportFailure(String message, boolean retryable) { super(message); this.retryable = retryable; }
public ExportFailure(String message, boolean retryable, Throwable cause) { super(message, cause); this.retryable = retryable; }
public boolean retryable() { return retryable; }
public static ExportFailure temporary(String message, Throwable cause) { return new ExportFailure(message, true, cause); }
public static ExportFailure permanent(String message) { return new ExportFailure(message, false); }
}
@@ -0,0 +1,22 @@
package com.baiyue.file.service;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
@Component
public class ExportMetrics {
private final MeterRegistry registry;
public ExportMetrics(MeterRegistry registry, JdbcTemplate jdbc) {
this.registry=registry;
Gauge.builder("export_task_queue_age_seconds",jdbc,j->number(j,"SELECT COALESCE(TIMESTAMPDIFF(SECOND,MIN(requested_at),CURRENT_TIMESTAMP(3)),0) FROM file_export_task WHERE status='PENDING'")).register(registry);
Gauge.builder("export_outbox_pending",jdbc,j->number(j,"SELECT COUNT(*) FROM file_export_outbox WHERE status IN ('PENDING','CLAIMED')")).register(registry);
Gauge.builder("export_outbox_oldest_age_seconds",jdbc,j->number(j,"SELECT COALESCE(TIMESTAMPDIFF(SECOND,MIN(created_at),CURRENT_TIMESTAMP(3)),0) FROM file_export_outbox WHERE status IN ('PENDING','CLAIMED')")).register(registry);
Gauge.builder("export_outbox_max_attempt",jdbc,j->number(j,"SELECT COALESCE(MAX(attempt_count),0) FROM file_export_outbox WHERE status IN ('PENDING','CLAIMED')")).register(registry);
Gauge.builder("export_storage_bytes",jdbc,j->number(j,"SELECT COALESCE(SUM(size_bytes),0) FROM file_export_task WHERE status='SUCCEEDED' AND expires_at>CURRENT_TIMESTAMP(3)")).register(registry);
}
public void created(String type){registry.counter("export_task_created_total","type",type).increment();}
public void completed(String type,String status){registry.counter("export_task_completed_total","type",type,"status",status).increment();}
private static double number(JdbcTemplate jdbc,String sql){try{Number value=jdbc.queryForObject(sql,Number.class);return value==null?0:value.doubleValue();}catch(RuntimeException ignored){return 0;}}
}
@@ -0,0 +1,51 @@
package com.baiyue.file.service;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportOutboxRepository;
import com.baiyue.file.repository.ExportTaskRepository;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
@Service
@ConditionalOnProperty(prefix="baiyue.file.export.reconciler", name="enabled", havingValue="true")
public class ExportReconciler {
private static final Logger LOGGER= LoggerFactory.getLogger(ExportReconciler.class);
private final ExportTaskRepository tasks; private final ExportOutboxRepository outbox;
private final ExportStateService states; private final ExportStorage storage; private final MeterRegistry meters;
private final int graceSeconds, sentStaleSeconds;
private final int retentionDays;
public ExportReconciler(ExportTaskRepository tasks, ExportOutboxRepository outbox, ExportStateService states,
ExportStorage storage, MeterRegistry meters,
@Value("${baiyue.file.export.reconciler.redispatch-grace-seconds:60}") int graceSeconds,
@Value("${baiyue.file.export.reconciler.sent-stale-seconds:600}") int sentStaleSeconds,
@Value("${baiyue.file.export.retention-days:7}") int retentionDays) {
this.tasks=tasks;this.outbox=outbox;this.states=states;this.storage=storage;this.meters=meters;
this.graceSeconds=graceSeconds;this.sentStaleSeconds=sentStaleSeconds;this.retentionDays=retentionDays;
}
@Scheduled(cron="${baiyue.file.export.reconciler.outbox-cleanup-cron:0 15 3 * * *}")
public void cleanupRetainedOutbox() {
count("outbox_clean",outbox.deleteRetainedSent(retentionDays,5000));
}
@Scheduled(fixedDelayString="${baiyue.file.export.reconciler.poll-interval-ms:30000}")
public void reconcile() {
int released=outbox.releaseExpiredClaims(); count("outbox_release",released);
int redispatched=outbox.redispatchSent(sentStaleSeconds,graceSeconds); count("redispatch",redispatched);
int missing=0; for(FileExportTask task:tasks.missingOutbox(graceSeconds)) if(states.ensureOutbox(task)) missing++;
count("redispatch",missing);
int recovered=0; for(FileExportTask task:tasks.staleRunning()) if(states.recoverStale(task)) recovered++;
count("lease_recover",recovered);
int expired=0; for(FileExportTask task:tasks.expiredArtifactsDb()) {
if(storage.delete(task.getObjectKey())) { tasks.markExpired(task.getId()); expired++; }
else LOGGER.warn("导出文件过期清理失败 taskId={}",task.getId());
}
count("expire",expired);
count("orphan_clean", storage.cleanupOrphans(Duration.ofHours(24), tasks::succeededObjectExists));
}
private void count(String action,int value){ if(value>0)meters.counter("export_reconciler_action_total","action",action).increment(value); }
}
@@ -0,0 +1,98 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportOutboxRepository;
import com.baiyue.file.repository.ExportTaskRepository;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@Service
public class ExportStateService {
private final ExportTaskRepository tasks;
private final ExportOutboxRepository outbox;
private final ExportCommandFactory commands;
private final ExportMetrics metrics;
public ExportStateService(ExportTaskRepository tasks, ExportOutboxRepository outbox, ExportCommandFactory commands,
ExportMetrics metrics) {
this.tasks = tasks; this.outbox = outbox; this.commands = commands; this.metrics = metrics;
}
@Transactional
public CreationResult createConstrained(FileExportTask task) {
tasks.lockCreationGuard();
FileExportTask duplicate = tasks.recentDuplicate(task.getTenantId(), task.getAppCode(), task.getRequestedBy(),
task.getExportType(), task.getParametersJson());
if (duplicate != null) return new CreationResult(duplicate, false);
if (tasks.activeForUser(task.getTenantId(), task.getAppCode(), task.getRequestedBy()) >= 3) {
throw new BizException(ResultCode.TOO_MANY_REQUESTS, "同一用户最多同时执行 3 个导出任务");
}
if (tasks.activeForType(task.getExportType()) >= 5) {
throw new BizException(ResultCode.TOO_MANY_REQUESTS, "该导出类型当前任务较多,请稍后重试");
}
tasks.insert(task);
tasks.initializeDispatchTime(task.getId());
if (!outbox.insert(commands.create(task, task.getDispatchVersion()))) {
throw new IllegalStateException("Duplicate export command while creating task");
}
return new CreationResult(task, true);
}
@Transactional
public boolean retryAfterFailure(FileExportTask current, String owner, int seconds, String message) {
if (!tasks.scheduleRetry(current.getId(), current.getDispatchVersion(), owner, seconds, safe(message))) return false;
FileExportTask refreshed = tasks.find(current.getId());
if (!outbox.insert(commands.create(refreshed, refreshed.getDispatchVersion()))) {
throw new IllegalStateException("Unable to create retry command");
}
return true;
}
@Transactional
public boolean fail(FileExportTask task, String owner, String message) {
return tasks.fail(task.getId(), task.getDispatchVersion(), owner, safe(message));
}
@Transactional
public boolean ensureOutbox(FileExportTask task) {
return outbox.insert(commands.create(task, task.getDispatchVersion()));
}
@Transactional
public boolean manualRetry(FileExportTask task) {
if (!tasks.manualRetry(task.getId())) return false;
FileExportTask refreshed = tasks.find(task.getId());
if (!outbox.insert(commands.create(refreshed, refreshed.getDispatchVersion()))) {
throw new IllegalStateException("Unable to create manual retry command");
}
return true;
}
@Transactional
public void quarantine(String eventId, long taskId, int version, String reason) {
FileExportTask current = tasks.find(taskId);
if (current == null || current.getDispatchVersion() != version || !"PENDING".equals(current.getStatus())) return;
boolean changed = outbox.quarantine(eventId, taskId, version, safe(reason));
if (changed) {
if (!tasks.failQuarantined(taskId, version, safe(reason))) {
throw new IllegalStateException("Unable to fail quarantined export task");
}
metrics.completed(current.getExportType(), "FAILED");
}
}
@Transactional
public boolean recoverStale(FileExportTask task) {
if (task.getAttemptCount() >= task.getMaxAttempts()) {
boolean failed = tasks.failStale(task.getId(), task.getDispatchVersion());
if (failed) metrics.completed(task.getExportType(), "FAILED");
return failed;
}
if (!tasks.recoverStale(task.getId(), task.getDispatchVersion())) return false;
FileExportTask refreshed = tasks.find(task.getId());
if (!outbox.insert(commands.create(refreshed, refreshed.getDispatchVersion()))) {
throw new IllegalStateException("Unable to create lease recovery command");
}
return true;
}
private String safe(String message) {
String value = message == null || message.isBlank() ? "导出任务执行失败" : message;
return value.length() > 1000 ? value.substring(0, 1000) : value;
}
public record CreationResult(FileExportTask task, boolean created) { }
}
@@ -0,0 +1,21 @@
package com.baiyue.file.service;
import org.springframework.core.io.Resource;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.Path;
import java.time.Duration;
import java.util.function.Predicate;
public interface ExportStorage {
PendingFile create(long taskId);
long size(String objectKey);
Resource load(String objectKey);
boolean delete(String objectKey);
int cleanupOrphans(Duration minimumAge, Predicate<String> referenced);
record PendingFile(String objectKey, Path path, OutputStream output) implements AutoCloseable {
@Override public void close() throws IOException { output.close(); }
}
}
@@ -0,0 +1,111 @@
package com.baiyue.file.service;
import com.baiyue.file.config.ExportDefinitionRegistry;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import jakarta.annotation.PreDestroy;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@Service
public class ExportTaskExecutor {
private static final Logger LOGGER = LoggerFactory.getLogger(ExportTaskExecutor.class);
private final ExportTaskRepository tasks;
private final ExportDefinitionRegistry definitions;
private final GenericXlsxExportRunner runner;
private final ExportStateService states;
private final ExportStorage storage;
private final MeterRegistry meters;
private final int leaseSeconds, heartbeatSeconds, maxDurationMinutes, retentionDays;
private final ScheduledExecutorService watchdogs = Executors.newScheduledThreadPool(4, runnable -> {
Thread thread = new Thread(runnable, "file-export-watchdog"); thread.setDaemon(true); return thread;
});
public ExportTaskExecutor(ExportTaskRepository tasks, ExportDefinitionRegistry definitions,
GenericXlsxExportRunner runner, ExportStateService states, ExportStorage storage, MeterRegistry meters,
@Value("${baiyue.file.export.task.lease-seconds:300}") int leaseSeconds,
@Value("${baiyue.file.export.task.heartbeat-seconds:60}") int heartbeatSeconds,
@Value("${baiyue.file.export.task.max-duration-minutes:30}") int maxDurationMinutes,
@Value("${baiyue.file.export.retention-days:7}") int retentionDays) {
this.tasks=tasks;this.definitions=definitions;this.runner=runner;this.states=states;this.storage=storage;this.meters=meters;
this.leaseSeconds=leaseSeconds;this.heartbeatSeconds=heartbeatSeconds;this.maxDurationMinutes=maxDurationMinutes;this.retentionDays=retentionDays;
}
public void execute(long taskId, int dispatchVersion, String owner) {
if (!tasks.claim(taskId, dispatchVersion, owner, leaseSeconds, maxDurationMinutes)) return;
FileExportTask task = tasks.find(taskId);
AtomicBoolean abort = new AtomicBoolean();
ScheduledFuture<?> watchdog = watchdogs.scheduleAtFixedRate(() -> heartbeat(taskId, dispatchVersion, owner, abort),
heartbeatSeconds, heartbeatSeconds, TimeUnit.SECONDS);
long started = System.nanoTime();
GenericXlsxExportRunner.Result result = null;
try {
result = runner.run(task, definitions.require(task.getExportType()), abort,
rows -> { if (!tasks.progress(taskId, dispatchVersion, owner, rows)) abort.set(true); });
if (!tasks.succeed(taskId, dispatchVersion, owner, result.fileName(), result.objectKey(),
result.contentType(), result.sizeBytes(), result.rows(), retentionDays)) {
deleteOrWarn(result.objectKey(), taskId, "最终状态提交未生效");
FileExportTask latest = tasks.find(taskId);
if (latest != null && "RUNNING".equals(latest.getStatus())
&& latest.getDispatchVersion() == dispatchVersion && owner.equals(latest.getLeaseOwner())) {
retryOrFail(latest, dispatchVersion, owner, ExportFailure.temporary("导出执行超过最大时长 " + maxDurationMinutes + " 分钟", null));
}
} else {
meters.counter("export_task_completed_total", "type", task.getExportType(), "status", "SUCCEEDED").increment();
}
} catch (RuntimeException exception) {
if (result != null) deleteOrWarn(result.objectKey(), taskId, "执行异常");
retryOrFail(tasks.find(taskId), dispatchVersion, owner, exception);
LOGGER.error("导出任务执行失败 taskId={} dispatchVersion={}", taskId, dispatchVersion, exception);
} finally {
watchdog.cancel(true);
meters.timer("export_task_duration_seconds", "type", task.getExportType())
.record(System.nanoTime() - started, TimeUnit.NANOSECONDS);
}
}
private void heartbeat(long id, int version, String owner, AtomicBoolean abort) {
try { if (!tasks.heartbeat(id, version, owner, leaseSeconds)) abort.set(true); }
catch (RuntimeException exception) { abort.set(true); LOGGER.error("导出 watchdog 续租失败 taskId={}", id, exception); }
}
private void deleteOrWarn(String objectKey, long taskId, String reason) {
try {
if (!storage.delete(objectKey)) {
LOGGER.warn("导出结果文件清理失败 taskId={} objectKey={} reason={}", taskId, objectKey, reason);
}
} catch (RuntimeException exception) {
LOGGER.warn("导出结果文件清理异常 taskId={} objectKey={} reason={}", taskId, objectKey, reason, exception);
}
}
private void retryOrFail(FileExportTask task, int expectedVersion, String owner, RuntimeException exception) {
if (task == null || !"RUNNING".equals(task.getStatus()) || task.getDispatchVersion() != expectedVersion
|| !owner.equals(task.getLeaseOwner())) return;
boolean retryable = !(exception instanceof ExportFailure failure) || failure.retryable();
String message = exception.getMessage() == null ? "导出任务执行失败" : exception.getMessage();
if (retryable && task.getAttemptCount() < task.getMaxAttempts()) {
int backoff = jitteredBackoffSeconds(task.getAttemptCount() <= 1 ? 60 : 900);
if (states.retryAfterFailure(task, owner, backoff, message)) return;
}
if (states.fail(task, owner, message)) {
meters.counter("export_task_completed_total", "type", task.getExportType(), "status", "FAILED").increment();
}
}
static int jitteredBackoffSeconds(int baseSeconds) {
int spread = Math.max(1, baseSeconds / 5);
return ThreadLocalRandom.current().nextInt(baseSeconds - spread, baseSeconds + spread + 1);
}
@PreDestroy
void shutdownWatchdogs() {
watchdogs.shutdownNow();
}
}
@@ -1,17 +0,0 @@
package com.baiyue.file.service;
import com.baiyue.file.entity.FileExportTask;
import java.util.Map;
public interface ExportTaskHandler {
String exportType();
String requiredPermission();
default void validateParameters(Map<String, Object> parameters) { }
ExportResult export(FileExportTask task, ProgressReporter progress);
record ExportResult(String fileName, String objectKey, String contentType, long sizeBytes, long rows) { }
@FunctionalInterface
interface ProgressReporter { void update(long rows); }
}
@@ -8,6 +8,7 @@ import com.baiyue.file.dto.ExportTaskDtos.CreateExportTaskRequest;
import com.baiyue.file.dto.ExportTaskDtos.ExportTaskResponse;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import com.baiyue.file.config.ExportDefinitionRegistry;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -15,38 +16,45 @@ import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.TreeMap;
@Service
public class ExportTaskService {
private final ExportTaskRepository repository;
private final LocalExportStorage storage;
private final ExportStorage storage;
private final ObjectMapper objectMapper;
private final Map<String, ExportTaskHandler> handlers;
private final ExportDefinitionRegistry registry;
private final ExportStateService stateService;
private final ExportMetrics metrics;
public ExportTaskService(ExportTaskRepository repository, LocalExportStorage storage,
ObjectMapper objectMapper, List<ExportTaskHandler> handlers) {
public ExportTaskService(ExportTaskRepository repository, ExportStorage storage,
ObjectMapper objectMapper, ExportDefinitionRegistry registry,
ExportStateService stateService, ExportMetrics metrics) {
this.repository = repository;
this.storage = storage;
this.objectMapper = objectMapper;
this.handlers = handlers.stream().collect(Collectors.toUnmodifiableMap(
handler -> handler.exportType().toUpperCase(Locale.ROOT), Function.identity()));
this.registry = registry;
this.stateService = stateService;
this.metrics = metrics;
}
public ExportTaskResponse create(AuthenticatedUser user, CreateExportTaskRequest request) {
requirePlatform(user);
String type = request.exportType().trim().toUpperCase(Locale.ROOT);
String format = request.format().trim().toUpperCase(Locale.ROOT);
ExportTaskHandler handler = handler(type);
var definition = registry.require(type);
if (!"XLSX".equals(format)) throw new BizException(ResultCode.VALIDATION_FAILED, "当前仅支持 XLSX 格式");
if (!user.permissions().contains(handler.requiredPermission())) {
String permission = definition.getRequiredPermission();
if (!definition.isEnabled() || !definition.getAppScope().equalsIgnoreCase(user.appCode())) {
throw new BizException(ResultCode.VALIDATION_FAILED, "该导出类型当前不可用");
}
if (!user.permissions().contains(permission)) {
throw new BizException(ResultCode.FORBIDDEN, "缺少该导出类型的操作权限");
}
handler.validateParameters(request.parameters() == null ? Map.of() : request.parameters());
validateParameters(request.parameters());
String parametersJson = parametersJson(request.parameters());
long id = IdWorker.getId();
LocalDateTime now = LocalDateTime.now();
FileExportTask task = new FileExportTask();
@@ -57,14 +65,19 @@ public class ExportTaskService {
task.setExportType(type);
task.setFormat(format);
task.setTaskName(request.taskName().trim());
task.setParametersJson(parametersJson(request.parameters()));
task.setParametersJson(parametersJson);
task.setStatus("PENDING");
task.setProcessedRows(0L);
task.setDispatchVersion(1);
task.setAttemptCount(0);
task.setMaxAttempts(definition.getMaxAttempts());
task.setNextRetryAt(now);
task.setRequestedBy(user.userId());
task.setRequestedUsername(user.username());
task.setRequestedAt(now);
repository.insert(task);
return response(task);
ExportStateService.CreationResult result = stateService.createConstrained(task);
if (result.created()) metrics.created(type);
return response(result.task());
}
public PageResult<ExportTaskResponse> page(AuthenticatedUser user, long page, long size,
@@ -90,7 +103,7 @@ public class ExportTaskService {
if (!"SUCCEEDED".equals(task.getStatus())) {
throw new BizException(ResultCode.CONFLICT, "导出任务尚未完成或文件已过期");
}
if (task.getExpiresAt() != null && !task.getExpiresAt().isAfter(LocalDateTime.now())) {
if (!repository.downloadable(id)) {
throw new BizException(ResultCode.CONFLICT, "导出文件已过期");
}
return new DownloadResource(storage.load(task.getObjectKey()), task.getFileName(), task.getContentType());
@@ -98,22 +111,25 @@ public class ExportTaskService {
public ExportTaskResponse retry(AuthenticatedUser user, long id) {
FileExportTask task = ownedTask(user, id);
ExportTaskHandler handler = handler(task.getExportType());
if (!user.permissions().contains(handler.requiredPermission())) {
if (!"FAILED".equals(task.getStatus())) {
throw new BizException(ResultCode.CONFLICT, "只有失败的导出任务可以重试");
}
var definition = registry.require(task.getExportType());
if (!definition.isEnabled()) throw new BizException(ResultCode.CONFLICT, "该导出类型已停用");
String permission = definition.getRequiredPermission();
if (!user.permissions().contains(permission)) {
throw new BizException(ResultCode.FORBIDDEN, "缺少该导出类型的操作权限");
}
if (!"FAILED".equals(task.getStatus()) || !repository.retry(id)) {
if (!storage.delete(task.getObjectKey())) {
throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "上次导出文件清理失败,请稍后重试");
}
boolean retried = stateService.manualRetry(task);
if (!retried) {
throw new BizException(ResultCode.CONFLICT, "只有失败的导出任务可以重试");
}
return response(repository.find(id));
}
ExportTaskHandler handler(String exportType) {
ExportTaskHandler handler = handlers.get(exportType == null ? "" : exportType.toUpperCase(Locale.ROOT));
if (handler == null) throw new BizException(ResultCode.VALIDATION_FAILED, "不支持的导出类型");
return handler;
}
private FileExportTask ownedTask(AuthenticatedUser user, long id) {
requirePlatform(user);
FileExportTask task = repository.find(id);
@@ -128,10 +144,47 @@ public class ExportTaskService {
}
private String parametersJson(Map<String, Object> parameters) {
try { return objectMapper.writeValueAsString(parameters == null ? Map.of() : parameters); }
try { return objectMapper.writeValueAsString(parameters == null ? Map.of() : new TreeMap<>(parameters)); }
catch (JsonProcessingException exception) { throw new BizException(ResultCode.BAD_REQUEST, "导出筛选条件无法序列化"); }
}
private void validateParameters(Map<String, Object> parameters) {
Map<String, Object> values = parameters == null ? Map.of() : parameters;
if (values.size() > 20) throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选条件最多 20 项");
values.forEach((key, value) -> {
if (key == null || key.isBlank() || key.length() > 64) throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选键无效");
validateParameterValue(value, 1);
});
try {
if (objectMapper.writeValueAsBytes(values).length > 4096) throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选条件不能超过 4KB");
} catch (JsonProcessingException exception) {
throw new BizException(ResultCode.BAD_REQUEST, "导出筛选条件无法序列化");
}
}
private void validateParameterValue(Object value, int depth) {
if (depth > 2) throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选条件嵌套不能超过 2 层");
if (value == null || value instanceof Number || value instanceof Boolean) return;
if (value instanceof String text) {
if (text.length() > 256) throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选值最多 256 字符");
return;
}
if (value instanceof Map<?, ?> map) {
map.forEach((key, nested) -> {
if (!(key instanceof String text) || text.isBlank() || text.length() > 64) {
throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选键无效");
}
validateParameterValue(nested, depth + 1);
});
return;
}
if (value instanceof Iterable<?> values) {
values.forEach(nested -> validateParameterValue(nested, depth + 1));
return;
}
throw new BizException(ResultCode.VALIDATION_FAILED, "导出筛选值类型无效");
}
private ExportTaskResponse response(FileExportTask task) {
return new ExportTaskResponse(String.valueOf(task.getId()), task.getTaskNo(), task.getExportType(),
task.getFormat(), task.getTaskName(), task.getStatus(), task.getTotalRows(),
@@ -1,75 +0,0 @@
package com.baiyue.file.service;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import com.baiyue.common.core.exception.BizException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
@Service
public class ExportTaskWorker {
private static final Logger LOGGER = LoggerFactory.getLogger(ExportTaskWorker.class);
private final ExportTaskRepository repository;
private final ExportTaskService service;
private final LocalExportStorage storage;
private final long retentionDays;
private final long staleRunningMinutes;
public ExportTaskWorker(ExportTaskRepository repository, ExportTaskService service,
LocalExportStorage storage,
@Value("${baiyue.file.export.retention-days:7}") long retentionDays,
@Value("${baiyue.file.export.stale-running-minutes:30}") long staleRunningMinutes) {
this.repository = repository;
this.service = service;
this.storage = storage;
this.retentionDays = Math.max(1, retentionDays);
this.staleRunningMinutes = Math.max(5, staleRunningMinutes);
}
@Scheduled(fixedDelayString = "${baiyue.file.export.worker-delay-ms:2000}")
public void poll() {
LocalDateTime now = LocalDateTime.now();
repository.requeueStale(now.minusMinutes(staleRunningMinutes));
expire(now);
processNext();
}
public boolean processNext() {
Long id = repository.firstPendingId();
if (id == null || !repository.claim(id, LocalDateTime.now())) return false;
FileExportTask task = repository.find(id);
try {
ExportTaskHandler handler = service.handler(task.getExportType());
var result = handler.export(task, rows -> repository.progress(id, rows));
LocalDateTime completed = LocalDateTime.now();
if (!repository.succeed(id, result.fileName(), result.objectKey(), result.contentType(),
result.sizeBytes(), result.rows(), completed, completed.plusDays(retentionDays))) {
storage.delete(result.objectKey());
throw new IllegalStateException("Export task state changed before completion");
}
} catch (Exception exception) {
String message = safeMessage(exception);
repository.fail(id, message, LocalDateTime.now());
LOGGER.error("导出任务执行失败,taskId={},exportType={}", id, task.getExportType(), exception);
}
return true;
}
private void expire(LocalDateTime now) {
for (FileExportTask task : repository.expiredArtifacts(now)) {
storage.delete(task.getObjectKey());
repository.markExpired(task.getId());
}
}
private String safeMessage(Exception exception) {
String message = exception instanceof BizException ? exception.getMessage() : "导出任务执行失败,请稍后重试";
if (message == null || message.isBlank()) message = "导出任务执行失败";
return message.length() <= 1000 ? message : message.substring(0, 1000);
}
}
@@ -0,0 +1,208 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.export.ExportContracts.ExportBatchRequest;
import com.baiyue.common.core.export.ExportContracts.ExportBatchResponse;
import com.baiyue.common.core.export.ExportContracts.ExportColumn;
import com.baiyue.file.client.ExportProviderClient;
import com.baiyue.file.client.ExportProviderResponseTooLargeException;
import com.baiyue.file.config.ExportDefinitionRegistry.Definition;
import com.baiyue.file.entity.FileExportTask;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.poi.ss.usermodel.*;
import org.apache.poi.xssf.streaming.SXSSFWorkbook;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.LongConsumer;
@Service
public class GenericXlsxExportRunner {
private static final Logger LOGGER = LoggerFactory.getLogger(GenericXlsxExportRunner.class);
public static final String CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final int EXCEL_SHEET_ROWS = 1_048_575;
private final ExportProviderClient provider;
private final ExportStorage storage;
private final ObjectMapper objectMapper;
private final String internalToken;
private final long maxFileBytes;
public GenericXlsxExportRunner(ExportProviderClient provider, ExportStorage storage, ObjectMapper objectMapper,
@Value("${baiyue.security.internal-service-token}") String internalToken,
@Value("${baiyue.file.export.task.max-file-size-mb:50}") long maxFileSizeMb) {
this.provider = provider; this.storage = storage; this.objectMapper = objectMapper;
this.internalToken = internalToken; this.maxFileBytes = Math.max(1, maxFileSizeMb) * 1024L * 1024L;
}
public Result run(FileExportTask task, Definition definition, AtomicBoolean abort, LongConsumer progress) {
Map<String, Object> parameters = parameters(task.getParametersJson());
ExportStorage.PendingFile target = storage.create(task.getId());
long rows = 0;
String cursor = null, snapshot = null, schemaHash = null;
List<ExportColumn> columns = null;
Set<String> columnKeys = Set.of();
Set<String> warnedUnknownColumns = new HashSet<>();
Set<String> warnedTruncatedColumns = new HashSet<>();
try (target; SXSSFWorkbook workbook = new SXSSFWorkbook(100)) {
workbook.setCompressTempFiles(true);
WorkbookStyles styles = new WorkbookStyles(workbook);
SheetState sheet = null;
while (true) {
checkAbort(abort);
ExportBatchRequest request = new ExportBatchRequest(String.valueOf(task.getId()),
String.valueOf(task.getTenantId()), task.getAppCode(), String.valueOf(task.getRequestedBy()),
task.getDispatchVersion(), parameters, cursor, snapshot, definition.getBatchSize());
ApiResponse<ExportBatchResponse> response;
try { response = provider.batch(internalToken, task.getExportType(), definition, request); }
catch (RuntimeException exception) {
if (hasCause(exception, ExportProviderResponseTooLargeException.class)) {
throw ExportFailure.permanent("Provider 单批响应超过 4MB");
}
throw ExportFailure.temporary("导出数据服务暂不可用", exception);
}
if (response == null) throw ExportFailure.temporary("导出数据服务无响应", null);
if (!response.successful()) {
String message = response.message() == null ? "导出数据服务返回失败" : response.message();
if (response.code() != null && response.code().startsWith("S")) throw ExportFailure.temporary(message, null);
throw ExportFailure.permanent(message);
}
if (response.data() == null) throw ExportFailure.permanent("Provider 成功响应缺少 data");
ExportBatchResponse batch = response.data();
try { if (objectMapper.writeValueAsBytes(batch).length > 4 * 1024 * 1024) throw ExportFailure.permanent("Provider 单批响应超过 4MB"); }
catch (IOException exception) { throw ExportFailure.permanent("Provider 响应无法序列化"); }
if (batch.rows() == null || batch.rows().size() > definition.getBatchSize()) throw ExportFailure.permanent("Provider 单批数据超过限制");
if (batch.totalRows() != null && batch.totalRows() > definition.getMaxRows()) throw ExportFailure.permanent("导出预计行数超过最大限制 " + definition.getMaxRows());
if (schemaHash == null) {
if (batch.schemaVersion() != 1 || blank(batch.schemaHash()) || blank(batch.snapshotToken()) || batch.snapshotToken().length() > 512) {
throw ExportFailure.permanent("Provider 首批契约无效");
}
columns = validateColumns(batch.columns()); columnKeys = columns.stream().map(ExportColumn::key).collect(java.util.stream.Collectors.toUnmodifiableSet()); schemaHash = batch.schemaHash(); snapshot = batch.snapshotToken();
sheet = newSheet(workbook, 1, columns);
} else if (!schemaHash.equals(batch.schemaHash())) {
throw ExportFailure.permanent("Provider 导出列定义在执行期间发生变化");
}
for (Map<String, Object> values : batch.rows()) {
if (abort.get()) throw ExportFailure.temporary("导出执行租约已失效", null);
if (++rows > definition.getMaxRows()) throw ExportFailure.permanent("导出行数超过最大限制 " + definition.getMaxRows());
if (sheet.rowIndex > EXCEL_SHEET_ROWS) sheet = newSheet(workbook, sheet.index + 1, columns);
writeRow(sheet.sheet.createRow(sheet.rowIndex++), columns, columnKeys, values, styles, warnedUnknownColumns, warnedTruncatedColumns);
if (rows % 10_000 == 0) checkAbort(abort);
}
progress.accept(rows);
if (!batch.hasMore()) break;
if (blank(batch.nextCursor()) || batch.nextCursor().equals(cursor)) throw ExportFailure.temporary("Provider 游标未推进", null);
cursor = batch.nextCursor();
}
checkAbort(abort);
workbook.write(new SizeLimitedOutputStream(target.output(), maxFileBytes));
} catch (SizeLimitedOutputStream.ExportSizeExceededException exception) {
deleteOrWarn(target.objectKey(), task.getId(), "文件超过大小上限"); throw ExportFailure.permanent(exception.getMessage());
} catch (IOException exception) {
deleteOrWarn(target.objectKey(), task.getId(), "文件写入失败"); throw ExportFailure.temporary("导出文件写入失败", exception);
} catch (RuntimeException exception) {
deleteOrWarn(target.objectKey(), task.getId(), "导出生成失败"); throw exception;
}
String name = task.getTaskName() + "-" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd-HHmmss")) + ".xlsx";
try {
return new Result(name, target.objectKey(), CONTENT_TYPE, storage.size(target.objectKey()), rows);
} catch (RuntimeException exception) {
deleteOrWarn(target.objectKey(), task.getId(), "读取最终文件大小失败");
throw ExportFailure.temporary("导出文件读取失败", exception);
}
}
private void deleteOrWarn(String objectKey, long taskId, String reason) {
try {
if (!storage.delete(objectKey)) {
LOGGER.warn("导出临时文件清理失败 taskId={} objectKey={} reason={}", taskId, objectKey, reason);
}
} catch (RuntimeException exception) {
LOGGER.warn("导出临时文件清理异常 taskId={} objectKey={} reason={}", taskId, objectKey, reason, exception);
}
}
private List<ExportColumn> validateColumns(List<ExportColumn> columns) {
if (columns == null || columns.isEmpty() || columns.size() > 100) throw ExportFailure.permanent("Provider 导出列定义无效");
Set<String> keys = new HashSet<>();
for (ExportColumn column : columns) {
if (column == null || blank(column.key()) || blank(column.title()) || column.type() == null || !keys.add(column.key())) {
throw ExportFailure.permanent("Provider 导出列定义无效");
}
}
return List.copyOf(columns);
}
private SheetState newSheet(SXSSFWorkbook workbook, int index, List<ExportColumn> columns) {
Sheet sheet = workbook.createSheet(index == 1 ? "导出数据" : "导出数据-" + index);
CellStyle header = workbook.createCellStyle(); header.setFillForegroundColor(IndexedColors.GREY_25_PERCENT.getIndex()); header.setFillPattern(FillPatternType.SOLID_FOREGROUND);
Font font = workbook.createFont(); font.setBold(true); header.setFont(font);
Row row = sheet.createRow(0);
for (int i = 0; i < columns.size(); i++) {
Cell cell = row.createCell(i); cell.setCellValue(columns.get(i).title()); cell.setCellStyle(header);
sheet.setColumnWidth(i, Math.min(255, Math.max(8, columns.get(i).width() == null ? 18 : columns.get(i).width())) * 256);
}
sheet.createFreezePane(0, 1);
return new SheetState(index, sheet, 1);
}
private void writeRow(Row row, List<ExportColumn> columns, Set<String> columnKeys, Map<String, Object> values,
WorkbookStyles styles, Set<String> warnedUnknownColumns, Set<String> warnedTruncatedColumns) {
Map<String, Object> source = values == null ? Map.of() : values;
source.keySet().stream().filter(key -> !columnKeys.contains(key) && warnedUnknownColumns.add(key))
.forEach(key -> LOGGER.warn("Provider 返回未声明字段,已忽略 column={}", key));
for (int i = 0; i < columns.size(); i++) writeCell(row.createCell(i), columns.get(i), source.get(columns.get(i).key()), styles, warnedTruncatedColumns);
}
private void writeCell(Cell cell, ExportColumn column, Object value, WorkbookStyles styles, Set<String> warnedTruncatedColumns) {
if (value == null) return;
try {
switch (column.type()) {
case STRING -> cell.setCellValue(safeString(String.valueOf(value), column.key(), warnedTruncatedColumns));
case INTEGER -> cell.setCellValue(new BigDecimal(String.valueOf(value)).longValueExact());
case DECIMAL -> cell.setCellValue(new BigDecimal(String.valueOf(value)).doubleValue());
case BOOLEAN -> cell.setCellValue(value instanceof Boolean b ? b : Boolean.parseBoolean(String.valueOf(value)));
case DATE -> { cell.setCellValue(LocalDate.parse(String.valueOf(value))); cell.setCellStyle(styles.date); }
case DATETIME -> { cell.setCellValue(LocalDateTime.parse(String.valueOf(value))); cell.setCellStyle(styles.dateTime); }
}
} catch (RuntimeException exception) { throw ExportFailure.permanent("Provider 字段类型与列定义不匹配: " + column.key()); }
}
private String safeString(String value, String column, Set<String> warnedTruncatedColumns) {
if (value.length() > 32767 && warnedTruncatedColumns.add(column)) {
LOGGER.warn("Provider 字符串超过 Excel 单元格上限,已截断 column={}", column);
}
String text = value.length() > 32767 ? value.substring(0, 32767) : value;
return !text.isEmpty() && "=+-@\t\r\n".indexOf(text.charAt(0)) >= 0 ? "'" + text : text;
}
private Map<String, Object> parameters(String json) {
try { return objectMapper.readValue(json, new TypeReference<>() { }); }
catch (Exception exception) { throw ExportFailure.permanent("导出筛选条件无效"); }
}
private void checkAbort(AtomicBoolean abort) { if (abort.get()) throw ExportFailure.temporary("导出执行超过最大时长或租约已失效", null); }
private boolean blank(String value) { return value == null || value.isBlank(); }
private boolean hasCause(Throwable error, Class<? extends Throwable> type) {
for (Throwable current=error; current!=null; current=current.getCause()) if (type.isInstance(current)) return true;
return false;
}
private static final class WorkbookStyles {
final CellStyle date;
final CellStyle dateTime;
WorkbookStyles(Workbook workbook) {
DataFormat formats = workbook.getCreationHelper().createDataFormat();
date = workbook.createCellStyle(); date.setDataFormat(formats.getFormat("yyyy-mm-dd"));
dateTime = workbook.createCellStyle(); dateTime.setDataFormat(formats.getFormat("yyyy-mm-dd hh:mm:ss"));
}
}
private static final class SheetState { final int index; final Sheet sheet; int rowIndex; SheetState(int index, Sheet sheet, int rowIndex) { this.index=index;this.sheet=sheet;this.rowIndex=rowIndex; } }
public record Result(String fileName, String objectKey, String contentType, long sizeBytes, long rows) { }
}
@@ -13,10 +13,14 @@ import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.time.Duration;
import java.time.Instant;
import java.util.function.Predicate;
import java.util.stream.Stream;
import java.util.UUID;
@Service
public class LocalExportStorage {
public class LocalExportStorage implements ExportStorage {
private final Path root;
private final Path exportRoot;
@@ -61,14 +65,35 @@ public class LocalExportStorage {
catch (Exception exception) { throw new BizException(ResultCode.NOT_FOUND, "导出文件不存在或已清理"); }
}
public void delete(String objectKey) {
if (objectKey == null || objectKey.isBlank()) return;
public boolean delete(String objectKey) {
if (objectKey == null || objectKey.isBlank()) return true;
Path target = resolve(objectKey);
try {
Files.deleteIfExists(target);
Path parent = target.getParent();
if (parent != null && parent.startsWith(exportRoot) && !parent.equals(exportRoot)) Files.deleteIfExists(parent);
} catch (IOException ignored) { }
} catch (IOException ignored) {
return false;
}
Path parent = target.getParent();
if (parent != null && parent.startsWith(exportRoot) && !parent.equals(exportRoot)) {
try { Files.deleteIfExists(parent); }
catch (IOException ignored) { /* 文件已经删除,空目录清理失败不影响任务过期。 */ }
}
return true;
}
@Override
public int cleanupOrphans(Duration minimumAge, Predicate<String> referenced) {
int removed = 0;
Instant cutoff = Instant.now().minus(minimumAge);
try (Stream<Path> paths = Files.walk(exportRoot)) {
for (Path path : paths.filter(Files::isRegularFile).toList()) {
String objectKey = root.relativize(path).toString().replace('\\', '/');
if (!objectKey.matches("exports/\\d+/[0-9a-fA-F-]{36}\\.xlsx")) continue;
if (Files.getLastModifiedTime(path).toInstant().isAfter(cutoff) || referenced.test(objectKey)) continue;
if (delete(objectKey)) removed++;
}
} catch (IOException ignored) { /* 下一轮继续清理;存储指标和日志负责暴露持续故障。 */ }
return removed;
}
private Path resolve(String objectKey) {
@@ -80,7 +105,4 @@ public class LocalExportStorage {
return target;
}
public record PendingFile(String objectKey, Path path, OutputStream output) implements AutoCloseable {
@Override public void close() throws IOException { output.close(); }
}
}
@@ -24,7 +24,7 @@ import java.util.UUID;
@Service
public class LocalQualificationStorage {
private static final Logger LOGGER = LoggerFactory.getLogger(LocalQualificationStorage.class);
private static final long MAX_SIZE = 10L * 1024 * 1024;
private static final long MAX_SIZE = 5L * 1024 * 1024;
private static final Set<String> TYPES = Set.of("image/jpeg", "image/png", "application/pdf");
private final Path root;
private final Path qualificationRoot;
@@ -53,7 +53,7 @@ public class LocalQualificationStorage {
public StoredFile store(long tenantId, MultipartFile file) {
if (file == null || file.isEmpty()) throw new BizException("One qualification file is required / 请选择资质文件");
if (file.getSize() > MAX_SIZE) throw new BizException("Qualification file must not exceed 10 MB / 资质文件不能超过 10MB");
if (file.getSize() > MAX_SIZE) throw new BizException("Qualification file must not exceed 5 MB / 资质文件不能超过 5MB");
String contentType = file.getContentType() == null ? "" : file.getContentType().toLowerCase(Locale.ROOT);
if (!TYPES.contains(contentType)) throw new BizException("Only PDF, JPG and PNG files are supported / 仅支持 PDF、JPG、PNG 文件");
if (!matchesSignature(file, contentType)) {
@@ -0,0 +1,19 @@
package com.baiyue.file.service;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.OutputStream;
final class SizeLimitedOutputStream extends FilterOutputStream {
private final long limit;
private long count;
SizeLimitedOutputStream(OutputStream out, long limit) { super(out); this.limit = limit; }
@Override public void write(int value) throws IOException { ensure(1); out.write(value); count++; }
@Override public void write(byte[] bytes, int offset, int length) throws IOException { ensure(length); out.write(bytes, offset, length); count += length; }
private void ensure(long added) throws IOException {
if (count + added > limit) throw new ExportSizeExceededException("导出文件超过最大限制 " + limit + " 字节");
}
static final class ExportSizeExceededException extends IOException {
ExportSizeExceededException(String message) { super(message); }
}
}
+57 -2
View File
@@ -20,10 +20,30 @@ spring:
group: ${NACOS_GROUP:BAIYUE}
namespace: ${NACOS_NAMESPACE:public}
config.namespace: ${NACOS_NAMESPACE:public}
openfeign:
client:
config:
iamExportProviderClient:
connect-timeout: 3000
read-timeout: 15000
orderExportProviderClient:
connect-timeout: 3000
read-timeout: 15000
warehouseExportProviderClient:
connect-timeout: 3000
read-timeout: 15000
firstlegExportProviderClient:
connect-timeout: 3000
read-timeout: 15000
servlet:
multipart:
max-file-size: 10MB
max-request-size: 11MB
rocketmq:
name-server: ${ROCKETMQ_NAME_SERVER:127.0.0.1:9876}
producer:
group: ${MQ_TOPIC_PREFIX:test-}baiyue-file-export-publisher-v1
send-message-timeout: 3000
baiyue:
security:
jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start}
@@ -35,8 +55,43 @@ baiyue:
storage-root: ${FILE_STORAGE_ROOT:${user.home}/baiyue-firstleg/uploads}
export:
retention-days: ${FILE_EXPORT_RETENTION_DAYS:7}
worker-delay-ms: ${FILE_EXPORT_WORKER_DELAY_MS:2000}
stale-running-minutes: ${FILE_EXPORT_STALE_RUNNING_MINUTES:30}
legacy-worker-enabled: false
instance-id: ${HOSTNAME:file-local}
task:
lease-seconds: ${FILE_EXPORT_TASK_LEASE_SECONDS:300}
heartbeat-seconds: ${FILE_EXPORT_TASK_HEARTBEAT_SECONDS:60}
max-duration-minutes: ${FILE_EXPORT_MAX_DURATION_MINUTES:30}
max-file-size-mb: ${FILE_EXPORT_MAX_FILE_SIZE_MB:50}
publisher:
enabled: ${FILE_EXPORT_PUBLISHER_ENABLED:false}
poll-interval-ms: ${FILE_EXPORT_PUBLISHER_POLL_MS:5000}
batch-size: 50
lease-seconds: 60
consumer:
enabled: ${FILE_EXPORT_CONSUMER_ENABLED:false}
concurrency: ${FILE_EXPORT_CONSUMER_CONCURRENCY:4}
consume-timeout-minutes: ${FILE_EXPORT_CONSUME_TIMEOUT_MINUTES:90}
max-reconsume-times: 2
reconciler:
enabled: ${FILE_EXPORT_RECONCILER_ENABLED:false}
poll-interval-ms: ${FILE_EXPORT_RECONCILER_POLL_MS:30000}
redispatch-grace-seconds: 60
sent-stale-seconds: 600
rocketmq:
topic: ${MQ_TOPIC_PREFIX:test-}baiyue-file-export-command-v1
consumer-group: ${MQ_TOPIC_PREFIX:test-}baiyue-file-export-worker-v1
dlq-topic: "%DLQ%${MQ_TOPIC_PREFIX:test-}baiyue-file-export-worker-v1"
dlq-consumer-group: ${MQ_TOPIC_PREFIX:test-}baiyue-file-export-dlq-handler-v1
definitions:
CUSTOMER_LIST:
provider-service: iam-service
provider-path: /internal/exports/v1/CUSTOMER_LIST/batches
required-permission: platform:customer:export
app-scope: PLATFORM
enabled: true
max-rows: 200000
max-attempts: 3
batch-size: 500
management.endpoints.web.exposure.include: health,info,prometheus
logging:
+43
View File
@@ -77,6 +77,14 @@ CREATE TABLE IF NOT EXISTS file_export_task (
content_type VARCHAR(128) NULL COMMENT '文件类型',
size_bytes BIGINT UNSIGNED NULL COMMENT '文件字节数',
failure_message VARCHAR(1000) NULL COMMENT '失败原因',
dispatch_version INT NOT NULL DEFAULT 1 COMMENT '派发版本,仅执行重试时递增',
attempt_count INT NOT NULL DEFAULT 0 COMMENT '已执行尝试次数',
max_attempts INT NOT NULL DEFAULT 3 COMMENT '自动重试上限',
next_retry_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '允许执行的最早时间',
lease_owner VARCHAR(128) NULL COMMENT '当前执行实例标识',
lease_until DATETIME(3) NULL COMMENT '执行租约截止时间',
last_heartbeat_at DATETIME(3) NULL COMMENT '最后心跳时间',
execution_deadline_at DATETIME(3) NULL COMMENT '本次执行截止时间',
requested_by BIGINT UNSIGNED NOT NULL COMMENT '申请用户ID',
requested_username VARCHAR(128) NULL COMMENT '申请账号快照',
requested_at DATETIME(3) NOT NULL COMMENT '申请时间',
@@ -93,5 +101,40 @@ CREATE TABLE IF NOT EXISTS file_export_task (
UNIQUE KEY uk_file_export_task_no (task_no),
KEY idx_file_export_owner (tenant_id, app_code, requested_by, requested_at),
KEY idx_file_export_status (status, requested_at),
KEY idx_file_export_dispatch (status, next_retry_at, id),
KEY idx_file_export_lease (status, lease_until),
KEY idx_file_export_expiry (status, expires_at)
) ENGINE=InnoDB COMMENT='公共异步导出任务表';
CREATE TABLE IF NOT EXISTS file_export_task_guard (
guard_key VARCHAR(64) NOT NULL COMMENT '任务创建串行化锁键',
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY (guard_key)
) ENGINE=InnoDB COMMENT='导出任务创建并发保护表:低频创建入口使用单行锁保证重复提交及活动上限';
INSERT IGNORE INTO file_export_task_guard (guard_key) VALUES ('GLOBAL_CREATE');
CREATE TABLE IF NOT EXISTS file_export_outbox (
id BIGINT UNSIGNED NOT NULL COMMENT 'Outbox记录ID',
event_id VARCHAR(64) NOT NULL COMMENT '投递意图唯一标识',
task_id BIGINT UNSIGNED NOT NULL COMMENT '关联导出任务ID',
dispatch_version INT NOT NULL COMMENT '任务派发版本',
topic VARCHAR(128) NOT NULL COMMENT '目标Topic',
tag VARCHAR(64) NOT NULL COMMENT '消息Tag',
message_key VARCHAR(64) NOT NULL COMMENT '消息Key',
payload_json JSON NOT NULL COMMENT '命令消息体',
status VARCHAR(24) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING/CLAIMED/SENT/QUARANTINED',
attempt_count INT NOT NULL DEFAULT 0 COMMENT '累计发送尝试次数',
next_retry_at DATETIME(3) NOT NULL COMMENT '允许发送的最早时间',
lease_owner VARCHAR(128) NULL COMMENT '领取实例标识',
lease_until DATETIME(3) NULL COMMENT '领取租约截止时间',
sent_at DATETIME(3) NULL COMMENT '发送成功时间',
last_error VARCHAR(1000) NULL COMMENT '最后发送错误',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
PRIMARY KEY (id),
UNIQUE KEY uk_file_export_outbox_event (event_id),
UNIQUE KEY uk_file_export_outbox_task_version (task_id, dispatch_version),
KEY idx_file_export_outbox_publish (status, next_retry_at, id),
KEY idx_file_export_outbox_task (task_id, dispatch_version, status, sent_at)
) ENGINE=InnoDB COMMENT='导出命令事务Outbox表';
@@ -0,0 +1,33 @@
package com.baiyue.file.client;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.export.ExportContracts.ExportBatchRequest;
import com.baiyue.common.core.export.ExportContracts.ExportBatchResponse;
import com.baiyue.file.config.ExportDefinitionRegistry;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.junit.jupiter.api.Test;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.mockito.Mockito.*;
class ExportProviderClientTest {
@Test void routesTheRequestToTheAllowlistedServiceSelectedByTheDefinition() {
IamExportProviderClient iam=mock(IamExportProviderClient.class);
OrderExportProviderClient order=mock(OrderExportProviderClient.class);
WarehouseExportProviderClient warehouse=mock(WarehouseExportProviderClient.class);
FirstlegExportProviderClient firstleg=mock(FirstlegExportProviderClient.class);
var response=ApiResponse.success(new ExportBatchResponse(1,"hash","snapshot",java.util.List.of(),java.util.List.of(),null,false,0L));
when(order.batch(anyString(),anyString(),any())).thenReturn(response);
var definition=new ExportDefinitionRegistry.Definition();definition.setProviderService("order-service");
ExportBatchRequest request=new ExportBatchRequest("9","0","PLATFORM","7",1, Map.of(),null,null,500);
var actual=new ExportProviderClient(iam,order,warehouse,firstleg,new SimpleMeterRegistry())
.batch("token","ORDER_LIST",definition,request);
assertSame(response,actual);
verify(order).batch("token","ORDER_LIST",request);
verifyNoInteractions(iam,warehouse,firstleg);
}
}
@@ -0,0 +1,25 @@
package com.baiyue.file.client;
import feign.Request;
import feign.Response;
import feign.codec.Decoder;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertThrows;
class ExportProviderFeignConfigurationTest {
@Test void rejectsAnOversizedProviderResponseBeforeJsonDeserialization() {
ExportProviderFeignConfiguration configuration=new ExportProviderFeignConfiguration();
Decoder decoder=configuration.exportProviderDecoder(HttpMessageConverters::new);
byte[] body=new byte[ExportProviderFeignConfiguration.MAX_RESPONSE_BYTES+1];
Response response=Response.builder().status(200).reason("OK")
.request(Request.create(Request.HttpMethod.POST,"http://iam-service/internal/exports/v1/CUSTOMER_LIST/batches",Map.of(),null, StandardCharsets.UTF_8,null))
.body(body).build();
assertThrows(ExportProviderResponseTooLargeException.class,() -> decoder.decode(response,Object.class));
}
}
@@ -0,0 +1,30 @@
package com.baiyue.file.config;
import org.junit.jupiter.api.Test;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertThrows;
class ExportDefinitionRegistryTest {
@Test void acceptsTheControlledCustomerProvider() {
ExportDefinitionRegistry registry=new ExportDefinitionRegistry(); registry.setDefinitions(Map.of("CUSTOMER_LIST", definition("iam-service","/internal/exports/v1/CUSTOMER_LIST/batches")));
assertDoesNotThrow(registry::validate);
}
@Test void rejectsSsrfAndPathTypeMismatch() {
ExportDefinitionRegistry registry=new ExportDefinitionRegistry();
assertThrows(IllegalStateException.class, () -> registry.setDefinitions(Map.of("CUSTOMER_LIST", definition("evil-service","/internal/exports/v1/OTHER/batches"))));
}
@Test void rejectedRefreshKeepsTheLastValidDefinitions() {
ExportDefinitionRegistry registry=new ExportDefinitionRegistry();
var valid=definition("iam-service","/internal/exports/v1/CUSTOMER_LIST/batches");
registry.setDefinitions(Map.of("CUSTOMER_LIST", valid));
assertThrows(IllegalStateException.class, () -> registry.setDefinitions(Map.of("CUSTOMER_LIST", definition("evil-service","/internal/exports/v1/CUSTOMER_LIST/batches"))));
assertDoesNotThrow(() -> registry.require("CUSTOMER_LIST"));
org.junit.jupiter.api.Assertions.assertSame(valid, registry.require("CUSTOMER_LIST"));
}
private ExportDefinitionRegistry.Definition definition(String service,String path){
var d=new ExportDefinitionRegistry.Definition();d.setProviderService(service);d.setProviderPath(path);d.setRequiredPermission("platform:customer:export");d.setAppScope("PLATFORM");d.setMaxRows(200000);d.setMaxAttempts(3);d.setBatchSize(500);return d;
}
}
@@ -0,0 +1,57 @@
package com.baiyue.file.config;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import static org.junit.jupiter.api.Assertions.assertTrue;
class ExportInfrastructureContractTest {
@Test void deploymentPinsConcurrencyFourAndStartsAllSchedulersDisabled() throws Exception {
String compose=read("..","deploy","docker-compose.yml");
assertTrue(compose.contains("FILE_EXPORT_CONSUMER_CONCURRENCY: ${FILE_EXPORT_CONSUMER_CONCURRENCY:-4}"));
assertTrue(compose.contains("FILE_EXPORT_CONSUMER_ENABLED: ${FILE_EXPORT_CONSUMER_ENABLED:-false}"));
assertTrue(compose.contains("FILE_EXPORT_PUBLISHER_ENABLED: ${FILE_EXPORT_PUBLISHER_ENABLED:-false}"));
assertTrue(compose.contains("FILE_EXPORT_RECONCILER_ENABLED: ${FILE_EXPORT_RECONCILER_ENABLED:-false}"));
assertTrue(compose.contains("-Xms1g -Xmx2g"));
assertTrue(compose.contains("MQ_TOPIC_PREFIX: ${MQ_TOPIC_PREFIX?"));
assertTrue(!compose.contains("MQ_TOPIC_PREFIX: ${MQ_TOPIC_PREFIX:-test-}"));
}
@Test void rocketMqBootstrapCreatesBothGroupsAndAReadableWorkerDlq() throws Exception {
String compose=read("..","deploy","docker-compose.rocketmq.yml");
assertTrue(compose.contains("baiyue-file-export-command-v1\" -r 4 -w 4"));
assertTrue(compose.contains("baiyue-file-export-worker-v1\""));
assertTrue(compose.contains("baiyue-file-export-dlq-handler-v1\""));
assertTrue(compose.contains("%DLQ%$${MQ_TOPIC_PREFIX}baiyue-file-export-worker-v1\" -r 1 -w 1 -p 6"));
assertTrue(compose.contains("MQ_TOPIC_PREFIX: ${MQ_TOPIC_PREFIX?"));
assertTrue(compose.contains("sh -eu -c 'until sh mqadmin clusterList"));
assertTrue(compose.contains("| grep -q \"BaiyueCluster\""));
assertTrue(compose.contains("restart: on-failure"));
String environment=read("..","deploy","env",".env.infra-test.example");
assertTrue(environment.contains("ROCKETMQ_BROKER_IP=rocketmq-broker"));
assertTrue(environment.contains("MQ_TOPIC_PREFIX=test-"));
assertTrue(read("..","deploy","rocketmq","broker.conf.template").contains("autoCreateSubscriptionGroup=false"));
}
@Test void brokerRendererRejectsUnreachableAndUnexpandedAddresses() throws Exception {
String script=read("..","deploy","rocketmq","render-broker-config.sh");
assertTrue(script.contains("localhost|127.0.0.1"));
assertTrue(script.contains("*'${'*"));
assertTrue(script.contains("broker.conf still contains an unresolved placeholder"));
}
@Test void databaseBaselineContainsTaskOutboxAndPreseededCreationGuard() throws Exception {
String sql=read("src","main","resources","db","file.sql");
assertTrue(sql.contains("CREATE TABLE IF NOT EXISTS file_export_task ("));
assertTrue(sql.contains("CREATE TABLE IF NOT EXISTS file_export_outbox ("));
assertTrue(sql.contains("CREATE TABLE IF NOT EXISTS file_export_task_guard ("));
assertTrue(sql.contains("VALUES ('GLOBAL_CREATE')"));
}
private String read(String first,String... more) throws Exception {
return Files.readString(Path.of(first,more), StandardCharsets.UTF_8);
}
}
@@ -186,6 +186,44 @@ class QualificationFileContractTest {
assertTrue(requestCaptor.getValue().replace());
}
@Test
void merchantUploadRejectsMissingOrUnsupportedTypesBeforeWritingTheFile() {
var storage = mock(LocalQualificationStorage.class);
var iamClient = mock(IamQualificationClient.class);
var controller = merchantController(storage, iamClient);
var file = new MockMultipartFile("file", "license.pdf", "application/pdf", "%PDF-1.4".getBytes());
assertThrows(BizException.class,
() -> controller.upload(merchantUser(), file, null, null, null, false));
assertThrows(BizException.class,
() -> controller.upload(merchantUser(), file, "UNKNOWN", null, null, false));
verify(storage, never()).store(any(Long.class), any());
verify(iamClient, never()).submit(any(), any());
}
@Test
void everyRequiredQualificationTypeIsAcceptedAndForwarded() {
for (String qualificationType : Set.of(
"BUSINESS_LICENSE", "LEGAL_PERSON_ID",
"FREIGHT_FORWARDER_CERTIFICATE", "BANK_ACCOUNT_LICENSE")) {
var storage = mock(LocalQualificationStorage.class);
var iamClient = mock(IamQualificationClient.class);
var controller = merchantController(storage, iamClient);
var file = new MockMultipartFile("file", "qualification.pdf", "application/pdf", "%PDF-1.4".getBytes());
when(storage.store(10001L, file)).thenReturn(storedFile());
when(iamClient.submit(eq("internal-token"), any())).thenReturn(ApiResponse.success(
new IamQualificationClient.QualificationResponse(31L, null, "PROFILE_REQUIRED", null)));
controller.upload(merchantUser(), file, qualificationType, null, null, false);
var requestCaptor = org.mockito.ArgumentCaptor.forClass(
IamQualificationClient.QualificationRequest.class);
verify(iamClient).submit(eq("internal-token"), requestCaptor.capture());
assertEquals(qualificationType, requestCaptor.getValue().qualificationType());
}
}
private static PlatformQualificationFileController platformController(
LocalQualificationStorage storage, IamQualificationClient iamClient) {
var controller = new PlatformQualificationFileController();
@@ -0,0 +1,54 @@
package com.baiyue.file.mapper;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import org.apache.ibatis.annotations.Delete;
import org.junit.jupiter.api.Test;
import java.lang.reflect.Method;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class ExportMapperContractTest {
@Test void finalSuccessCommitIsGuardedByVersionLeaseAndDatabaseDeadline() throws Exception {
String sql=update(FileExportTaskMapper.class,"succeed");
assertTrue(sql.contains("dispatch_version=#{dispatchVersion}"));
assertTrue(sql.contains("lease_owner=#{owner}"));
assertTrue(sql.contains("execution_deadline_at>CURRENT_TIMESTAMP(3)"));
}
@Test void sentRedispatchKeepsTheSameVersionAndExcludesQuarantinedRows() throws Exception {
String sql=update(FileExportOutboxMapper.class,"redispatchSent");
assertTrue(sql.contains("o.status='SENT'"));
assertTrue(sql.contains("t.dispatch_version=o.dispatch_version"));
assertFalse(sql.contains("dispatch_version=dispatch_version+1"));
assertFalse(sql.contains("QUARANTINED"));
}
@Test void missingOutboxDetectionUsesDatabaseTimeAndLetsAnyExistingStatusBlockDuplication() throws Exception {
String sql=select(FileExportTaskMapper.class,"missingOutbox");
assertTrue(sql.contains("CURRENT_TIMESTAMP(3)"));
assertTrue(sql.contains("NOT EXISTS"));
assertTrue(sql.contains("o.dispatch_version=file_export_task.dispatch_version"));
assertFalse(sql.contains("o.status"));
}
@Test void retentionCleanupCanOnlyDeleteOldSentRows() throws Exception {
Method target=java.util.Arrays.stream(FileExportOutboxMapper.class.getDeclaredMethods()).filter(candidate -> candidate.getName().equals("deleteRetainedSent")).findFirst().orElseThrow();
String sql=String.join(" ",target.getAnnotation(Delete.class).value()).replaceAll("\\s+"," ");
assertTrue(sql.contains("status='SENT'"));
assertTrue(sql.contains("CURRENT_TIMESTAMP(3)"));
assertFalse(sql.contains("PENDING"));
assertFalse(sql.contains("CLAIMED"));
}
private String update(Class<?> type,String method) throws Exception {
Method target=java.util.Arrays.stream(type.getDeclaredMethods()).filter(candidate -> candidate.getName().equals(method)).findFirst().orElseThrow();
return String.join(" ",target.getAnnotation(Update.class).value()).replaceAll("\\s+"," ");
}
private String select(Class<?> type,String method) throws Exception {
Method target=java.util.Arrays.stream(type.getDeclaredMethods()).filter(candidate -> candidate.getName().equals(method)).findFirst().orElseThrow();
return String.join(" ",target.getAnnotation(Select.class).value()).replaceAll("\\s+"," ");
}
}
@@ -0,0 +1,38 @@
package com.baiyue.file.mq;
import com.baiyue.file.service.ExportTaskExecutor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
class ExportCommandConsumerTest {
@Test void appliesConfiguredConcurrencyAndDispatchesAValidCommand() {
ExportTaskExecutor executor=mock(ExportTaskExecutor.class);
ExportCommandConsumer consumer=new ExportCommandConsumer(new ObjectMapper(),executor,"file-1",4,90,2);
DefaultMQPushConsumer pushConsumer=new DefaultMQPushConsumer();
consumer.prepareStart(pushConsumer);
assertEquals(4,pushConsumer.getConsumeThreadMin());
assertEquals(4,pushConsumer.getConsumeThreadMax());
assertEquals(90,pushConsumer.getConsumeTimeout());
assertEquals(2,pushConsumer.getMaxReconsumeTimes());
MessageExt message=new MessageExt();
message.setBody("{\"schemaVersion\":1,\"eventId\":\"event-1\",\"taskId\":\"9\",\"exportType\":\"CUSTOMER_LIST\",\"dispatchVersion\":3}".getBytes(StandardCharsets.UTF_8));
consumer.onMessage(message);
verify(executor).execute(9L,3,"file-1");
}
@Test void rejectsMalformedMessagesSoRocketMqCanRedeliverThem() {
ExportCommandConsumer consumer=new ExportCommandConsumer(new ObjectMapper(),mock(ExportTaskExecutor.class),"file-1",4,90,2);
MessageExt message=new MessageExt();message.setBody("not-json".getBytes(StandardCharsets.UTF_8));
assertThrows(IllegalArgumentException.class,() -> consumer.onMessage(message));
}
}
@@ -0,0 +1,29 @@
package com.baiyue.file.mq;
import com.baiyue.file.service.ExportStateService;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.junit.jupiter.api.Test;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import static org.mockito.Mockito.*;
class ExportDlqHandlerTest {
@Test void databaseFailuresCannotFallIntoAnUnconsumedSecondaryDlq() {
RocketMQMessageListener listener=ExportDlqHandler.class.getAnnotation(RocketMQMessageListener.class);
org.junit.jupiter.api.Assertions.assertEquals(Integer.MAX_VALUE,listener.maxReconsumeTimes());
}
@Test void quarantinesUsingUserPropertiesWithoutParsingTheBody() {
ExportStateService states=mock(ExportStateService.class); MessageExt message=new MessageExt();
SimpleMeterRegistry meters=new SimpleMeterRegistry();
message.putUserProperty("eventId","event-1");message.putUserProperty("taskId","9");message.putUserProperty("dispatchVersion","3");message.putUserProperty("schemaVersion","1");
new ExportDlqHandler(states,meters).onMessage(message);
verify(states).quarantine(eq("event-1"),eq(9L),eq(3),contains("自动隔离"));
org.junit.jupiter.api.Assertions.assertEquals(1,meters.counter("export_dlq_messages").count());
}
@Test void malformedPropertiesAreAcknowledgedWithoutTouchingTasks() {
ExportStateService states=mock(ExportStateService.class); SimpleMeterRegistry meters=new SimpleMeterRegistry();
new ExportDlqHandler(states,meters).onMessage(new MessageExt());
verifyNoInteractions(states); org.junit.jupiter.api.Assertions.assertEquals(1,meters.counter("export_dlq_messages").count());
}
}
@@ -0,0 +1,67 @@
package com.baiyue.file.mq;
import com.baiyue.file.entity.FileExportOutbox;
import com.baiyue.file.repository.ExportOutboxRepository;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.messaging.Message;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class ExportOutboxPublisherTest {
@Test void publishesAllDlqLocatorFieldsAsRocketMqUserProperties() {
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
RocketMQTemplate rocket=mock(RocketMQTemplate.class);
FileExportOutbox row=row();
when(outbox.publishableIds(50)).thenReturn(List.of(1L));when(outbox.claim(1L,"file-1",60)).thenReturn(true);when(outbox.find(1L)).thenReturn(row);
new ExportOutboxPublisher(outbox,rocket,"file-1",50,60).publish();
@SuppressWarnings("rawtypes") ArgumentCaptor<Message> message=ArgumentCaptor.forClass(Message.class);
verify(rocket).syncSend(eq("test-export:EXPORT_TASK"),message.capture(),eq(3000L));
assertEquals("event-1",message.getValue().getHeaders().get("eventId"));
assertEquals("9",message.getValue().getHeaders().get("taskId"));
assertEquals("3",message.getValue().getHeaders().get("dispatchVersion"));
assertEquals("1",message.getValue().getHeaders().get("schemaVersion"));
verify(outbox).sent(1L,"file-1");
}
@Test void returnsFailedSendsToPendingWithTheFirstBackoffInsteadOfGivingUp() {
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
RocketMQTemplate rocket=mock(RocketMQTemplate.class);
FileExportOutbox row=row();row.setAttemptCount(1);
when(outbox.publishableIds(50)).thenReturn(List.of(1L));when(outbox.claim(1L,"file-1",60)).thenReturn(true);when(outbox.find(1L)).thenReturn(row);
doThrow(new RuntimeException("broker down")).when(rocket).syncSend(anyString(),any(Message.class),anyLong());
new ExportOutboxPublisher(outbox,rocket,"file-1",50,60).publish();
verify(outbox).sendFailed(1L,"file-1",5,"broker down");
verify(outbox,never()).sent(anyLong(),anyString());
}
@Test void stillRetriesAfterThirtyConsecutiveBrokerFailuresWithoutADeadState() {
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
RocketMQTemplate rocket=mock(RocketMQTemplate.class);
FileExportOutbox row=row();row.setAttemptCount(30);
when(outbox.publishableIds(50)).thenReturn(List.of(1L));
when(outbox.claim(1L,"file-1",60)).thenReturn(true);
when(outbox.find(1L)).thenReturn(row);
doThrow(new RuntimeException("broker down")).when(rocket).syncSend(anyString(),any(Message.class),anyLong());
ExportOutboxPublisher publisher=new ExportOutboxPublisher(outbox,rocket,"file-1",50,60);
for(int i=0;i<30;i++) publisher.publish();
verify(outbox,times(30)).sendFailed(1L,"file-1",900,"broker down");
verify(outbox,never()).sent(anyLong(),anyString());
}
private FileExportOutbox row(){
FileExportOutbox row=new FileExportOutbox();row.setId(1L);row.setEventId("event-1");row.setTaskId(9L);row.setDispatchVersion(3);
row.setTopic("test-export");row.setTag("EXPORT_TASK");row.setMessageKey("9");row.setPayloadJson("{}");row.setAttemptCount(1);return row;
}
}
@@ -1,55 +0,0 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.file.client.IamExportClient;
import com.baiyue.file.entity.FileExportTask;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
class CustomerExportTaskHandlerTest {
@TempDir Path tempDir;
@Test
void generatesARealXlsxFromIamBatchesAndReportsProgress() throws Exception {
IamExportClient client = mock(IamExportClient.class);
LocalExportStorage storage = new LocalExportStorage(tempDir.toString());
storage.initialize();
var row = new IamExportClient.CustomerExportRow(
"101", "TEN101", "柏悦物流", "柏悦物流有限公司", "91440300MA5ABCDE12",
"有限责任公司", null, null, "CNY", "深圳市", "张三", "13800138000",
"ops@example.com", "baiyue", "李四", "基础版", 30, "ACTIVE", null, null);
when(client.customers("internal-token", null, 500, "柏悦", "ACTIVE"))
.thenReturn(ApiResponse.success(new IamExportClient.CustomerExportBatch(List.of(row), 101L, false)));
CustomerExportTaskHandler handler = new CustomerExportTaskHandler(
client, storage, new ObjectMapper().findAndRegisterModules(), "internal-token");
FileExportTask task = new FileExportTask();
task.setId(9L);
task.setParametersJson("{\"keyword\":\"柏悦\",\"status\":\"ACTIVE\"}");
AtomicLong progress = new AtomicLong();
var result = handler.export(task, progress::set);
assertEquals(1L, result.rows());
assertEquals(1L, progress.get());
assertTrue(result.fileName().endsWith(".xlsx"));
Path workbookPath = tempDir.resolve(result.objectKey());
assertTrue(Files.isRegularFile(workbookPath));
try (XSSFWorkbook workbook = new XSSFWorkbook(Files.newInputStream(workbookPath))) {
assertEquals("客户ID", workbook.getSheetAt(0).getRow(0).getCell(0).getStringCellValue());
assertEquals("柏悦物流", workbook.getSheetAt(0).getRow(1).getCell(2).getStringCellValue());
assertEquals("13800138000", workbook.getSheetAt(0).getRow(1).getCell(11).getStringCellValue());
}
}
}
@@ -0,0 +1,44 @@
package com.baiyue.file.service;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportOutboxRepository;
import com.baiyue.file.repository.ExportTaskRepository;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.List;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class ExportReconcilerTest {
@Test void performsEveryRecoveryResponsibilityWithoutExecutingBusinessExports() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
ExportStateService states=mock(ExportStateService.class);
ExportStorage storage=mock(ExportStorage.class);
FileExportTask missing=task(1L,"PENDING"), stale=task(2L,"RUNNING"), expired=task(3L,"SUCCEEDED");expired.setObjectKey("exports/3/file.xlsx");
when(outbox.releaseExpiredClaims()).thenReturn(2);when(outbox.redispatchSent(600,60)).thenReturn(3);
when(tasks.missingOutbox(60)).thenReturn(List.of(missing));when(states.ensureOutbox(missing)).thenReturn(true);
when(tasks.staleRunning()).thenReturn(List.of(stale));when(states.recoverStale(stale)).thenReturn(true);
when(tasks.expiredArtifactsDb()).thenReturn(List.of(expired));when(storage.delete(expired.getObjectKey())).thenReturn(true);
when(storage.cleanupOrphans(eq(Duration.ofHours(24)),any())).thenReturn(1);
ExportReconciler reconciler=new ExportReconciler(tasks,outbox,states,storage,new SimpleMeterRegistry(),60,600,7);
reconciler.reconcile();
verify(outbox).releaseExpiredClaims();verify(outbox).redispatchSent(600,60);
verify(states).ensureOutbox(missing);verify(states).recoverStale(stale);verify(tasks).markExpired(3L);
verify(storage).cleanupOrphans(eq(Duration.ofHours(24)),any());
}
@Test void deletesOnlyTheConfiguredDailyBatchOfRetainedSentOutboxRows() {
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
ExportReconciler reconciler=new ExportReconciler(mock(ExportTaskRepository.class),outbox,mock(ExportStateService.class),mock(ExportStorage.class),new SimpleMeterRegistry(),60,600,7);
reconciler.cleanupRetainedOutbox();
verify(outbox).deleteRetainedSent(7,5000);
}
private FileExportTask task(long id,String status){FileExportTask task=new FileExportTask();task.setId(id);task.setStatus(status);return task;}
}
@@ -0,0 +1,110 @@
package com.baiyue.file.service;
import com.baiyue.file.repository.ExportOutboxRepository;
import com.baiyue.file.repository.ExportTaskRepository;
import com.baiyue.file.entity.FileExportTask;
import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.*;
class ExportStateServiceTest {
@Test void creationLimitsAndDuplicateDetectionRunUnderTheDatabaseGuard() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
ExportCommandFactory commands=mock(ExportCommandFactory.class);
FileExportTask task=task(9L,1,"PENDING");
task.setTenantId(0L);task.setAppCode("PLATFORM");task.setRequestedBy(7L);task.setExportType("CUSTOMER_LIST");task.setParametersJson("{}");
when(commands.create(task,1)).thenReturn(mock(com.baiyue.file.entity.FileExportOutbox.class));
when(outbox.insert(any())).thenReturn(true);
ExportStateService service=new ExportStateService(tasks,outbox,commands,mock(ExportMetrics.class));
ExportStateService.CreationResult result=service.createConstrained(task);
org.junit.jupiter.api.Assertions.assertTrue(result.created());
InOrder order=inOrder(tasks,outbox);
order.verify(tasks).lockCreationGuard();
order.verify(tasks).recentDuplicate(0L,"PLATFORM",7L,"CUSTOMER_LIST","{}");
order.verify(tasks).activeForUser(0L,"PLATFORM",7L);
order.verify(tasks).activeForType("CUSTOMER_LIST");
order.verify(tasks).insert(task);
order.verify(tasks).initializeDispatchTime(9L);
order.verify(outbox).insert(any());
}
@Test void duplicateCreationReturnsTheExistingTaskWithoutWritingAnotherOutbox() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
FileExportTask request=task(9L,1,"PENDING"), existing=task(8L,1,"PENDING");
request.setTenantId(0L);request.setAppCode("PLATFORM");request.setRequestedBy(7L);request.setExportType("CUSTOMER_LIST");request.setParametersJson("{}");
when(tasks.recentDuplicate(0L,"PLATFORM",7L,"CUSTOMER_LIST","{}")).thenReturn(existing);
ExportStateService service=new ExportStateService(tasks,outbox,mock(ExportCommandFactory.class),mock(ExportMetrics.class));
ExportStateService.CreationResult result=service.createConstrained(request);
org.junit.jupiter.api.Assertions.assertFalse(result.created());
org.junit.jupiter.api.Assertions.assertSame(existing,result.task());
verify(tasks).lockCreationGuard();
verify(tasks,never()).insert(any());
verifyNoInteractions(outbox);
}
@Test void activeTaskLimitsAreEnforcedAfterTakingTheCreationGuard() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
FileExportTask request=task(9L,1,"PENDING");
request.setTenantId(0L);request.setAppCode("PLATFORM");request.setRequestedBy(7L);request.setExportType("CUSTOMER_LIST");request.setParametersJson("{}");
when(tasks.activeForUser(0L,"PLATFORM",7L)).thenReturn(3L);
ExportStateService service=new ExportStateService(tasks,outbox,mock(ExportCommandFactory.class),mock(ExportMetrics.class));
assertThrows(com.baiyue.common.core.exception.BizException.class,() -> service.createConstrained(request));
InOrder order=inOrder(tasks);
order.verify(tasks).lockCreationGuard();
order.verify(tasks).recentDuplicate(0L,"PLATFORM",7L,"CUSTOMER_LIST","{}");
order.verify(tasks).activeForUser(0L,"PLATFORM",7L);
verify(tasks,never()).insert(any());
verifyNoInteractions(outbox);
}
@Test void quarantineRequiresOutboxAndTaskToChangeAtomically() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
when(tasks.find(9L)).thenReturn(task(9L,3,"PENDING"));
when(outbox.quarantine("event-1",9L,3,"broken")).thenReturn(true);
when(tasks.failQuarantined(9L,3,"broken")).thenReturn(false);
ExportStateService service=new ExportStateService(tasks,outbox,mock(ExportCommandFactory.class),mock(ExportMetrics.class));
assertThrows(IllegalStateException.class,() -> service.quarantine("event-1",9L,3,"broken"));
verify(tasks).failQuarantined(9L,3,"broken");
}
@Test void oldDlqMessageThatMatchesNoOutboxDoesNotTouchTheCurrentTask() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
when(tasks.find(9L)).thenReturn(task(9L,3,"PENDING"));
when(outbox.quarantine(anyString(),anyLong(),anyInt(),anyString())).thenReturn(false);
ExportStateService service=new ExportStateService(tasks,outbox,mock(ExportCommandFactory.class),mock(ExportMetrics.class));
assertDoesNotThrow(() -> service.quarantine("old-event",9L,2,"broken"));
verify(tasks).find(9L);
verifyNoMoreInteractions(tasks);
}
@Test void matchingDlqMessageDoesNotDamageAnAlreadyRunningTask() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportOutboxRepository outbox=mock(ExportOutboxRepository.class);
when(tasks.find(9L)).thenReturn(task(9L,3,"RUNNING"));
ExportStateService service=new ExportStateService(tasks,outbox,mock(ExportCommandFactory.class),mock(ExportMetrics.class));
assertDoesNotThrow(() -> service.quarantine("event-1",9L,3,"broken"));
verifyNoInteractions(outbox);
verify(tasks,never()).failQuarantined(anyLong(),anyInt(),anyString());
}
private FileExportTask task(long id,int version,String status){
FileExportTask task=new FileExportTask();task.setId(id);task.setDispatchVersion(version);task.setStatus(status);return task;
}
}
@@ -0,0 +1,105 @@
package com.baiyue.file.service;
import com.baiyue.file.config.ExportDefinitionRegistry;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.junit.jupiter.api.Test;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class ExportTaskExecutorTest {
@Test void aCommandThatCannotClaimTheTaskIsAcknowledgedWithoutExecutingOrChangingState() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportDefinitionRegistry definitions=mock(ExportDefinitionRegistry.class);
GenericXlsxExportRunner runner=mock(GenericXlsxExportRunner.class);
ExportStateService states=mock(ExportStateService.class);
when(tasks.claim(9L,1,"file-1",300,30)).thenReturn(false);
ExportTaskExecutor executor=new ExportTaskExecutor(tasks,definitions,runner,states,mock(ExportStorage.class),new SimpleMeterRegistry(),300,60,30,7);
executor.execute(9L,1,"file-1");
verifyNoInteractions(definitions,runner,states);
executor.shutdownWatchdogs();
}
@Test void watchdogRenewsTheLeaseWhileTheRunnerIsStillWriting() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportDefinitionRegistry definitions=mock(ExportDefinitionRegistry.class);
GenericXlsxExportRunner runner=mock(GenericXlsxExportRunner.class);
ExportStateService states=mock(ExportStateService.class);
FileExportTask running=task(1,"file-1");
when(tasks.claim(9L,1,"file-1",5,30)).thenReturn(true);when(tasks.find(9L)).thenReturn(running);
when(tasks.heartbeat(9L,1,"file-1",5)).thenReturn(true);when(tasks.succeed(anyLong(),anyInt(),anyString(),anyString(),anyString(),anyString(),anyLong(),anyLong(),anyInt())).thenReturn(true);
when(definitions.require("CUSTOMER_LIST")).thenReturn(new ExportDefinitionRegistry.Definition());
when(runner.run(eq(running),any(),any(AtomicBoolean.class),any())).thenAnswer(invocation -> {
Thread.sleep(1200);
return new GenericXlsxExportRunner.Result("file.xlsx","exports/9/test.xlsx","application/xlsx",100,1);
});
ExportTaskExecutor executor=new ExportTaskExecutor(tasks,definitions,runner,states,mock(ExportStorage.class),new SimpleMeterRegistry(),5,1,30,7);
executor.execute(9L,1,"file-1");
verify(tasks,atLeastOnce()).heartbeat(9L,1,"file-1",5);
executor.shutdownWatchdogs();
}
@Test void aLateFailureFromAnOldExecutionCannotRetryOrFailANewerVersionOwnedByTheSameInstance() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportDefinitionRegistry definitions=mock(ExportDefinitionRegistry.class);
GenericXlsxExportRunner runner=mock(GenericXlsxExportRunner.class);
ExportStateService states=mock(ExportStateService.class);
ExportStorage storage=mock(ExportStorage.class);
when(tasks.claim(9L,1,"file-1",300,30)).thenReturn(true);
FileExportTask first=task(1,"file-1"), newer=task(2,"file-1");
when(tasks.find(9L)).thenReturn(first,newer);
when(definitions.require("CUSTOMER_LIST")).thenReturn(new ExportDefinitionRegistry.Definition());
when(runner.run(eq(first),any(),any(),any())).thenThrow(new RuntimeException("late failure"));
ExportTaskExecutor executor=new ExportTaskExecutor(tasks,definitions,runner,states,storage,new SimpleMeterRegistry(),300,60,30,7);
executor.execute(9L,1,"file-1");
verifyNoInteractions(states);
executor.shutdownWatchdogs();
}
@Test void deletesTheGeneratedFileWhenTheFinalDatabaseCommitThrows() {
ExportTaskRepository tasks=mock(ExportTaskRepository.class);
ExportDefinitionRegistry definitions=mock(ExportDefinitionRegistry.class);
GenericXlsxExportRunner runner=mock(GenericXlsxExportRunner.class);
ExportStateService states=mock(ExportStateService.class);
ExportStorage storage=mock(ExportStorage.class);
FileExportTask running=task(1,"file-1");
when(tasks.claim(9L,1,"file-1",300,30)).thenReturn(true);
when(tasks.find(9L)).thenReturn(running);
when(definitions.require("CUSTOMER_LIST")).thenReturn(new ExportDefinitionRegistry.Definition());
when(runner.run(eq(running),any(),any(),any())).thenReturn(
new GenericXlsxExportRunner.Result("file.xlsx","exports/9/result.xlsx","application/xlsx",100,1));
when(tasks.succeed(anyLong(),anyInt(),anyString(),anyString(),anyString(),anyString(),anyLong(),anyLong(),anyInt()))
.thenThrow(new RuntimeException("database unavailable"));
when(states.retryAfterFailure(eq(running),eq("file-1"),intThat(value -> value >= 48 && value <= 72),anyString())).thenReturn(true);
ExportTaskExecutor executor=new ExportTaskExecutor(tasks,definitions,runner,states,storage,new SimpleMeterRegistry(),300,60,30,7);
executor.execute(9L,1,"file-1");
verify(storage).delete("exports/9/result.xlsx");
verify(states).retryAfterFailure(eq(running),eq("file-1"),intThat(value -> value >= 48 && value <= 72),eq("database unavailable"));
executor.shutdownWatchdogs();
}
@Test void retryBackoffAlwaysStaysWithinTheDocumentedTwentyPercentJitter() {
for (int i=0;i<100;i++) {
int first=ExportTaskExecutor.jitteredBackoffSeconds(60);
int second=ExportTaskExecutor.jitteredBackoffSeconds(900);
org.junit.jupiter.api.Assertions.assertTrue(first>=48 && first<=72);
org.junit.jupiter.api.Assertions.assertTrue(second>=720 && second<=1080);
}
}
private FileExportTask task(int version,String owner){
FileExportTask task=new FileExportTask();task.setId(9L);task.setExportType("CUSTOMER_LIST");task.setStatus("RUNNING");
task.setDispatchVersion(version);task.setLeaseOwner(owner);task.setAttemptCount(1);task.setMaxAttempts(3);return task;
}
}
@@ -9,8 +9,9 @@ import com.baiyue.file.repository.ExportTaskRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import java.util.List;
import com.baiyue.file.config.ExportDefinitionRegistry;
import java.util.Map;
import java.util.List;
import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -25,19 +26,18 @@ class ExportTaskServiceTest {
@Test
void createsPersistedTaskOnlyWhenSourceExportPermissionIsPresent() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
ExportTaskHandler handler = mock(ExportTaskHandler.class);
when(handler.exportType()).thenReturn("CUSTOMER_LIST");
when(handler.requiredPermission()).thenReturn("platform:customer:export");
ExportTaskService service = new ExportTaskService(repository, mock(LocalExportStorage.class),
new ObjectMapper(), List.of(handler));
doAnswer(invocation -> null).when(repository).insert(any());
ExportStateService states=mock(ExportStateService.class);
when(states.createConstrained(any(FileExportTask.class))).thenAnswer(invocation ->
new ExportStateService.CreationResult(invocation.getArgument(0), true));
ExportTaskService service = new ExportTaskService(repository, mock(ExportStorage.class),
new ObjectMapper(), registry(), states, mock(ExportMetrics.class));
var response = service.create(user(1L, Set.of("platform:customer:export")),
new CreateExportTaskRequest("CUSTOMER_LIST", "XLSX", "客户列表", Map.of("status", "ACTIVE")));
assertEquals("PENDING", response.status());
assertEquals("CUSTOMER_LIST", response.exportType());
verify(repository).insert(any(FileExportTask.class));
verify(states).createConstrained(any(FileExportTask.class));
BizException denied = assertThrows(BizException.class, () -> service.create(user(1L, Set.of()),
new CreateExportTaskRequest("CUSTOMER_LIST", "XLSX", "客户列表", Map.of())));
assertEquals(ResultCode.FORBIDDEN, denied.getResultCode());
@@ -46,10 +46,8 @@ class ExportTaskServiceTest {
@Test
void refusesAnotherUsersTaskUnlessDownloadAllIsGranted() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
ExportTaskHandler handler = mock(ExportTaskHandler.class);
when(handler.exportType()).thenReturn("CUSTOMER_LIST");
ExportTaskService service = new ExportTaskService(repository, mock(LocalExportStorage.class),
new ObjectMapper(), List.of(handler));
ExportTaskService service = new ExportTaskService(repository, mock(ExportStorage.class),
new ObjectMapper(), registry(), mock(ExportStateService.class), mock(ExportMetrics.class));
FileExportTask task = new FileExportTask();
task.setId(9L); task.setTenantId(0L); task.setAppCode("PLATFORM"); task.setRequestedBy(2L);
when(repository.find(9L)).thenReturn(task);
@@ -59,7 +57,22 @@ class ExportTaskServiceTest {
assertEquals("9", service.get(user(1L, Set.of("platform:download:all")), 9L).id());
}
@Test
void rejectsParametersDeeperThanTwoLevels() {
ExportTaskService service = new ExportTaskService(mock(ExportTaskRepository.class), mock(ExportStorage.class),
new ObjectMapper(), registry(), mock(ExportStateService.class), mock(ExportMetrics.class));
var request = new CreateExportTaskRequest("CUSTOMER_LIST", "XLSX", "客户列表",
Map.of("level1", Map.of("level2", List.of(Map.of("level3", "value")))));
BizException error=assertThrows(BizException.class,() -> service.create(user(1L,Set.of("platform:customer:export")),request));
assertEquals(ResultCode.VALIDATION_FAILED,error.getResultCode());
}
private AuthenticatedUser user(long id, Set<String> permissions) {
return new AuthenticatedUser(id, 0L, 0L, id, "user" + id, "PLATFORM", "WEB", permissions);
}
private ExportDefinitionRegistry registry(){
var definition=new ExportDefinitionRegistry.Definition();definition.setProviderService("iam-service");definition.setProviderPath("/internal/exports/v1/CUSTOMER_LIST/batches");definition.setRequiredPermission("platform:customer:export");definition.setAppScope("PLATFORM");definition.setMaxRows(200000);definition.setMaxAttempts(3);definition.setBatchSize(500);
ExportDefinitionRegistry registry=mock(ExportDefinitionRegistry.class);when(registry.require("CUSTOMER_LIST")).thenReturn(definition);return registry;
}
}
@@ -1,60 +0,0 @@
package com.baiyue.file.service;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class ExportTaskWorkerTest {
@Test
void onlyExecutesATaskAfterTheAtomicClaimSucceeds() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
ExportTaskService service = mock(ExportTaskService.class);
LocalExportStorage storage = mock(LocalExportStorage.class);
ExportTaskHandler handler = mock(ExportTaskHandler.class);
FileExportTask task = new FileExportTask();
task.setId(9L); task.setExportType("CUSTOMER_LIST");
when(repository.firstPendingId()).thenReturn(9L);
when(repository.claim(eq(9L), any(LocalDateTime.class))).thenReturn(true);
when(repository.find(9L)).thenReturn(task);
when(service.handler("CUSTOMER_LIST")).thenReturn(handler);
when(handler.export(eq(task), any())).thenAnswer(invocation -> {
ExportTaskHandler.ProgressReporter progress = invocation.getArgument(1);
progress.update(12L);
return new ExportTaskHandler.ExportResult("客户列表.xlsx", "exports/9/file.xlsx",
CustomerExportTaskHandler.CONTENT_TYPE, 1024L, 12L);
});
when(repository.succeed(eq(9L), eq("客户列表.xlsx"), eq("exports/9/file.xlsx"),
eq(CustomerExportTaskHandler.CONTENT_TYPE), eq(1024L), eq(12L),
any(LocalDateTime.class), any(LocalDateTime.class))).thenReturn(true);
ExportTaskWorker worker = new ExportTaskWorker(repository, service, storage, 7, 30);
worker.processNext();
verify(repository).progress(9L, 12L);
verify(repository).succeed(eq(9L), eq("客户列表.xlsx"), eq("exports/9/file.xlsx"),
eq(CustomerExportTaskHandler.CONTENT_TYPE), eq(1024L), eq(12L),
any(LocalDateTime.class), any(LocalDateTime.class));
}
@Test
void skipsAContendedTaskThatAnotherWorkerClaimed() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
when(repository.firstPendingId()).thenReturn(9L);
when(repository.claim(eq(9L), any(LocalDateTime.class))).thenReturn(false);
ExportTaskWorker worker = new ExportTaskWorker(repository, mock(ExportTaskService.class),
mock(LocalExportStorage.class), 7, 30);
worker.processNext();
verify(repository, never()).find(9L);
}
}
@@ -0,0 +1,67 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.export.ExportContracts.*;
import com.baiyue.file.client.ExportProviderClient;
import com.baiyue.file.config.ExportDefinitionRegistry;
import com.baiyue.file.entity.FileExportTask;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.poi.ss.usermodel.CellType;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.ArgumentCaptor;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
import java.util.stream.IntStream;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
class GenericXlsxExportRunnerTest {
@TempDir Path temp;
@Test void writesRealTypedWorkbookAndCarriesSnapshotAcrossBatches() throws Exception {
ExportProviderClient provider=mock(ExportProviderClient.class); LocalExportStorage storage=new LocalExportStorage(temp.toString());storage.initialize();
List<ExportColumn> columns=List.of(new ExportColumn("text","文本",ExportColumnType.STRING,18),new ExportColumn("integer","整数",ExportColumnType.INTEGER,12),new ExportColumn("decimal","金额",ExportColumnType.DECIMAL,12),new ExportColumn("flag","布尔",ExportColumnType.BOOLEAN,10),new ExportColumn("date","日期",ExportColumnType.DATE,12),new ExportColumn("time","时间",ExportColumnType.DATETIME,20));
List<Map<String,Object>> firstRows=IntStream.range(0,100).mapToObj(index -> Map.<String,Object>of("text",index==0?"=SUM(1,2)":"row-"+index,"integer",12,"decimal","3.50","flag",true,"date","2026-08-03","time","2026-08-03T21:00:00")).toList();
var first=new ExportBatchResponse(1,"sha256:test","snapshot-9",columns,firstRows,"10",true,101L);
var second=new ExportBatchResponse(1,"sha256:test","snapshot-9",null,List.of(Map.of("text","\n=SUM(1,2)","integer",13,"decimal",4.5,"flag",false,"date","2026-08-04","time","2026-08-04T21:00:00")),null,false,101L);
when(provider.batch(anyString(),eq("CUSTOMER_LIST"),any(),any())).thenReturn(ApiResponse.success(first),ApiResponse.success(second));
FileExportTask task=new FileExportTask();task.setId(9L);task.setTenantId(0L);task.setAppCode("PLATFORM");task.setRequestedBy(7L);task.setDispatchVersion(1);task.setExportType("CUSTOMER_LIST");task.setTaskName("客户列表");task.setParametersJson("{}");
var definition=new ExportDefinitionRegistry.Definition();definition.setMaxRows(200000);definition.setBatchSize(500);
var result=new GenericXlsxExportRunner(provider,storage,new ObjectMapper(),"internal-token",50).run(task,definition,new AtomicBoolean(),ignored->{ });
assertEquals(101,result.rows());
try(var workbook=new XSSFWorkbook(storage.load(result.objectKey()).getInputStream())){
var row=workbook.getSheetAt(0).getRow(1);assertEquals("'=SUM(1,2)",row.getCell(0).getStringCellValue());
assertEquals(CellType.NUMERIC,row.getCell(1).getCellType());assertEquals(CellType.NUMERIC,row.getCell(2).getCellType());assertEquals(CellType.BOOLEAN,row.getCell(3).getCellType());assertEquals(CellType.NUMERIC,row.getCell(4).getCellType());assertEquals(CellType.NUMERIC,row.getCell(5).getCellType());
assertEquals("'\n=SUM(1,2)",workbook.getSheetAt(0).getRow(101).getCell(0).getStringCellValue());
assertTrue(workbook.getNumCellStyles()<10,"日期样式必须复用,不能按单元格创建");
}
ArgumentCaptor<ExportBatchRequest> requests=ArgumentCaptor.forClass(ExportBatchRequest.class);verify(provider,times(2)).batch(anyString(),eq("CUSTOMER_LIST"),eq(definition),requests.capture());
assertNull(requests.getAllValues().get(0).snapshotToken());assertEquals("snapshot-9",requests.getAllValues().get(1).snapshotToken());
}
@Test void deletesTheCompletedFileWhenReadingItsFinalSizeFails() throws Exception {
ExportProviderClient provider=mock(ExportProviderClient.class);
LocalExportStorage storage=spy(new LocalExportStorage(temp.toString()));storage.initialize();
var batch=new ExportBatchResponse(1,"sha256:test","snapshot-9",
List.of(new ExportColumn("text","文本",ExportColumnType.STRING,18)),
List.of(Map.of("text","row")),null,false,1L);
when(provider.batch(anyString(),eq("CUSTOMER_LIST"),any(),any())).thenReturn(ApiResponse.success(batch));
FileExportTask task=new FileExportTask();task.setId(9L);task.setTenantId(0L);task.setAppCode("PLATFORM");task.setRequestedBy(7L);task.setDispatchVersion(1);task.setExportType("CUSTOMER_LIST");task.setTaskName("客户列表");task.setParametersJson("{}");
var definition=new ExportDefinitionRegistry.Definition();definition.setMaxRows(200000);definition.setBatchSize(500);
doThrow(new RuntimeException("size unavailable")).when(storage).size(anyString());
assertThrows(ExportFailure.class,() -> new GenericXlsxExportRunner(provider,storage,new ObjectMapper(),"internal-token",50)
.run(task,definition,new AtomicBoolean(),ignored->{ }));
try(var paths=Files.walk(temp.resolve("exports"))) {
assertEquals(0,paths.filter(Files::isRegularFile).count());
}
}
}
@@ -64,6 +64,19 @@ class LocalQualificationStorageTest {
storage.load(10001L, stored.fileName()).getFile().toPath().toAbsolutePath());
}
@Test
void rejectsQualificationFilesLargerThanTheClientFiveMegabyteLimit() {
var storage = new LocalQualificationStorage(storageRoot.toString());
storage.initialize();
var upload = new MockMultipartFile("file", "oversized.pdf", "application/pdf",
new byte[5 * 1024 * 1024 + 1]);
BizException exception = assertThrows(BizException.class,
() -> storage.store(10001L, upload));
assertTrue(exception.getMessage().contains("5 MB"));
}
@Test
void deletesTheReplacedQualificationByItsValidatedDocumentUrl() {
var storage = new LocalQualificationStorage(storageRoot.toString());
@@ -0,0 +1,19 @@
package com.baiyue.file.service;
import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class SizeLimitedOutputStreamTest {
@Test void rejectsTheWriteBeforeAnyBytesBeyondTheLimitReachStorage() throws Exception {
ByteArrayOutputStream target=new ByteArrayOutputStream();
SizeLimitedOutputStream output=new SizeLimitedOutputStream(target,4);
output.write(new byte[]{1,2,3,4});
assertThrows(SizeLimitedOutputStream.ExportSizeExceededException.class,() -> output.write(5));
assertArrayEquals(new byte[]{1,2,3,4},target.toByteArray());
}
}