Merge remote-tracking branch 'origin/test' into test-01

# Conflicts:
#	config/nacos/cn/file-service.yml
#	config/nacos/local/file-service.yml
#	deploy/sql/README.md
#	firstleg/pom.xml
This commit is contained in:
2026-08-04 12:27:25 +08:00
136 changed files with 7995 additions and 191 deletions
@@ -3,12 +3,14 @@ package com.baiyue.file;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
* 文件中心服务启动入口。
*/
@SpringBootApplication(scanBasePackages = "com.baiyue")
@EnableFeignClients(basePackages = "com.baiyue.file.client")
@EnableScheduling
public class FileServiceApplication {
public static void main(String[] args) {
SpringApplication.run(FileServiceApplication.class, args);
@@ -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 { }
@@ -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 { }
@@ -20,8 +20,10 @@ public interface IamQualificationClient {
@RequestBody PlatformQualificationRequest request);
record QualificationRequest(Long tenantId, String qualificationType,
String certificateNo, String documentUrl) { }
record QualificationResponse(Long qualificationId, Long reviewId, String tenantStatus) { }
String certificateNo, String documentUrl,
java.time.LocalDate validTo, boolean replace) { }
record QualificationResponse(Long qualificationId, Long reviewId, String tenantStatus,
String replacedDocumentUrl) { }
record PlatformQualificationResponse(Long qualificationId, Long reviewId, String tenantStatus,
String replacedDocumentUrl) { }
record PlatformQualificationRequest(Long tenantId, String qualificationType,
@@ -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; }
}
}
@@ -0,0 +1,81 @@
package com.baiyue.file.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.core.api.PageResult;
import com.baiyue.common.security.AuthenticatedUser;
import com.baiyue.file.dto.ExportTaskDtos.CreateExportTaskRequest;
import com.baiyue.file.dto.ExportTaskDtos.ExportTaskResponse;
import com.baiyue.file.service.ExportTaskService;
import jakarta.validation.Valid;
import jakarta.validation.constraints.Max;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.Pattern;
import org.springframework.http.CacheControl;
import org.springframework.http.ContentDisposition;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
@RestController
@RequestMapping("/api/platform/v1/files/export-tasks")
@Validated
public class PlatformExportTaskController {
private final ExportTaskService service;
public PlatformExportTaskController(ExportTaskService service) { this.service = service; }
@PostMapping
@PreAuthorize("isAuthenticated()")
public ApiResponse<ExportTaskResponse> create(@AuthenticationPrincipal AuthenticatedUser user,
@Valid @RequestBody CreateExportTaskRequest request) {
return ApiResponse.success(service.create(user, request));
}
@GetMapping
@PreAuthorize("hasAuthority('platform:download:list')")
public ApiResponse<PageResult<ExportTaskResponse>> page(
@AuthenticationPrincipal AuthenticatedUser user,
@RequestParam(defaultValue = "1") @Min(1) long page,
@RequestParam(defaultValue = "20") @Min(1) @Max(100) long size,
@RequestParam(required = false) String exportType,
@RequestParam(required = false) @Pattern(regexp = "PENDING|RUNNING|SUCCEEDED|FAILED|EXPIRED") String status,
@RequestParam(required = false) LocalDate requestedFrom,
@RequestParam(required = false) LocalDate requestedTo) {
return ApiResponse.success(service.page(user, page, size, exportType, status, requestedFrom, requestedTo));
}
@GetMapping("/{id:\\d+}")
@PreAuthorize("hasAuthority('platform:download:list')")
public ApiResponse<ExportTaskResponse> get(@AuthenticationPrincipal AuthenticatedUser user,
@PathVariable long id) {
return ApiResponse.success(service.get(user, id));
}
@GetMapping("/{id:\\d+}/download")
@PreAuthorize("hasAuthority('platform:download:list')")
public ResponseEntity<org.springframework.core.io.Resource> download(
@AuthenticationPrincipal AuthenticatedUser user, @PathVariable long id) {
var download = service.download(user, id);
ContentDisposition disposition = ContentDisposition.attachment()
.filename(download.fileName(), StandardCharsets.UTF_8).build();
return ResponseEntity.ok()
.cacheControl(CacheControl.noStore())
.header(HttpHeaders.CONTENT_DISPOSITION, disposition.toString())
.contentType(MediaType.parseMediaType(download.contentType()))
.body(download.resource());
}
@PostMapping("/{id:\\d+}/retry")
@PreAuthorize("hasAuthority('platform:download:list')")
public ApiResponse<ExportTaskResponse> retry(@AuthenticationPrincipal AuthenticatedUser user,
@PathVariable long id) {
return ApiResponse.success(service.retry(user, id));
}
}
@@ -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,
@@ -18,6 +18,8 @@ import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.time.LocalDate;
@RestController
@RequestMapping("/api/merchant/v1/files/qualifications")
public class QualificationFileController {
@@ -31,13 +33,17 @@ public class QualificationFileController {
public ApiResponse<UploadResponse> upload(
@AuthenticationPrincipal AuthenticatedUser user,
@RequestPart("file") MultipartFile file,
@RequestParam(value = "certificateNo", required = false) String certificateNo) {
@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,
new IamQualificationClient.QualificationRequest(tenantId, "BUSINESS_LICENSE",
certificateNo, stored.documentUrl()));
new IamQualificationClient.QualificationRequest(tenantId, qualificationType,
certificateNo, stored.documentUrl(), validTo, replace));
if (response == null) {
throw new IllegalStateException("IAM qualification response is empty");
}
@@ -52,6 +58,9 @@ public class QualificationFileController {
throw new IllegalStateException("IAM qualification response data is empty");
}
var data = response.data();
if (!stored.documentUrl().equals(data.replacedDocumentUrl())) {
storage.deleteByDocumentUrl(tenantId, data.replacedDocumentUrl());
}
return ApiResponse.success(new UploadResponse(data.qualificationId(), data.reviewId(),
data.tenantStatus(), stored.documentUrl()));
} catch (BizException exception) {
@@ -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,24 @@
package com.baiyue.file.dto;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.Size;
import java.time.LocalDateTime;
import java.util.Map;
public final class ExportTaskDtos {
private ExportTaskDtos() { }
public record CreateExportTaskRequest(
@NotBlank @Size(max = 64) String exportType,
@NotBlank @Size(max = 16) String format,
@NotBlank @Size(max = 128) String taskName,
Map<String, Object> parameters) { }
public record ExportTaskResponse(
String id, String taskNo, String exportType, String format, String taskName,
String status, Long totalRows, long processedRows, String fileName,
String contentType, Long sizeBytes, String failureMessage,
String requestedBy, String requestedUsername, LocalDateTime requestedAt,
LocalDateTime startedAt, LocalDateTime completedAt, LocalDateTime expiresAt) { }
}
@@ -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;
}
@@ -0,0 +1,45 @@
package com.baiyue.file.entity;
import com.baiyue.common.core.model.BaseEntity;
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_task")
public class FileExportTask extends BaseEntity {
@TableId private Long id;
private Long tenantId;
private String appCode;
private String taskNo;
private String exportType;
private String format;
private String taskName;
private String parametersJson;
private String status;
private Long totalRows;
private Long processedRows;
private String fileName;
private String objectKey;
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;
private LocalDateTime startedAt;
private LocalDateTime completedAt;
private LocalDateTime expiresAt;
}
@@ -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);
}
@@ -0,0 +1,138 @@
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> {
@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); }
}
@@ -0,0 +1,106 @@
package com.baiyue.file.repository;
import com.baiyue.common.core.api.PageResult;
import com.baiyue.file.entity.FileExportTask;
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;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
@Repository
public class ExportTaskRepository {
private final FileExportTaskMapper mapper;
public ExportTaskRepository(FileExportTaskMapper mapper) {
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); }
public PageResult<FileExportTask> page(long page, long size, long tenantId, String appCode,
Long requestedBy, String exportType, String status,
LocalDate requestedFrom, LocalDate requestedTo) {
var query = new LambdaQueryWrapper<FileExportTask>()
.eq(FileExportTask::getTenantId, tenantId)
.eq(FileExportTask::getAppCode, appCode)
.eq(requestedBy != null, FileExportTask::getRequestedBy, requestedBy)
.eq(StringUtils.hasText(exportType), FileExportTask::getExportType, exportType)
.eq(StringUtils.hasText(status), FileExportTask::getStatus, status)
.ge(requestedFrom != null, FileExportTask::getRequestedAt,
requestedFrom == null ? null : requestedFrom.atStartOfDay())
.lt(requestedTo != null, FileExportTask::getRequestedAt,
requestedTo == null ? null : requestedTo.plusDays(1).atStartOfDay())
.orderByDesc(FileExportTask::getRequestedAt)
.orderByDesc(FileExportTask::getId);
return PageResult.from(mapper.selectPage(new Page<>(page, size), query));
}
public void markExpired(long id) {
mapper.update(null, new LambdaUpdateWrapper<FileExportTask>()
.eq(FileExportTask::getId, id)
.eq(FileExportTask::getStatus, "SUCCEEDED")
.set(FileExportTask::getStatus, "EXPIRED"));
}
}
@@ -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();
}
}
@@ -0,0 +1,207 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.PageResult;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.common.security.AuthenticatedUser;
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;
import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Locale;
import java.util.Map;
import java.util.TreeMap;
@Service
public class ExportTaskService {
private final ExportTaskRepository repository;
private final ExportStorage storage;
private final ObjectMapper objectMapper;
private final ExportDefinitionRegistry registry;
private final ExportStateService stateService;
private final ExportMetrics metrics;
public ExportTaskService(ExportTaskRepository repository, ExportStorage storage,
ObjectMapper objectMapper, ExportDefinitionRegistry registry,
ExportStateService stateService, ExportMetrics metrics) {
this.repository = repository;
this.storage = storage;
this.objectMapper = objectMapper;
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);
var definition = registry.require(type);
if (!"XLSX".equals(format)) throw new BizException(ResultCode.VALIDATION_FAILED, "当前仅支持 XLSX 格式");
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, "缺少该导出类型的操作权限");
}
validateParameters(request.parameters());
String parametersJson = parametersJson(request.parameters());
long id = IdWorker.getId();
LocalDateTime now = LocalDateTime.now();
FileExportTask task = new FileExportTask();
task.setId(id);
task.setTenantId(user.tenantId());
task.setAppCode(user.appCode().toUpperCase(Locale.ROOT));
task.setTaskNo("EXP" + id);
task.setExportType(type);
task.setFormat(format);
task.setTaskName(request.taskName().trim());
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);
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,
String exportType, String status,
LocalDate requestedFrom, LocalDate requestedTo) {
requirePlatform(user);
if (requestedFrom != null && requestedTo != null && requestedFrom.isAfter(requestedTo)) {
throw new BizException(ResultCode.VALIDATION_FAILED, "开始日期不能晚于结束日期");
}
Long owner = user.permissions().contains("platform:download:all") ? null : user.userId();
PageResult<FileExportTask> source = repository.page(page, size, user.tenantId(), user.appCode(), owner,
normalize(exportType), normalize(status), requestedFrom, requestedTo);
return new PageResult<>(source.records().stream().map(this::response).toList(),
source.page(), source.size(), source.total(), source.pages());
}
public ExportTaskResponse get(AuthenticatedUser user, long id) {
return response(ownedTask(user, id));
}
public DownloadResource download(AuthenticatedUser user, long id) {
FileExportTask task = ownedTask(user, id);
if (!"SUCCEEDED".equals(task.getStatus())) {
throw new BizException(ResultCode.CONFLICT, "导出任务尚未完成或文件已过期");
}
if (!repository.downloadable(id)) {
throw new BizException(ResultCode.CONFLICT, "导出文件已过期");
}
return new DownloadResource(storage.load(task.getObjectKey()), task.getFileName(), task.getContentType());
}
public ExportTaskResponse retry(AuthenticatedUser user, long id) {
FileExportTask task = ownedTask(user, id);
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 (!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));
}
private FileExportTask ownedTask(AuthenticatedUser user, long id) {
requirePlatform(user);
FileExportTask task = repository.find(id);
if (task == null || !task.getTenantId().equals(user.tenantId())
|| !task.getAppCode().equalsIgnoreCase(user.appCode())) {
throw new BizException(ResultCode.NOT_FOUND, "导出任务不存在");
}
if (!task.getRequestedBy().equals(user.userId()) && !user.permissions().contains("platform:download:all")) {
throw new BizException(ResultCode.FORBIDDEN, "不能访问其他用户的导出任务");
}
return task;
}
private String parametersJson(Map<String, Object> 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(),
task.getProcessedRows() == null ? 0 : task.getProcessedRows(), task.getFileName(),
task.getContentType(), task.getSizeBytes(), task.getFailureMessage(),
String.valueOf(task.getRequestedBy()), task.getRequestedUsername(), task.getRequestedAt(),
task.getStartedAt(), task.getCompletedAt(), task.getExpiresAt());
}
private void requirePlatform(AuthenticatedUser user) {
if (user == null || user.userId() == null || user.tenantId() == null
|| !"PLATFORM".equalsIgnoreCase(user.appCode())) {
throw new BizException(ResultCode.FORBIDDEN, "Platform identity is required");
}
}
private String normalize(String value) { return value == null || value.isBlank() ? null : value.trim().toUpperCase(Locale.ROOT); }
public record DownloadResource(org.springframework.core.io.Resource resource, String fileName, String contentType) { }
}
@@ -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) { }
}
@@ -0,0 +1,108 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.stereotype.Service;
import java.io.IOException;
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 implements ExportStorage {
private final Path root;
private final Path exportRoot;
public LocalExportStorage(@Value("${baiyue.file.storage-root}") String root) {
this.root = Path.of(root).toAbsolutePath().normalize();
this.exportRoot = this.root.resolve("exports").normalize();
}
@PostConstruct
void initialize() {
try {
Files.createDirectories(exportRoot);
if (!exportRoot.startsWith(root) || !Files.isDirectory(exportRoot) || !Files.isWritable(exportRoot)) {
throw new IOException("Export storage is not writable");
}
} catch (IOException exception) {
throw new IllegalStateException("导出文件存储目录不可用", exception);
}
}
public PendingFile create(long taskId) {
String objectKey = "exports/" + taskId + "/" + UUID.randomUUID() + ".xlsx";
Path target = resolve(objectKey);
try {
Files.createDirectories(target.getParent());
OutputStream output = Files.newOutputStream(target, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
return new PendingFile(objectKey, target, output);
} catch (IOException exception) {
throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "导出文件存储暂不可用");
}
}
public long size(String objectKey) {
try { return Files.size(resolve(objectKey)); }
catch (IOException exception) { throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "导出文件读取失败"); }
}
public Resource load(String objectKey) {
Path target = resolve(objectKey);
if (!Files.isRegularFile(target)) throw new BizException(ResultCode.NOT_FOUND, "导出文件不存在或已清理");
try { return new UrlResource(target.toUri()); }
catch (Exception exception) { throw new BizException(ResultCode.NOT_FOUND, "导出文件不存在或已清理"); }
}
public boolean delete(String objectKey) {
if (objectKey == null || objectKey.isBlank()) return true;
Path target = resolve(objectKey);
try {
Files.deleteIfExists(target);
} 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) {
if (objectKey == null || !objectKey.matches("exports/\\d+/[0-9a-fA-F-]{36}\\.xlsx")) {
throw new BizException(ResultCode.BAD_REQUEST, "Invalid export object key");
}
Path target = root.resolve(objectKey).normalize();
if (!target.startsWith(exportRoot)) throw new BizException(ResultCode.BAD_REQUEST, "Invalid export object key");
return target;
}
}
@@ -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); }
}
}
+59
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,6 +55,45 @@ baiyue:
url: ${INTEGRATION_SERVICE_URL:}
file:
storage-root: ${FILE_STORAGE_ROOT:${user.home}/baiyue-firstleg/uploads}
export:
retention-days: ${FILE_EXPORT_RETENTION_DAYS:7}
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:
+79
View File
@@ -59,3 +59,82 @@ CREATE TABLE IF NOT EXISTS file_download_log (
KEY idx_download_log_file (tenant_id, file_id, downloaded_at),
KEY idx_download_log_user (tenant_id, user_id, downloaded_at)
) ENGINE=InnoDB COMMENT='文件下载审计日志表';
CREATE TABLE IF NOT EXISTS file_export_task (
id BIGINT UNSIGNED NOT NULL COMMENT '导出任务ID(雪花ID)',
tenant_id BIGINT UNSIGNED NOT NULL COMMENT '申请人所属租户ID,平台任务为0',
app_code VARCHAR(24) NOT NULL COMMENT '申请端:PLATFORM或MERCHANT',
task_no VARCHAR(40) NOT NULL COMMENT '导出任务编号',
export_type VARCHAR(64) NOT NULL COMMENT '导出类型',
format VARCHAR(16) NOT NULL DEFAULT 'XLSX' COMMENT '导出格式',
task_name VARCHAR(128) NOT NULL COMMENT '任务名称',
parameters_json JSON NOT NULL COMMENT '筛选条件快照',
status VARCHAR(24) NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING/RUNNING/SUCCEEDED/FAILED/EXPIRED',
total_rows BIGINT UNSIGNED NULL COMMENT '预计总行数',
processed_rows BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已处理行数',
file_name VARCHAR(255) NULL COMMENT '下载文件名',
object_key VARCHAR(512) NULL COMMENT '存储对象键',
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 '申请时间',
started_at DATETIME(3) NULL COMMENT '开始时间',
completed_at DATETIME(3) NULL COMMENT '完成时间',
expires_at DATETIME(3) NULL COMMENT '文件过期时间',
created_by BIGINT UNSIGNED NULL COMMENT '创建人ID',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
updated_by BIGINT UNSIGNED NULL COMMENT '最后修改人ID',
updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
deleted TINYINT(1) NOT NULL DEFAULT 0 COMMENT '逻辑删除标记',
version INT NOT NULL DEFAULT 0 COMMENT '乐观锁版本号',
PRIMARY KEY (id),
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,61 @@
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("ROCKETMQ_NAME_SERVER: ${ROCKETMQ_NAME_SERVER-}"));
String deployScript=read("..","deploy","scripts","deploy-services.sh");
assertTrue(deployScript.contains("部署 file 服务时,环境文件必须显式包含 MQ_TOPIC_PREFIX"));
assertTrue(deployScript.contains("测试环境部署 file 服务时 MQ_TOPIC_PREFIX 必须为 test-"));
assertTrue(deployScript.contains("环境文件必须包含非空 ROCKETMQ_NAME_SERVER"));
}
@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);
}
}
@@ -0,0 +1,23 @@
package com.baiyue.file.controller;
import com.baiyue.common.security.AuthenticatedUser;
import com.baiyue.file.dto.ExportTaskDtos.CreateExportTaskRequest;
import org.junit.jupiter.api.Test;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.RequestMapping;
import static org.junit.jupiter.api.Assertions.assertEquals;
class PlatformExportTaskControllerContractTest {
@Test
void exposesThePlatformFileRouteAndProtectsListingAndDownloads() throws Exception {
RequestMapping mapping = PlatformExportTaskController.class.getAnnotation(RequestMapping.class);
assertEquals("/api/platform/v1/files/export-tasks", mapping.value()[0]);
assertEquals("isAuthenticated()", PlatformExportTaskController.class
.getDeclaredMethod("create", AuthenticatedUser.class, CreateExportTaskRequest.class)
.getAnnotation(PreAuthorize.class).value());
assertEquals("hasAuthority('platform:download:list')", PlatformExportTaskController.class
.getDeclaredMethod("download", AuthenticatedUser.class, long.class)
.getAnnotation(PreAuthorize.class).value());
}
}
@@ -159,12 +159,71 @@ class QualificationFileContractTest {
when(iamClient.submit(eq("internal-token"), any())).thenThrow(new RuntimeException("timeout"));
var exception = assertThrows(BizException.class,
() -> controller.upload(merchantUser(), file, null));
() -> controller.upload(merchantUser(), file, "BUSINESS_LICENSE", null, null, false));
assertEquals(ResultCode.SERVICE_UNAVAILABLE, exception.getResultCode());
verify(storage, never()).delete(stored);
}
@Test
void merchantUploadForwardsItsQualificationTypeAndCleansTheReplacedFile() {
var storage = mock(LocalQualificationStorage.class);
var iamClient = mock(IamQualificationClient.class);
var controller = merchantController(storage, iamClient);
var file = new MockMultipartFile("file", "legal-person.png", "image/png", new byte[] {1, 2, 3});
var stored = storedFile();
var oldUrl = "/api/merchant/v1/files/qualifications/22222222-2222-2222-2222-222222222222.png";
when(storage.store(10001L, file)).thenReturn(stored);
when(iamClient.submit(eq("internal-token"), any())).thenReturn(ApiResponse.success(
new IamQualificationClient.QualificationResponse(31L, null, "PROFILE_REQUIRED", oldUrl)));
controller.upload(merchantUser(), file, "LEGAL_PERSON_ID", null, LocalDate.of(2030, 1, 1), true);
var requestCaptor = org.mockito.ArgumentCaptor.forClass(IamQualificationClient.QualificationRequest.class);
verify(iamClient).submit(eq("internal-token"), requestCaptor.capture());
verify(storage).deleteByDocumentUrl(10001L, oldUrl);
assertEquals("LEGAL_PERSON_ID", requestCaptor.getValue().qualificationType());
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;
}
}
@@ -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;
}
}
@@ -0,0 +1,78 @@
package com.baiyue.file.service;
import com.baiyue.common.core.api.ResultCode;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.common.security.AuthenticatedUser;
import com.baiyue.file.dto.ExportTaskDtos.CreateExportTaskRequest;
import com.baiyue.file.entity.FileExportTask;
import com.baiyue.file.repository.ExportTaskRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
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;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class ExportTaskServiceTest {
@Test
void createsPersistedTaskOnlyWhenSourceExportPermissionIsPresent() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
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(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());
}
@Test
void refusesAnotherUsersTaskUnlessDownloadAllIsGranted() {
ExportTaskRepository repository = mock(ExportTaskRepository.class);
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);
BizException denied = assertThrows(BizException.class, () -> service.get(user(1L, Set.of()), 9L));
assertEquals(ResultCode.FORBIDDEN, denied.getResultCode());
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;
}
}
@@ -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());
}
}