fix: reject unavailable self-operated collection commands before initialization

This commit is contained in:
2026-10-12 20:45:31 +08:00
parent 3b137f8946
commit d782b2cd24
4 changed files with 139 additions and 11 deletions
@@ -1,6 +1,7 @@
package com.baiyue.lastleg.selfoperated.analytics.sync;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baiyue.common.core.exception.BizException;
import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
import com.baiyue.lastleg.selfoperated.analytics.model.*;
import org.springframework.stereotype.Service;
@@ -23,18 +24,39 @@ public class SelfopCommandService {
}
@Transactional
public SelfopSyncTask request(Command c,String operator){
identity(c.requestKey(),100);identity(c.warehouseCode(),64);require(config.getWarehouses().containsKey(c.warehouseCode()),"SELFOP_WAREHOUSE_UNCONFIGURED");
taskService.initialize();
if("RETRY".equals(c.action())){identity(c.taskNo(),64);identity(c.reason(),500);
try { return requestChecked(c,operator); }
catch(IllegalStateException failure){
String message=failure.getMessage()==null?null:REJECTIONS.get(failure.getMessage());
if(message==null)throw failure;
throw new BizException(message);
}
}
private SelfopSyncTask requestChecked(Command c,String operator){
if(!config.isExecutionEnabled())throw new BizException("采集功能未启用,当前环境不允许执行采集或重试");
if(c==null)throw new BizException("请提供采集命令");
input(c.requestKey(),100,"请求标识");input(c.warehouseCode(),64,"仓库编码");
require(config.getWarehouses().containsKey(c.warehouseCode()),"SELFOP_WAREHOUSE_UNCONFIGURED");
if(c.action()==null||!Set.of("RETRY","INVENTORY","BACKFILL","RECHECK","REBUILD","VALIDATION").contains(c.action()))
throw new BizException("不支持此采集操作");
if("RETRY".equals(c.action())){input(c.taskNo(),64,"任务编号");input(c.reason(),500,"重试原因");
var original=tasks.selectOne(new QueryWrapper<SelfopSyncTask>().eq("task_no",c.taskNo()).eq("source_account",config.getSourceAccount()));require(original!=null,"SELFOP_TASK_SCOPE_MISMATCH");
require("WAREHOUSE_SYNC".equals(original.getTaskRole())&&c.warehouseCode().equals(original.getWarehouseCode()),"SELFOP_TASK_SCOPE_MISMATCH");
require(!"WAITING_SOURCE".equals(original.getStatus()),"SELFOP_RETRY_WAITING_SOURCE");
require("BUSINESS".equals(original.getExecutionPurpose())&&"BUSINESS".equals(config.getCollectionMode()),"SELFOP_BUSINESS_PURPOSE_REQUIRED");
return taskService.retry(original.getId(),c.requestKey(),operator,c.reason());}
require(Set.of("INVENTORY","BACKFILL","RECHECK","REBUILD","VALIDATION").contains(c.action()),"SELFOP_COMMAND_INVALID");
String purpose=SelfopCollectionPolicy.configuredPurpose(config);
require(("VALIDATION".equals(purpose))==("VALIDATION".equals(c.action())),"SELFOP_VALIDATION_ENTRY_REQUIRED");
if("RECHECK".equals(c.action())){input(c.customerCode(),64,"货主编码");input(c.deliveryNo(),128,"出库单号");}
if(Set.of("BACKFILL","VALIDATION").contains(c.action())){
require(c.sourceFrom()!=null&&c.sourceUntil()!=null&&c.sourceFrom().isBefore(c.sourceUntil()),"SELFOP_QUERY_WINDOW_INVALID");
require(!c.sourceUntil().isAfter(c.sourceFrom().plusDays(1)),"SELFOP_SLICE_MAX_ONE_DAY");
if(config.getSourceQueryTimezone()!=null){var source=ZoneId.of(config.getSourceQueryTimezone());
require(source.getRules().getValidOffsets(c.sourceFrom()).size()==1&&source.getRules().getValidOffsets(c.sourceUntil()).size()==1,"SELFOP_SOURCE_TIME_AMBIGUOUS");}
}
taskService.initialize();
var scope=scopes.selectOne(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount()).eq("warehouse_code",c.warehouseCode())
.eq("data_family","INVENTORY".equals(c.action())?"INVENTORY":"OUTBOUND"));
require(scope!=null,"SELFOP_SCOPE_MISSING");
var task=new SelfopSyncTask();task.setSourceAccount(config.getSourceAccount());task.setRequestedBy(operator);task.setRequestKey("manual:"+c.requestKey());
task.setExecutionPurpose(purpose);
task.setTaskType("INVENTORY".equals(c.action())?"BACKFILL":c.action());task.setScopeHash(scope.getScopeHash());
@@ -42,6 +64,7 @@ public class SelfopCommandService {
task.setTaskRole("WAREHOUSE_SYNC");task.setScopeId(scope.getId());task.setWarehouseCode(c.warehouseCode());task.setQueryScopeJson(scope.getScopeJson());
}else{
var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",config.getSourceAccount()));
require(account!=null,"SELFOP_SCOPE_MISSING");
task.setTaskRole("ACCOUNT_DISCOVERY");task.setDiscoveryScopeId(account.getId());
if("VALIDATION".equals(c.action())){
require(c.sourceFrom()!=null&&c.sourceUntil()!=null&&c.sourceFrom().isBefore(c.sourceUntil())&&!c.sourceUntil().isAfter(c.sourceFrom().plusDays(1)),"SELFOP_QUERY_WINDOW_INVALID");
@@ -62,4 +85,24 @@ public class SelfopCommandService {
}
task.setRequestHash(hash(Arrays.asList(c,task.getScopeHash())));return taskService.create(task);
}
private static void input(String value,int max,String label){
if(value==null||value.isBlank()||value.length()>max)throw new BizException(label+"不能为空且不能超过"+max+"个字符");
}
private static final Map<String,String> REJECTIONS=Map.ofEntries(
Map.entry("SELFOP_WAREHOUSE_UNCONFIGURED","该仓库尚未配置采集,暂不能提交任务"),
Map.entry("SELFOP_TASK_SCOPE_MISMATCH","任务不存在或不属于当前账号和仓库"),
Map.entry("SELFOP_TASK_MISSING","任务不存在,请刷新后重试"),
Map.entry("SELFOP_RETRY_WAITING_SOURCE","任务正在等待来源数据,暂不能重试"),
Map.entry("SELFOP_BUSINESS_PURPOSE_REQUIRED","验证任务不能通过正式采集重试恢复"),
Map.entry("SELFOP_VALIDATION_ENTRY_REQUIRED","当前采集模式与所选操作不一致,请使用对应入口"),
Map.entry("SELFOP_QUERY_WINDOW_INVALID","采集时间范围不能为空,结束时间必须晚于开始时间"),
Map.entry("SELFOP_SLICE_MAX_ONE_DAY","单次采集时间范围不能超过一天"),
Map.entry("SELFOP_SOURCE_TIME_AMBIGUOUS","来源时间存在时区歧义,请重新选择时间范围"),
Map.entry("SELFOP_SCOPE_MISSING","采集作用域尚未准备就绪,请联系管理员"),
Map.entry("SELFOP_IDEMPOTENCY_CONFLICT","请求标识已用于其他内容,请刷新后重新提交"),
Map.entry("SELFOP_RETRY_LEASE_BUSY","任务正在执行,暂不能重试"),
Map.entry("SELFOP_TASK_ALREADY_SUCCEEDED","任务已成功完成,无需重试"),
Map.entry("SELFOP_COLLECTION_MODE_REJECTED","当前采集模式不允许此操作"),
Map.entry("SELFOP_BUSINESS_ENVIRONMENT_REJECTED","当前环境不允许正式采集"),
Map.entry("SELFOP_BUDGET_INVALID","采集预算配置无效,请联系管理员"));
}
@@ -18,7 +18,7 @@ public class SyncController {
@PreAuthorize("hasAuthority('platform:self-operated-analytics:sync')")
@Operation(summary="创建自营分析人工采集或恢复任务",description="平台独立写权限 platform:self-operated-analytics:sync;持久化后异步执行,执行开关默认关闭。BACKFILL使用不超过一天的来源时间切片;未知源时区仅允许暂存。")
public ApiResponse<Accepted> request(@AuthenticationPrincipal AuthenticatedUser user,@RequestBody SelfopCommandService.Command command){
if(user==null||user.tenantId()==null||user.tenantId()!=0||!"PLATFORM".equals(user.appCode())||user.impersonated()||!user.permissions().contains("platform:self-operated-analytics:sync"))
if(user==null||user.userId()==null||user.tenantId()==null||user.tenantId()!=0||!"PLATFORM".equals(user.appCode())||user.impersonated()||!user.permissions().contains("platform:self-operated-analytics:sync"))
throw new ResponseStatusException(HttpStatus.FORBIDDEN,"仅限已授权平台人员");
var task=commands.request(command,String.valueOf(user.userId()));return ApiResponse.success(new Accepted(task.getTaskNo(),command.warehouseCode(),"ACCEPTED"));
}