Merge pull request 'feat(mcp): 增量同步框架与 users/properties/contracts changes 工具 (REQ-20260911-0013)' (#12) from feat/REQ-5628 into main
This commit was merged in pull request #12.
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
-- 回滚 V5__tool_description_length.sql
|
||||
-- 先截断超长描述再收窄列，避免 Data truncation
|
||||
UPDATE tool SET description = LEFT(description, 512) WHERE CHAR_LENGTH(description) > 512;
|
||||
|
||||
ALTER TABLE tool
|
||||
MODIFY COLUMN description VARCHAR(512) NULL COMMENT '工具描述';
|
||||
@@ -33,18 +33,17 @@ SELECT s.id, t.id, 0 FROM agent_scope s JOIN tool t
|
||||
WHERE s.code = 'hardware' AND t.name IN ('hardware.list.query', 'hardware.detail.query')
|
||||
AND NOT EXISTS (SELECT 1 FROM scope_tool st WHERE st.scope_id = s.id AND st.tool_id = t.id);
|
||||
|
||||
-- 作用域：sync（SYNC 增量同步，REQ-20260911-0014 预留；SYNC 工具本身是后续需求 5628）
|
||||
-- 作用域：sync（SYNC 增量同步；changes 工具由 REQ-20260911-0013 注册并绑定到本作用域）
|
||||
-- 个人 Token 禁调 SYNC 工具靠 token_scope 白名单天然生效：个人 Token 不绑定本作用域即 PERMISSION_DENIED，
|
||||
-- 无需任何拦截代码。5628 注册 *.changes.list 工具后，按上面同款 INSERT 把工具绑进本作用域，
|
||||
-- 再只为服务 Token（principal_type=service）勾选 sync 作用域。
|
||||
-- 无需任何拦截代码。只为服务 Token（principal_type=service）勾选 sync 作用域。
|
||||
INSERT INTO agent_scope (name, code, memo, status, create_id)
|
||||
SELECT '数据同步', 'sync', 'SYNC 增量同步作用域（5628 注册 *.changes.list 工具后绑定，仅服务 Token 使用）', 1, 0
|
||||
WHERE NOT EXISTS (SELECT 1 FROM agent_scope WHERE code = 'sync');
|
||||
|
||||
-- sync 作用域-工具：integration.context.get（服务 Token 自检入口；*.changes.list 由 5628 注册后追加）
|
||||
-- sync 作用域-工具：integration.context.get（服务 Token 自检入口）+ 三个 changes 工具（REQ-20260911-0013）
|
||||
INSERT INTO scope_tool (scope_id, tool_id, create_id)
|
||||
SELECT s.id, t.id, 0 FROM agent_scope s JOIN tool t
|
||||
WHERE s.code = 'sync' AND t.name IN ('integration.context.get')
|
||||
WHERE s.code = 'sync' AND t.name IN ('integration.context.get', 'users.changes.list', 'properties.changes.list', 'contracts.changes.list')
|
||||
AND NOT EXISTS (SELECT 1 FROM scope_tool st WHERE st.scope_id = s.id AND st.tool_id = t.id);
|
||||
|
||||
COMMIT;
|
||||
|
||||
@@ -31,12 +31,15 @@ public class M0DataSourceConfig {
|
||||
|
||||
@Primary
|
||||
@Bean(name = "m0SqlSessionFactory")
|
||||
public SqlSessionFactory m0SqlSessionFactory(@Qualifier("m0DataSource") DataSource dataSource) throws Exception {
|
||||
public SqlSessionFactory m0SqlSessionFactory(@Qualifier("m0DataSource") DataSource dataSource,
|
||||
SyncQueryTimeoutInterceptor syncQueryTimeoutInterceptor)
|
||||
throws Exception {
|
||||
MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
|
||||
bean.setDataSource(dataSource);
|
||||
MybatisConfiguration configuration = new MybatisConfiguration();
|
||||
configuration.setMapUnderscoreToCamelCase(true);
|
||||
bean.setConfiguration(configuration);
|
||||
bean.setPlugins(syncQueryTimeoutInterceptor);
|
||||
return bean.getObject();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,12 +27,15 @@ public class M1DataSourceConfig {
|
||||
}
|
||||
|
||||
@Bean(name = "m1SqlSessionFactory")
|
||||
public SqlSessionFactory m1SqlSessionFactory(@Qualifier("m1DataSource") DataSource dataSource) throws Exception {
|
||||
public SqlSessionFactory m1SqlSessionFactory(@Qualifier("m1DataSource") DataSource dataSource,
|
||||
SyncQueryTimeoutInterceptor syncQueryTimeoutInterceptor)
|
||||
throws Exception {
|
||||
MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
|
||||
bean.setDataSource(dataSource);
|
||||
MybatisConfiguration configuration = new MybatisConfiguration();
|
||||
configuration.setMapUnderscoreToCamelCase(true);
|
||||
bean.setConfiguration(configuration);
|
||||
bean.setPlugins(syncQueryTimeoutInterceptor);
|
||||
return bean.getObject();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,12 +27,15 @@ public class M2DataSourceConfig {
|
||||
}
|
||||
|
||||
@Bean(name = "m2SqlSessionFactory")
|
||||
public SqlSessionFactory m2SqlSessionFactory(@Qualifier("m2DataSource") DataSource dataSource) throws Exception {
|
||||
public SqlSessionFactory m2SqlSessionFactory(@Qualifier("m2DataSource") DataSource dataSource,
|
||||
SyncQueryTimeoutInterceptor syncQueryTimeoutInterceptor)
|
||||
throws Exception {
|
||||
MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
|
||||
bean.setDataSource(dataSource);
|
||||
MybatisConfiguration configuration = new MybatisConfiguration();
|
||||
configuration.setMapUnderscoreToCamelCase(true);
|
||||
bean.setConfiguration(configuration);
|
||||
bean.setPlugins(syncQueryTimeoutInterceptor);
|
||||
return bean.getObject();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package me.iiv.agent.config;
|
||||
|
||||
import me.iiv.agent.common.AuditService;
|
||||
import me.iiv.agent.mcp.GuardedToolCallbackProvider;
|
||||
import me.iiv.agent.mcp.RateLimitService;
|
||||
import me.iiv.agent.mcp.ToolObject;
|
||||
import me.iiv.agent.security.ToolGuard;
|
||||
import org.springframework.ai.tool.ToolCallbackProvider;
|
||||
@@ -20,10 +21,11 @@ import java.util.List;
|
||||
public class McpToolConfig {
|
||||
|
||||
@Bean
|
||||
public ToolCallbackProvider xiaoquTools(List<ToolObject> toolObjects, ToolGuard toolGuard, AuditService auditService) {
|
||||
public ToolCallbackProvider xiaoquTools(List<ToolObject> toolObjects, ToolGuard toolGuard,
|
||||
AuditService auditService, RateLimitService rateLimitService) {
|
||||
ToolCallbackProvider methodProvider = MethodToolCallbackProvider.builder()
|
||||
.toolObjects(toolObjects.toArray())
|
||||
.build();
|
||||
return new GuardedToolCallbackProvider(methodProvider, toolGuard, auditService);
|
||||
return new GuardedToolCallbackProvider(methodProvider, toolGuard, auditService, rateLimitService);
|
||||
}
|
||||
}
|
||||
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
package me.iiv.agent.config;
|
||||
|
||||
import me.iiv.agent.mcp.sync.SyncQueryTimeoutHolder;
|
||||
import org.apache.ibatis.executor.statement.StatementHandler;
|
||||
import org.apache.ibatis.plugin.Interceptor;
|
||||
import org.apache.ibatis.plugin.Intercepts;
|
||||
import org.apache.ibatis.plugin.Invocation;
|
||||
import org.apache.ibatis.plugin.Signature;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.Statement;
|
||||
|
||||
/**
|
||||
* 慢查询超时（REQ-20260911-0013 资源隔离）：SyncQueryTimeoutHolder 非空时，
|
||||
* 给本线程 prepare 出的 Statement 套 queryTimeout（秒），
|
||||
* 防止 SYNC 工具的后台全量扫描拖慢销售在线点查。未设置时完全不影响现有查询。
|
||||
*/
|
||||
@Component
|
||||
@Intercepts(@Signature(type = StatementHandler.class, method = "prepare",
|
||||
args = {Connection.class, Integer.class}))
|
||||
public class SyncQueryTimeoutInterceptor implements Interceptor {
|
||||
|
||||
@Override
|
||||
public Object intercept(Invocation invocation) throws Throwable {
|
||||
Object statement = invocation.proceed();
|
||||
Integer timeout = SyncQueryTimeoutHolder.current();
|
||||
if (timeout != null && timeout > 0 && statement instanceof Statement stmt) {
|
||||
stmt.setQueryTimeout(timeout);
|
||||
}
|
||||
return statement;
|
||||
}
|
||||
}
|
||||
@@ -21,19 +21,30 @@ import org.springframework.data.redis.RedisSystemException;
|
||||
* 所有异常收敛为 ToolCallException（message 为 {"error":{code,message,request_id}} JSON），
|
||||
* 上游不可用 → UPSTREAM_UNAVAILABLE，其余未预期异常 → DATA_INTEGRITY_ERROR 兜底，
|
||||
* 内部细节（含可能带手机号的异常文本）脱敏后只落服务端日志，不透出给客户端。
|
||||
* REQ-20260911-0013 追加：分档限流（SYNC/QUERY 互不占配额，超额 RATE_LIMITED）+
|
||||
* *.changes.list 工具的 tenant_id/scope 入参硬拒绝（数据域只能从 Token 推导）。
|
||||
*/
|
||||
public class GuardedToolCallback implements ToolCallback {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(GuardedToolCallback.class);
|
||||
|
||||
/** SYNC 工具数据域硬约束：禁止调用方用参数覆盖 Token 推导的数据域（REQ-20260911-0013） */
|
||||
private static final java.util.Set<String> SYNC_FORBIDDEN_PARAMS = java.util.Set.of(
|
||||
"tenant_id", "tenantId", "scope");
|
||||
|
||||
private final ToolCallback delegate;
|
||||
private final ToolGuard toolGuard;
|
||||
private final AuditService auditService;
|
||||
private final RateLimitService rateLimitService;
|
||||
private final com.fasterxml.jackson.databind.ObjectMapper objectMapper =
|
||||
new com.fasterxml.jackson.databind.ObjectMapper();
|
||||
|
||||
public GuardedToolCallback(ToolCallback delegate, ToolGuard toolGuard, AuditService auditService) {
|
||||
public GuardedToolCallback(ToolCallback delegate, ToolGuard toolGuard, AuditService auditService,
|
||||
RateLimitService rateLimitService) {
|
||||
this.delegate = delegate;
|
||||
this.toolGuard = toolGuard;
|
||||
this.auditService = auditService;
|
||||
this.rateLimitService = rateLimitService;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -71,6 +82,8 @@ public class GuardedToolCallback implements ToolCallback {
|
||||
throw denied;
|
||||
}
|
||||
try {
|
||||
rateLimitService.check(ctx.getTokenId(), toolName);
|
||||
rejectForbiddenSyncParams(toolName, toolInput);
|
||||
String result = toolContext == null ? delegate.call(toolInput) : delegate.call(toolInput, toolContext);
|
||||
auditService.record(ctx, toolName, toolInput, "success", null, System.currentTimeMillis() - start);
|
||||
return result;
|
||||
@@ -79,6 +92,33 @@ public class GuardedToolCallback implements ToolCallback {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* SYNC 工具（*.changes.list）数据域硬约束：tenant_id/scope 入参一律 PERMISSION_DENIED。
|
||||
* 工具方法签名本就不含这些参数，这里拦的是调用方在原始 JSON 里多传的键
|
||||
* （Spring AI 对未知参数默认宽松，不能依赖反序列化兜底）。
|
||||
*/
|
||||
private void rejectForbiddenSyncParams(String toolName, String toolInput) {
|
||||
if (!toolName.endsWith(IntegrationContextService.CHANGES_TOOL_SUFFIX)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
com.fasterxml.jackson.databind.JsonNode args = objectMapper.readTree(toolInput);
|
||||
if (!args.isObject()) {
|
||||
return;
|
||||
}
|
||||
for (String forbidden : SYNC_FORBIDDEN_PARAMS) {
|
||||
if (args.has(forbidden)) {
|
||||
throw new ToolCallException(McpErrorCode.PERMISSION_DENIED,
|
||||
"SYNC 工具数据域只能从 Token 推导，禁止传 " + forbidden + " 入参");
|
||||
}
|
||||
}
|
||||
} catch (ToolCallException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
// 非 JSON 入参交给下游反序列化报错，此处不拦截
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 异常 → 统一错误码。MethodToolCallback 会把工具方法内抛出的异常包成
|
||||
* ToolExecutionException（message=cause.message），先拆包取根因再分类。
|
||||
|
||||
+5
-2
@@ -15,17 +15,20 @@ public class GuardedToolCallbackProvider implements ToolCallbackProvider {
|
||||
private final ToolCallbackProvider delegate;
|
||||
private final ToolGuard toolGuard;
|
||||
private final AuditService auditService;
|
||||
private final RateLimitService rateLimitService;
|
||||
|
||||
public GuardedToolCallbackProvider(ToolCallbackProvider delegate, ToolGuard toolGuard, AuditService auditService) {
|
||||
public GuardedToolCallbackProvider(ToolCallbackProvider delegate, ToolGuard toolGuard,
|
||||
AuditService auditService, RateLimitService rateLimitService) {
|
||||
this.delegate = delegate;
|
||||
this.toolGuard = toolGuard;
|
||||
this.auditService = auditService;
|
||||
this.rateLimitService = rateLimitService;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ToolCallback[] getToolCallbacks() {
|
||||
return Arrays.stream(delegate.getToolCallbacks())
|
||||
.map(callback -> new GuardedToolCallback(callback, toolGuard, auditService))
|
||||
.map(callback -> new GuardedToolCallback(callback, toolGuard, auditService, rateLimitService))
|
||||
.toArray(ToolCallback[]::new);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
package me.iiv.agent.mcp;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.data.redis.RedisSystemException;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
|
||||
/**
|
||||
* 工具调用限流（REQ-20260911-0013 资源隔离）：按 Token 分档计数——
|
||||
* *.changes.list 走 SYNC 档位，其余走 QUERY 档位，互不占配额。
|
||||
* Redis 固定窗口（分钟），超额抛 RATE_LIMITED；Redis 不可用时放行并告警
|
||||
* （限流是保护措施，不应成为可用性单点）。
|
||||
* 配额 0 或负数 = 关闭该档位。
|
||||
*/
|
||||
@Component
|
||||
public class RateLimitService {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(RateLimitService.class);
|
||||
|
||||
private final StringRedisTemplate redisTemplate;
|
||||
private final int queryPerMinute;
|
||||
private final int syncPerMinute;
|
||||
|
||||
public RateLimitService(StringRedisTemplate redisTemplate,
|
||||
@Value("${app.rate-limit.query-per-minute:600}") int queryPerMinute,
|
||||
@Value("${app.rate-limit.sync-per-minute:120}") int syncPerMinute) {
|
||||
this.redisTemplate = redisTemplate;
|
||||
this.queryPerMinute = queryPerMinute;
|
||||
this.syncPerMinute = syncPerMinute;
|
||||
}
|
||||
|
||||
public void check(Long tokenId, String toolName) {
|
||||
if (tokenId == null) {
|
||||
return;
|
||||
}
|
||||
boolean sync = toolName.endsWith(IntegrationContextService.CHANGES_TOOL_SUFFIX);
|
||||
int limit = sync ? syncPerMinute : queryPerMinute;
|
||||
if (limit <= 0) {
|
||||
return;
|
||||
}
|
||||
String tier = sync ? "sync" : "query";
|
||||
String key = "xq:rl:" + tier + ":" + tokenId + ":" + Instant.now().getEpochSecond() / 60;
|
||||
try {
|
||||
Long count = redisTemplate.opsForValue().increment(key);
|
||||
if (count != null && count == 1L) {
|
||||
redisTemplate.expire(key, Duration.ofSeconds(120));
|
||||
}
|
||||
if (count != null && count > limit) {
|
||||
throw new ToolCallException(McpErrorCode.RATE_LIMITED,
|
||||
"触发限流（" + tier + " 档位每 Token 每分钟 " + limit + " 次），请稍后重试");
|
||||
}
|
||||
} catch (ToolCallException e) {
|
||||
throw e;
|
||||
} catch (RedisSystemException e) {
|
||||
log.warn("限流计数失败（Redis 不可用），本次放行: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* 同步实体公共部分：(fromTs 不含, cutoffTs 含] 窗口 + (生效修改时间, id) 复合游标推进。
|
||||
* 生效修改时间一律 COALESCE(modify_time, create_time)——m0.user 存在 modify_time 为 NULL 的存量行，
|
||||
* 直接按 modify_time 过滤会丢数据（2026-09-14 核对：dev 4 行）。
|
||||
* MySQL 5.7 秒级精度，复合游标 (时间, id) 保证同秒多行不丢不重。
|
||||
*/
|
||||
abstract class AbstractSyncEntity {
|
||||
|
||||
static final String EFFECTIVE_TIME = "COALESCE(modify_time, create_time)";
|
||||
|
||||
<T> void applyWindow(LambdaQueryWrapper<T> wrapper, Long fromTs, long cutoffTs,
|
||||
Long cursorMt, Long cursorId) {
|
||||
wrapper.apply(EFFECTIVE_TIME + " <= {0}", new Date(cutoffTs * 1000));
|
||||
if (fromTs != null) {
|
||||
wrapper.apply(EFFECTIVE_TIME + " > {0}", new Date(fromTs * 1000));
|
||||
}
|
||||
if (cursorMt != null && cursorId != null) {
|
||||
wrapper.apply("(" + EFFECTIVE_TIME + " > {0} OR (" + EFFECTIVE_TIME + " = {0} AND id > {1}))",
|
||||
new Date(cursorMt * 1000), cursorId);
|
||||
}
|
||||
}
|
||||
|
||||
/** 正序 + 条数上限（int 直拼，无注入面）。 */
|
||||
<T> void applyOrderAndLimit(LambdaQueryWrapper<T> wrapper, int limit) {
|
||||
wrapper.last("ORDER BY " + EFFECTIVE_TIME + " ASC, id ASC LIMIT " + limit);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.mcp.IntegrationContextService;
|
||||
import me.iiv.agent.mcp.McpErrorCode;
|
||||
import me.iiv.agent.mcp.ToolCallException;
|
||||
import me.iiv.agent.security.AuthContext;
|
||||
import me.iiv.agent.security.AuthContextHolder;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.time.OffsetDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* 通用 changes 同步引擎（REQ-20260911-0013，协议只实现一次）：
|
||||
* <ul>
|
||||
* <li>每轮同步签发 snapshot_id 钉住 data_cutoff_at，页间按 (modify_time, id) 复合游标推进，
|
||||
* 不依赖页码；</li>
|
||||
* <li>中间页原样回传 sync_token + next_cursor，只有最后一页（has_more=false）签发
|
||||
* next_sync_token（from=本轮 cutoff）；</li>
|
||||
* <li>软删行由实体输出 tombstone（deleted_at 非空），游标过期/篡改/跨快照/数据域变更
|
||||
* 统一 SYNC_CURSOR_EXPIRED；</li>
|
||||
* <li>数据域只能从 Token 推导：本引擎无任何 tenant_id/scope 入参，行级权限跟随
|
||||
* Token 绑定的人（服务 Token 无人员绑定，不加行过滤）。</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Component
|
||||
public class ChangesEngine {
|
||||
|
||||
/** 业务时间字段的展示时区（老系统全部为北京时间） */
|
||||
public static final ZoneId ZONE = ZoneId.of("Asia/Shanghai");
|
||||
|
||||
private final SyncEntityRegistry registry;
|
||||
private final SyncTokenCodec codec;
|
||||
private final IntegrationContextService integrationContextService;
|
||||
private final int queryTimeoutSeconds;
|
||||
private final int defaultLimit;
|
||||
private final int maxLimit;
|
||||
|
||||
public ChangesEngine(SyncEntityRegistry registry, SyncTokenCodec codec,
|
||||
IntegrationContextService integrationContextService,
|
||||
@Value("${app.sync.query-timeout-seconds:10}") int queryTimeoutSeconds,
|
||||
@Value("${app.sync.default-limit:100}") int defaultLimit,
|
||||
@Value("${app.sync.max-limit:200}") int maxLimit) {
|
||||
this.registry = registry;
|
||||
this.codec = codec;
|
||||
this.integrationContextService = integrationContextService;
|
||||
this.queryTimeoutSeconds = queryTimeoutSeconds;
|
||||
this.defaultLimit = defaultLimit;
|
||||
this.maxLimit = maxLimit;
|
||||
}
|
||||
|
||||
public Map<String, Object> execute(String entityName, String syncToken, String cursor, Integer limit) {
|
||||
SyncEntity entity = registry.get(entityName);
|
||||
if (entity == null) {
|
||||
throw new ToolCallException(McpErrorCode.DATA_INTEGRITY_ERROR, "同步实体未注册: " + entityName);
|
||||
}
|
||||
AuthContext ctx = AuthContextHolder.get();
|
||||
long scopeVersion = dataScopeVersion(ctx);
|
||||
boolean truncated = limit != null && limit > maxLimit;
|
||||
int pageSize = limit == null || limit < 1 ? defaultLimit : Math.min(limit, maxLimit);
|
||||
|
||||
// 1. 解析/签发本轮锚点：from（下界，不含）+ cutoff（上界，含）+ snapshot
|
||||
Long fromTs = null;
|
||||
long cutoffTs;
|
||||
String snapshotId;
|
||||
if (!StringUtils.hasText(syncToken)) {
|
||||
cutoffTs = nowSec();
|
||||
snapshotId = newSnapshotId();
|
||||
} else {
|
||||
SyncTokenCodec.SyncToken token = codec.parseToken(syncToken, entityName);
|
||||
if (token.scopeVersion() != scopeVersion) {
|
||||
throw codec.expired("数据域已变更（data_scope_version 不一致），请不带 sync_token 重新全量同步");
|
||||
}
|
||||
fromTs = token.fromTs();
|
||||
if (SyncTokenCodec.TYPE_ROUND.equals(token.type())) {
|
||||
// 轮内锚点：钉住的 cutoff/snapshot 不变；不带 cursor 表示从本轮开头重读
|
||||
cutoffTs = token.cutoffTs();
|
||||
snapshotId = token.snapshotId();
|
||||
} else {
|
||||
cutoffTs = nowSec();
|
||||
snapshotId = newSnapshotId();
|
||||
}
|
||||
}
|
||||
|
||||
// 2. 页内游标：必须与本轮同一快照
|
||||
Long cursorMt = null;
|
||||
Long cursorId = null;
|
||||
if (StringUtils.hasText(cursor)) {
|
||||
SyncTokenCodec.Cursor c = codec.parseCursor(cursor);
|
||||
if (!c.snapshotId().equals(snapshotId)) {
|
||||
throw codec.expired("cursor 与 sync_token 不属于同一快照，请使用该快照首屏返回的游标链");
|
||||
}
|
||||
cursorMt = c.modifyTs();
|
||||
cursorId = c.id();
|
||||
}
|
||||
|
||||
// 3. 拉页（多取一行判 has_more），套用慢查询超时
|
||||
List<SyncRow> rows;
|
||||
SyncQueryTimeoutHolder.set(queryTimeoutSeconds);
|
||||
try {
|
||||
rows = entity.fetchPage(fromTs, cutoffTs, cursorMt, cursorId, pageSize + 1,
|
||||
ctx == null ? null : ctx.getUser());
|
||||
} finally {
|
||||
SyncQueryTimeoutHolder.clear();
|
||||
}
|
||||
boolean hasMore = rows.size() > pageSize;
|
||||
if (hasMore) {
|
||||
rows = new ArrayList<>(rows.subList(0, pageSize));
|
||||
}
|
||||
|
||||
// 4. 组装响应：中间页回传锚点 + next_cursor；最后一页签发 next_sync_token
|
||||
String roundToken = codec.encodeRound(entityName, fromTs, cutoffTs, snapshotId, scopeVersion);
|
||||
Map<String, Object> result = new LinkedHashMap<>();
|
||||
result.put("entity", entityName);
|
||||
result.put("snapshot_id", snapshotId);
|
||||
result.put("data_cutoff_at", iso(new Date(cutoffTs * 1000)));
|
||||
result.put("limit", pageSize);
|
||||
if (truncated) {
|
||||
result.put("notice", "limit 超过上限 " + maxLimit + "，已截断为 " + maxLimit);
|
||||
}
|
||||
List<Map<String, Object>> items = new ArrayList<>();
|
||||
for (SyncRow row : rows) {
|
||||
items.add(row.item());
|
||||
}
|
||||
result.put("count", items.size());
|
||||
result.put("items", items);
|
||||
result.put("has_more", hasMore);
|
||||
result.put("sync_token", roundToken);
|
||||
if (hasMore) {
|
||||
SyncRow last = rows.get(rows.size() - 1);
|
||||
result.put("next_cursor", codec.encodeCursor(snapshotId, last.modifyTs(), last.id()));
|
||||
result.put("next_sync_token", null);
|
||||
} else {
|
||||
result.put("next_cursor", null);
|
||||
result.put("next_sync_token", codec.encodeHandoff(entityName, cutoffTs, scopeVersion));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private long dataScopeVersion(AuthContext ctx) {
|
||||
if (ctx == null || ctx.getTokenId() == null) {
|
||||
return 0L;
|
||||
}
|
||||
Object version = integrationContextService.buildContext(ctx).get("data_scope_version");
|
||||
return version instanceof Number number ? number.longValue() : 0L;
|
||||
}
|
||||
|
||||
private static String newSnapshotId() {
|
||||
return UUID.randomUUID().toString();
|
||||
}
|
||||
|
||||
private static long nowSec() {
|
||||
return System.currentTimeMillis() / 1000;
|
||||
}
|
||||
|
||||
/** Date（含 java.sql.Date）→ 带时区 ISO 串；null 原样返回 null。 */
|
||||
public static String iso(Date date) {
|
||||
if (date == null) {
|
||||
return null;
|
||||
}
|
||||
return OffsetDateTime.ofInstant(java.time.Instant.ofEpochMilli(date.getTime()), ZONE)
|
||||
.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);
|
||||
}
|
||||
|
||||
/** 生效修改时间 epoch 秒：modify_time 为空时回退 create_time，再空为 0。 */
|
||||
public static long effectiveTs(Date modifyTime, Date createTime) {
|
||||
Date time = modifyTime != null ? modifyTime : createTime;
|
||||
return time == null ? 0L : time.getTime() / 1000;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import me.iiv.agent.entity.m0.Properyt;
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.entity.m2.ContractNew;
|
||||
import me.iiv.agent.mapper.m0.ProperytMapper;
|
||||
import me.iiv.agent.mapper.m2.ContractNewMapper;
|
||||
import me.iiv.agent.permission.PermissionService;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* contracts 同步实体（m2.contract_new 合同主数据）。
|
||||
* 字段口径（2026-09-14 字段核对）：owner_user_id → sale_id（稳定 user.id）；
|
||||
* customer_id 经 property_id 回查 properyt.customerId（金蝶客户 id）；金额转分（amount_minor）；
|
||||
* start_at/end_at 为带时区 ISO；version 由 modify_time 派生（表无 version 列，决策 1）。
|
||||
* 表无 currency / renewal_of_contract_id / renewal_result / deleted_at 列，按铁律"表里不存在的字段不造"省略，
|
||||
* 软删 tombstone 的 deleted_at 以 modify_time 充当。
|
||||
* 行级权限：Token 绑定人时走 PermissionService.applyContractRowFilter（与 contract.list.query 同一 ACL）。
|
||||
*/
|
||||
@Component
|
||||
public class ContractsSyncEntity extends AbstractSyncEntity implements SyncEntity {
|
||||
|
||||
public static final String NAME = "contracts";
|
||||
|
||||
final ContractNewMapper contractNewMapper;
|
||||
final ProperytMapper properytMapper;
|
||||
final PermissionService permissionService;
|
||||
|
||||
public ContractsSyncEntity(ContractNewMapper contractNewMapper, ProperytMapper properytMapper,
|
||||
PermissionService permissionService) {
|
||||
this.contractNewMapper = contractNewMapper;
|
||||
this.properytMapper = properytMapper;
|
||||
this.permissionService = permissionService;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<SyncRow> fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId,
|
||||
int limit, User user) {
|
||||
LambdaQueryWrapper<ContractNew> wrapper = new LambdaQueryWrapper<>();
|
||||
if (user != null) {
|
||||
permissionService.applyContractRowFilter(wrapper, user);
|
||||
}
|
||||
applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId);
|
||||
applyOrderAndLimit(wrapper, limit);
|
||||
List<ContractNew> contracts = contractNewMapper.selectList(wrapper);
|
||||
|
||||
// customer_id 批量富化（property_id → properyt.customerId，一次 IN 查询，消除 N+1）
|
||||
List<Long> propertyIds = contracts.stream().map(ContractNew::getPropertyId)
|
||||
.filter(Objects::nonNull).distinct().collect(Collectors.toList());
|
||||
Map<Long, String> customerIds = propertyIds.isEmpty()
|
||||
? Map.of()
|
||||
: customerIdsByProperty(propertyIds);
|
||||
|
||||
List<SyncRow> rows = new ArrayList<>();
|
||||
for (ContractNew row : contracts) {
|
||||
long mt = ChangesEngine.effectiveTs(row.getModifyTime(), row.getCreateTime());
|
||||
rows.add(new SyncRow(toItem(row, mt, customerIds), mt, row.getId()));
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
private Map<String, Object> toItem(ContractNew row, long mt, Map<Long, String> customerIds) {
|
||||
boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1;
|
||||
Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime();
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("contract_id", row.getId());
|
||||
item.put("contract_no", row.getContractNo());
|
||||
item.put("contract_name", row.getContractName());
|
||||
item.put("status", row.getContractStatus());
|
||||
item.put("start_at", ChangesEngine.iso(row.getStartDate()));
|
||||
item.put("end_at", ChangesEngine.iso(row.getEndDate()));
|
||||
item.put("amount_minor", toMinor(row.getAmount()));
|
||||
item.put("customer_id", row.getPropertyId() == null ? null : customerIds.get(row.getPropertyId()));
|
||||
item.put("property_id", row.getPropertyId());
|
||||
item.put("owner_user_id", row.getSaleId());
|
||||
item.put("expire_remind", row.getExpireRemind());
|
||||
item.put("updated_at", ChangesEngine.iso(effective));
|
||||
item.put("version", mt);
|
||||
item.put("deleted_at", deleted ? ChangesEngine.iso(effective) : null);
|
||||
return item;
|
||||
}
|
||||
|
||||
/** 元 → 分（amount 为 decimal(14,2)，精确换算）；null 原样返回 null。 */
|
||||
static Long toMinor(BigDecimal amount) {
|
||||
return amount == null ? null : amount.movePointRight(2).longValue();
|
||||
}
|
||||
|
||||
/**
|
||||
* property_id → customerId 映射。properyt.customerId 存在 NULL 行（2026-09-14 dev 冒烟命中），
|
||||
* Collectors.toMap 遇 null 值抛 NPE，必须用 put 逐条装填。
|
||||
*/
|
||||
private Map<Long, String> customerIdsByProperty(List<Long> propertyIds) {
|
||||
Map<Long, String> map = new LinkedHashMap<>();
|
||||
for (Properyt property : properytMapper.selectBatchIds(propertyIds)) {
|
||||
map.put(property.getId(), property.getCustomerId());
|
||||
}
|
||||
return map;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import me.iiv.agent.entity.m0.Properyt;
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.mapper.m0.ProperytMapper;
|
||||
import me.iiv.agent.permission.PermissionService;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* properties 同步实体（m0.properyt 物业/客户主数据）。
|
||||
* 字段口径（2026-09-14 字段核对）：unified_social_credit_code → identification_number，
|
||||
* external_customer_no → serialnumber，customer_id → customerId（金蝶客户 id），
|
||||
* customer_name → jinide_name（金蝶客户名称，与合同快照 customer_name 同源），
|
||||
* status 对齐 soft_delete。行级权限：Token 绑定人时走 PermissionService.applyPropertyRowFilter
|
||||
* （与 property.list.query 同一 ACL）；服务 Token 无人员绑定，不加行过滤。
|
||||
*/
|
||||
@Component
|
||||
public class PropertiesSyncEntity extends AbstractSyncEntity implements SyncEntity {
|
||||
|
||||
public static final String NAME = "properties";
|
||||
|
||||
final ProperytMapper properytMapper;
|
||||
final PermissionService permissionService;
|
||||
|
||||
public PropertiesSyncEntity(ProperytMapper properytMapper, PermissionService permissionService) {
|
||||
this.properytMapper = properytMapper;
|
||||
this.permissionService = permissionService;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<SyncRow> fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId,
|
||||
int limit, User user) {
|
||||
LambdaQueryWrapper<Properyt> wrapper = new LambdaQueryWrapper<>();
|
||||
if (user != null) {
|
||||
permissionService.applyPropertyRowFilter(wrapper, user);
|
||||
}
|
||||
applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId);
|
||||
applyOrderAndLimit(wrapper, limit);
|
||||
List<SyncRow> rows = new ArrayList<>();
|
||||
for (Properyt row : properytMapper.selectList(wrapper)) {
|
||||
long mt = ChangesEngine.effectiveTs(row.getModifyTime(), row.getCreateTime());
|
||||
rows.add(new SyncRow(toItem(row, mt), mt, row.getId()));
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
private Map<String, Object> toItem(Properyt row, long mt) {
|
||||
boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1;
|
||||
Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime();
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("property_id", row.getId());
|
||||
item.put("customer_id", row.getCustomerId());
|
||||
item.put("property_name", row.getName());
|
||||
item.put("customer_name", row.getJinideName());
|
||||
item.put("unified_social_credit_code", row.getIdentificationNumber());
|
||||
item.put("external_customer_no", row.getSerialnumber());
|
||||
item.put("status", deleted ? "inactive" : "active");
|
||||
item.put("updated_at", ChangesEngine.iso(effective));
|
||||
item.put("version", mt);
|
||||
item.put("deleted_at", deleted ? ChangesEngine.iso(effective) : null);
|
||||
return item;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 增量同步实体注册接口（REQ-20260911-0013）：协议（snapshot/cursor/tombstone/限流/超时）
|
||||
* 只在 ChangesEngine 实现一次，新增实体只需 @Component 实现本接口并声明 name()，
|
||||
* 再在 ChangesTools 加同名 *.changes.list 工具方法。
|
||||
*/
|
||||
public interface SyncEntity {
|
||||
|
||||
/** 实体名（users / properties / contracts），与工具名前缀一致 */
|
||||
String name();
|
||||
|
||||
/**
|
||||
* 拉取一页：(fromTs 不含, cutoffTs 含] 窗口内、游标 (cursorMt, cursorId) 之后，
|
||||
* 按 (生效修改时间, id) 正序，最多 limit 行。生效修改时间 = COALESCE(modify_time, create_time)，
|
||||
* 软删行必须包含在结果内（由实现方输出 tombstone）。
|
||||
*
|
||||
* @param user 当前 Token 绑定的人（服务 Token 为 null，表示不加行级过滤）
|
||||
*/
|
||||
List<SyncRow> fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId,
|
||||
int limit, User user);
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 同步实体登记表：启动时收集全部 SyncEntity 实现，按 name() 索引。
|
||||
*/
|
||||
@Component
|
||||
public class SyncEntityRegistry {
|
||||
|
||||
private final Map<String, SyncEntity> byName;
|
||||
|
||||
public SyncEntityRegistry(List<SyncEntity> entities) {
|
||||
Map<String, SyncEntity> map = new HashMap<>();
|
||||
for (SyncEntity entity : entities) {
|
||||
map.put(entity.name(), entity);
|
||||
}
|
||||
this.byName = Map.copyOf(map);
|
||||
}
|
||||
|
||||
public SyncEntity get(String name) {
|
||||
return byName.get(name);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
/**
|
||||
* 慢查询超时开关（ThreadLocal 秒数）：ChangesEngine 在拉页前 set、finally 中 clear；
|
||||
* SyncQueryTimeoutInterceptor 在 prepare Statement 时读取并套用 queryTimeout，
|
||||
* 防止后台全量扫描拖慢在线点查（REQ-20260911-0013 资源隔离）。
|
||||
*/
|
||||
public final class SyncQueryTimeoutHolder {
|
||||
|
||||
private static final ThreadLocal<Integer> TIMEOUT_SECONDS = new ThreadLocal<>();
|
||||
|
||||
private SyncQueryTimeoutHolder() {
|
||||
}
|
||||
|
||||
public static void set(int seconds) {
|
||||
TIMEOUT_SECONDS.set(seconds);
|
||||
}
|
||||
|
||||
public static Integer current() {
|
||||
return TIMEOUT_SECONDS.get();
|
||||
}
|
||||
|
||||
public static void clear() {
|
||||
TIMEOUT_SECONDS.remove();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 一页中的一行：已映射的对外字段 + 引擎推进游标所需的 (生效修改时间 epoch 秒, id)。
|
||||
*/
|
||||
public record SyncRow(Map<String, Object> item, long modifyTs, long id) {
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.node.ObjectNode;
|
||||
import me.iiv.agent.mcp.McpErrorCode;
|
||||
import me.iiv.agent.mcp.ToolCallException;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.crypto.Mac;
|
||||
import javax.crypto.spec.SecretKeySpec;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.util.Base64;
|
||||
|
||||
/**
|
||||
* sync_token / cursor 的签发与校验（REQ-20260911-0013）：无状态 HMAC-SHA256 签名令牌，
|
||||
* 服务端不存快照状态，快照（snapshot_id + data_cutoff_at）由令牌自身钉住。
|
||||
* <ul>
|
||||
* <li>anchor 令牌（typ=r）：一轮同步的锚点，含 from/cutoff/snap，页间原样回传；</li>
|
||||
* <li>handoff 令牌（typ=n）：一轮最后一页签发的 next_sync_token，只含 from（=上轮 cutoff），
|
||||
* 下一轮首次调用时由服务端钉新的 cutoff 与 snapshot；</li>
|
||||
* <li>cursor（typ=c）：页内位置 (modify_time, id)，绑定 snapshot，跨快照使用判过期。</li>
|
||||
* </ul>
|
||||
* 签名/结构/TTL/实体不匹配一律抛 SYNC_CURSOR_EXPIRED（调用方重新全量同步即可恢复）。
|
||||
* 密钥：app.sync.token-key；为空时用 app.token-cipher-key 加域分隔派生，两者皆空则同步功能不可用。
|
||||
*/
|
||||
@Component
|
||||
public class SyncTokenCodec {
|
||||
|
||||
private static final ObjectMapper MAPPER = new ObjectMapper();
|
||||
private static final Base64.Encoder B64 = Base64.getUrlEncoder().withoutPadding();
|
||||
private static final Base64.Decoder B64D = Base64.getUrlDecoder();
|
||||
|
||||
/** 令牌类型：一轮同步锚点 */
|
||||
public static final String TYPE_ROUND = "r";
|
||||
/** 令牌类型：下一轮交接（next_sync_token） */
|
||||
public static final String TYPE_HANDOFF = "n";
|
||||
|
||||
private final String tokenKey;
|
||||
private final String cipherKey;
|
||||
private final long cursorTtlHours;
|
||||
|
||||
public SyncTokenCodec(@Value("${app.sync.token-key:}") String tokenKey,
|
||||
@Value("${app.token-cipher-key:}") String cipherKey,
|
||||
@Value("${app.sync.cursor-ttl-hours:24}") long cursorTtlHours) {
|
||||
this.tokenKey = tokenKey;
|
||||
this.cipherKey = cipherKey;
|
||||
this.cursorTtlHours = cursorTtlHours;
|
||||
}
|
||||
|
||||
/** 一轮同步锚点令牌。fromTs 为 null 表示全量（无下界）。 */
|
||||
public String encodeRound(String entity, Long fromTs, long cutoffTs, String snapshotId, long scopeVersion) {
|
||||
ObjectNode node = MAPPER.createObjectNode();
|
||||
node.put("typ", TYPE_ROUND);
|
||||
node.put("entity", entity);
|
||||
if (fromTs == null) {
|
||||
node.putNull("from");
|
||||
} else {
|
||||
node.put("from", fromTs);
|
||||
}
|
||||
node.put("cutoff", cutoffTs);
|
||||
node.put("snap", snapshotId);
|
||||
node.put("iat", nowSec());
|
||||
node.put("sv", scopeVersion);
|
||||
return pack(node);
|
||||
}
|
||||
|
||||
/** 下一轮交接令牌（next_sync_token）：from=已完成轮的 cutoff，cutoff/snap 待下轮首调用钉住。 */
|
||||
public String encodeHandoff(String entity, long fromTs, long scopeVersion) {
|
||||
ObjectNode node = MAPPER.createObjectNode();
|
||||
node.put("typ", TYPE_HANDOFF);
|
||||
node.put("entity", entity);
|
||||
node.put("from", fromTs);
|
||||
node.put("iat", nowSec());
|
||||
node.put("sv", scopeVersion);
|
||||
return pack(node);
|
||||
}
|
||||
|
||||
/** 页内游标：绑定快照的 (modify_time epoch 秒, id)。 */
|
||||
public String encodeCursor(String snapshotId, long modifyTs, long id) {
|
||||
ObjectNode node = MAPPER.createObjectNode();
|
||||
node.put("typ", "c");
|
||||
node.put("snap", snapshotId);
|
||||
node.put("mt", modifyTs);
|
||||
node.put("id", id);
|
||||
node.put("iat", nowSec());
|
||||
return pack(node);
|
||||
}
|
||||
|
||||
public SyncToken parseToken(String token, String expectedEntity) {
|
||||
ObjectNode node = unpack(token, "sync_token");
|
||||
String typ = text(node, "typ");
|
||||
if (!TYPE_ROUND.equals(typ) && !TYPE_HANDOFF.equals(typ)) {
|
||||
throw expired("sync_token 类型不受支持，请重新全量同步");
|
||||
}
|
||||
String entity = text(node, "entity");
|
||||
if (entity == null || !entity.equals(expectedEntity)) {
|
||||
throw expired("sync_token 与当前同步实体不匹配，请使用对应工具签发的 sync_token");
|
||||
}
|
||||
checkTtl(node, "sync_token");
|
||||
JsonNode from = node.get("from");
|
||||
JsonNode cutoff = node.get("cutoff");
|
||||
if (TYPE_ROUND.equals(typ) && (cutoff == null || !cutoff.canConvertToLong())) {
|
||||
throw expired("sync_token 缺少快照截止时间，请重新全量同步");
|
||||
}
|
||||
return new SyncToken(typ, entity,
|
||||
from == null || from.isNull() ? null : from.asLong(),
|
||||
cutoff == null || cutoff.isNull() ? null : cutoff.asLong(),
|
||||
text(node, "snap"), node.path("sv").asLong(0L));
|
||||
}
|
||||
|
||||
public Cursor parseCursor(String cursor) {
|
||||
ObjectNode node = unpack(cursor, "cursor");
|
||||
if (!"c".equals(text(node, "typ"))) {
|
||||
throw expired("cursor 类型不受支持，请使用上一页返回的 next_cursor");
|
||||
}
|
||||
checkTtl(node, "cursor");
|
||||
JsonNode mt = node.get("mt");
|
||||
JsonNode id = node.get("id");
|
||||
String snap = text(node, "snap");
|
||||
if (mt == null || !mt.canConvertToLong() || id == null || !id.canConvertToLong() || snap == null) {
|
||||
throw expired("cursor 结构不完整，请使用上一页返回的 next_cursor");
|
||||
}
|
||||
return new Cursor(snap, mt.asLong(), id.asLong());
|
||||
}
|
||||
|
||||
/** 游标/令牌过期（TTL、篡改、跨快照、跨实体、数据域变更均归一到此码，调用方重新全量即可恢复）。 */
|
||||
public ToolCallException expired(String message) {
|
||||
return new ToolCallException(McpErrorCode.SYNC_CURSOR_EXPIRED, message);
|
||||
}
|
||||
|
||||
private void checkTtl(ObjectNode node, String kind) {
|
||||
long iat = node.path("iat").asLong(0L);
|
||||
if (iat <= 0 || nowSec() - iat > cursorTtlHours * 3600) {
|
||||
throw expired(kind + " 已过期（快照有效期 " + cursorTtlHours + " 小时），请重新全量同步");
|
||||
}
|
||||
}
|
||||
|
||||
private String pack(ObjectNode node) {
|
||||
try {
|
||||
byte[] payload = MAPPER.writeValueAsBytes(node);
|
||||
String body = B64.encodeToString(payload);
|
||||
return body + "." + B64.encodeToString(hmac(payload));
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("sync_token 签发失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
private ObjectNode unpack(String token, String kind) {
|
||||
try {
|
||||
String[] parts = token == null ? new String[0] : token.split("\\.");
|
||||
if (parts.length != 2) {
|
||||
throw new IllegalArgumentException("bad format");
|
||||
}
|
||||
byte[] payload = B64D.decode(parts[0]);
|
||||
byte[] mac = B64D.decode(parts[1]);
|
||||
if (!MessageDigest.isEqual(hmac(payload), mac)) {
|
||||
throw new IllegalArgumentException("bad signature");
|
||||
}
|
||||
JsonNode node = MAPPER.readTree(payload);
|
||||
if (!node.isObject()) {
|
||||
throw new IllegalArgumentException("bad payload");
|
||||
}
|
||||
return (ObjectNode) node;
|
||||
} catch (ToolCallException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
throw expired(kind + " 无效或已被篡改，请重新全量同步");
|
||||
}
|
||||
}
|
||||
|
||||
private byte[] hmac(byte[] payload) {
|
||||
try {
|
||||
Mac mac = Mac.getInstance("HmacSHA256");
|
||||
mac.init(new SecretKeySpec(requireKey(), "HmacSHA256"));
|
||||
return mac.doFinal(payload);
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("sync_token 签名失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
private byte[] requireKey() {
|
||||
try {
|
||||
String material = tokenKey != null && !tokenKey.isBlank() ? tokenKey.trim() : null;
|
||||
if (material == null && cipherKey != null && !cipherKey.isBlank()) {
|
||||
// 未单独配置时从 ak 加密密钥派生（域分隔，避免与 AES 密钥混用同一材料）
|
||||
material = cipherKey.trim() + "|xq-sync-token";
|
||||
}
|
||||
if (material == null) {
|
||||
throw new IllegalStateException("未配置 app.sync.token-key（或 app.token-cipher-key），增量同步不可用");
|
||||
}
|
||||
return MessageDigest.getInstance("SHA-256").digest(material.getBytes(StandardCharsets.UTF_8));
|
||||
} catch (IllegalStateException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
throw new IllegalStateException("sync_token 密钥派生失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
private static String text(ObjectNode node, String field) {
|
||||
JsonNode value = node.get(field);
|
||||
return value == null || value.isNull() ? null : value.asText();
|
||||
}
|
||||
|
||||
private static long nowSec() {
|
||||
return System.currentTimeMillis() / 1000;
|
||||
}
|
||||
|
||||
/** sync_token 负载：fromTs 为本轮下界（不含，null=全量）；cutoffTs/snapshotId 仅 anchor 令牌有。 */
|
||||
public record SyncToken(String type, String entity, Long fromTs, Long cutoffTs,
|
||||
String snapshotId, long scopeVersion) {
|
||||
}
|
||||
|
||||
/** 页内游标负载。 */
|
||||
public record Cursor(String snapshotId, long modifyTs, long id) {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.mapper.m0.UserMapper;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* users 同步实体（m0.user）：人员主数据增量。
|
||||
* 字段口径（2026-09-14 字段核对）：status 对齐 soft_delete（在职状态不用 work_status，那是考勤）；
|
||||
* mobile 即唯一手机号字段（verified_mobile 无对应列，按决策砍掉）；organization_ids 取 dept_id；
|
||||
* 无人员合并机制，merged_into_user_id 省略。users 表无行级过滤先例，SYNC 工具以 scope 白名单为闸。
|
||||
*/
|
||||
@Component
|
||||
public class UsersSyncEntity extends AbstractSyncEntity implements SyncEntity {
|
||||
|
||||
public static final String NAME = "users";
|
||||
|
||||
final UserMapper userMapper;
|
||||
|
||||
public UsersSyncEntity(UserMapper userMapper) {
|
||||
this.userMapper = userMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<SyncRow> fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId,
|
||||
int limit, User user) {
|
||||
LambdaQueryWrapper<User> wrapper = new LambdaQueryWrapper<>();
|
||||
applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId);
|
||||
applyOrderAndLimit(wrapper, limit);
|
||||
List<SyncRow> rows = new ArrayList<>();
|
||||
for (User row : userMapper.selectList(wrapper)) {
|
||||
long mt = ChangesEngine.effectiveTs(row.getModifyTime(), row.getCreateTime());
|
||||
rows.add(new SyncRow(toItem(row, mt), mt, row.getId()));
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
private Map<String, Object> toItem(User row, long mt) {
|
||||
boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1;
|
||||
Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime();
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("user_id", row.getId());
|
||||
item.put("display_name", row.getName());
|
||||
item.put("status", deleted ? "inactive" : "active");
|
||||
item.put("employee_no", row.getNo());
|
||||
item.put("mobile", row.getMobile());
|
||||
item.put("organization_ids", row.getDeptId() == null ? List.of() : List.of(row.getDeptId()));
|
||||
item.put("updated_at", ChangesEngine.iso(effective));
|
||||
item.put("version", mt);
|
||||
item.put("deleted_at", deleted ? ChangesEngine.iso(effective) : null);
|
||||
return item;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package me.iiv.agent.mcp.tool;
|
||||
|
||||
import me.iiv.agent.mcp.ToolObject;
|
||||
import me.iiv.agent.mcp.XqTool;
|
||||
import me.iiv.agent.mcp.sync.ChangesEngine;
|
||||
import me.iiv.agent.mcp.sync.ContractsSyncEntity;
|
||||
import me.iiv.agent.mcp.sync.PropertiesSyncEntity;
|
||||
import me.iiv.agent.mcp.sync.UsersSyncEntity;
|
||||
import org.springframework.ai.tool.annotation.Tool;
|
||||
import org.springframework.ai.tool.annotation.ToolParam;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 增量同步工具（REQ-20260911-0013）：users/properties/contracts 三个 changes 工具，
|
||||
* 协议（snapshot/cursor/tombstone/sync_token）由 ChangesEngine 统一实现。
|
||||
* 数据域硬约束：本类不声明 tenant_id/scope 入参，数据域只能从 Token 推导；
|
||||
* GuardedToolCallback 对 *.changes.list 工具的原始入参做二次拦截，传 tenant_id/scope 一律拒绝。
|
||||
*/
|
||||
@Component
|
||||
public class ChangesTools implements ToolObject {
|
||||
|
||||
private static final String PROTOCOL_DESC = "增量同步协议：首轮不传 sync_token/cursor 发起全量拉取，"
|
||||
+ "响应钉住 snapshot_id 与 data_cutoff_at；页内用返回的 sync_token + next_cursor 继续翻页"
|
||||
+ "（不依赖页码，按 modify_time+id 复合游标推进，快照内结果一致、可断点续传）；"
|
||||
+ "has_more=false 的最后一页才签发 next_sync_token，下一轮增量同步以它作为 sync_token 入参。"
|
||||
+ "软删数据返回 tombstone（deleted_at 非空，status=inactive）；version 由 modify_time 派生（epoch 秒）。"
|
||||
+ "游标过期/篡改/跨快照返回 SYNC_CURSOR_EXPIRED，需重新全量同步。"
|
||||
+ "数据域只能从 Token 推导，禁止 tenant_id/scope 入参（传了会被拒绝）。";
|
||||
|
||||
private static final String SYNC_TOKEN_DESC = "同步令牌：首轮留空；轮内翻页传上一页原样回传的 sync_token；"
|
||||
+ "新一轮增量传上一轮最后一页签发的 next_sync_token";
|
||||
private static final String CURSOR_DESC = "页内游标：传上一页返回的 next_cursor；每轮第一页留空";
|
||||
private static final String LIMIT_DESC = "每页条数，默认 100，上限 200（超出截断为 200 并在 notice 提示）";
|
||||
|
||||
final ChangesEngine changesEngine;
|
||||
|
||||
public ChangesTools(ChangesEngine changesEngine) {
|
||||
this.changesEngine = changesEngine;
|
||||
}
|
||||
|
||||
@XqTool(category = "sync", kind = "read")
|
||||
@Tool(name = "users.changes.list", description = "增量同步人员主数据（m0.user）。" + PROTOCOL_DESC
|
||||
+ "items 字段：user_id/display_name/status（active=在职 inactive=删除，对齐软删标记）/"
|
||||
+ "employee_no/mobile/organization_ids（部门 id 列表）/updated_at（带时区）/version/deleted_at。")
|
||||
public Map<String, Object> usersChangesList(
|
||||
@ToolParam(required = false, description = SYNC_TOKEN_DESC) String syncToken,
|
||||
@ToolParam(required = false, description = CURSOR_DESC) String cursor,
|
||||
@ToolParam(required = false, description = LIMIT_DESC) Integer limit) {
|
||||
return changesEngine.execute(UsersSyncEntity.NAME, syncToken, cursor, limit);
|
||||
}
|
||||
|
||||
@XqTool(category = "sync", kind = "read")
|
||||
@Tool(name = "properties.changes.list", description = "增量同步物业（客户）主数据（m0.properyt）。" + PROTOCOL_DESC
|
||||
+ "items 字段：property_id/customer_id（金蝶客户 id）/property_name/customer_name（金蝶客户名称）/"
|
||||
+ "unified_social_credit_code/external_customer_no/status/updated_at/version/deleted_at。"
|
||||
+ "行级权限跟随 Token 绑定的人（与 property.list.query 同一 ACL）。")
|
||||
public Map<String, Object> propertiesChangesList(
|
||||
@ToolParam(required = false, description = SYNC_TOKEN_DESC) String syncToken,
|
||||
@ToolParam(required = false, description = CURSOR_DESC) String cursor,
|
||||
@ToolParam(required = false, description = LIMIT_DESC) Integer limit) {
|
||||
return changesEngine.execute(PropertiesSyncEntity.NAME, syncToken, cursor, limit);
|
||||
}
|
||||
|
||||
@XqTool(category = "sync", kind = "read")
|
||||
@Tool(name = "contracts.changes.list", description = "增量同步合同主数据（m2.contract_new）。" + PROTOCOL_DESC
|
||||
+ "items 字段：contract_id/contract_no/contract_name/status（0=待审 1=驳回 2=正常 3=过期 4=终止）/"
|
||||
+ "start_at/end_at（带时区）/amount_minor（金额，分）/customer_id/property_id/"
|
||||
+ "owner_user_id（负责人稳定 user.id）/expire_remind/updated_at/version/deleted_at。"
|
||||
+ "行级权限跟随 Token 绑定的人（与 contract.list.query 同一 ACL）。")
|
||||
public Map<String, Object> contractsChangesList(
|
||||
@ToolParam(required = false, description = SYNC_TOKEN_DESC) String syncToken,
|
||||
@ToolParam(required = false, description = CURSOR_DESC) String cursor,
|
||||
@ToolParam(required = false, description = LIMIT_DESC) Integer limit) {
|
||||
return changesEngine.execute(ContractsSyncEntity.NAME, syncToken, cursor, limit);
|
||||
}
|
||||
}
|
||||
@@ -2,17 +2,20 @@ package me.iiv.agent.mcp.tool;
|
||||
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import me.iiv.agent.entity.m0.Agent;
|
||||
import me.iiv.agent.entity.m0.Properyt;
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.entity.m2.ContractFile;
|
||||
import me.iiv.agent.entity.m2.ContractNew;
|
||||
import me.iiv.agent.entity.m2.ContractProcess;
|
||||
import me.iiv.agent.mapper.m0.AgentMapper;
|
||||
import me.iiv.agent.mapper.m0.ProperytMapper;
|
||||
import me.iiv.agent.mapper.m0.UserMapper;
|
||||
import me.iiv.agent.mapper.m2.ContractFileMapper;
|
||||
import me.iiv.agent.mapper.m2.ContractNewMapper;
|
||||
import me.iiv.agent.mapper.m2.ContractProcessMapper;
|
||||
import me.iiv.agent.mcp.ToolObject;
|
||||
import me.iiv.agent.mcp.XqTool;
|
||||
import me.iiv.agent.mcp.sync.ChangesEngine;
|
||||
import me.iiv.agent.permission.PermissionService;
|
||||
import me.iiv.agent.security.AuthContextHolder;
|
||||
import org.springframework.ai.tool.annotation.Tool;
|
||||
@@ -54,16 +57,19 @@ public class ContractTools implements ToolObject {
|
||||
final ContractProcessMapper contractProcessMapper;
|
||||
final AgentMapper agentMapper;
|
||||
final UserMapper userMapper;
|
||||
final ProperytMapper properytMapper;
|
||||
final PermissionService permissionService;
|
||||
|
||||
public ContractTools(ContractNewMapper contractNewMapper, ContractFileMapper contractFileMapper,
|
||||
ContractProcessMapper contractProcessMapper, AgentMapper agentMapper,
|
||||
UserMapper userMapper, PermissionService permissionService) {
|
||||
UserMapper userMapper, ProperytMapper properytMapper,
|
||||
PermissionService permissionService) {
|
||||
this.contractNewMapper = contractNewMapper;
|
||||
this.contractFileMapper = contractFileMapper;
|
||||
this.contractProcessMapper = contractProcessMapper;
|
||||
this.agentMapper = agentMapper;
|
||||
this.userMapper = userMapper;
|
||||
this.properytMapper = properytMapper;
|
||||
this.permissionService = permissionService;
|
||||
}
|
||||
|
||||
@@ -75,7 +81,9 @@ public class ContractTools implements ToolObject {
|
||||
+ "总部身份（isagent=0）可看全部。"
|
||||
+ "分页：page 从 1 开始，rows 默认 20、上限 100，超出自动截断为 100 并在返回的 notice 字段提示。"
|
||||
+ "contractStatus 含义：0=待审 1=驳回 2=正常 3=过期 4=终止；expireRemind 含义：0=关 1=开。"
|
||||
+ "返回 total（符合条件的总数）、records（本页记录，按 id 倒序，含 contractStatusText 状态文案）。")
|
||||
+ "返回 total（符合条件的总数）、records（本页记录，按 id 倒序，含 contractStatusText 状态文案；"
|
||||
+ "并含增量同步对齐字段 contractId/status/startAt/endAt/customerId/updatedAt/version，"
|
||||
+ "version 由 modify_time 派生 epoch 秒）。")
|
||||
public Map<String, Object> contractListQuery(
|
||||
@ToolParam(required = false, description = "合同编号，模糊匹配") String contractNo,
|
||||
@ToolParam(required = false, description = "合同名称，模糊匹配") String contractName,
|
||||
@@ -156,12 +164,13 @@ public class ContractTools implements ToolObject {
|
||||
? Map.of()
|
||||
: userMapper.selectBatchIds(creatorIds).stream()
|
||||
.collect(Collectors.toMap(User::getId, User::getName, (a, b) -> a));
|
||||
Map<Long, String> customerIds = customerIdsOf(contracts);
|
||||
|
||||
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
||||
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
|
||||
List<Map<String, Object>> records = new ArrayList<>();
|
||||
for (ContractNew contract : contracts) {
|
||||
records.add(toRecord(contract, agentNames, creatorNames, sdf, df));
|
||||
records.add(toRecord(contract, agentNames, creatorNames, customerIds, sdf, df));
|
||||
}
|
||||
|
||||
Map<String, Object> result = new LinkedHashMap<>();
|
||||
@@ -182,7 +191,8 @@ public class ContractTools implements ToolObject {
|
||||
@XqTool(category = "customer", kind = "read")
|
||||
@Tool(name = "contract.detail.query", description = "按 id 或 contractNo 查询合同详情（二选一必填），数据来源 m2 库 contract_new 表。"
|
||||
+ "行级权限与 contract.list.query 一致；合同不存在或当前身份无权查看时 record 为 null 并在 notice 字段说明（两者不区分）。"
|
||||
+ "返回 record（合同全字段，含终止原因和状态文案 contractStatusText）、"
|
||||
+ "返回 record（合同全字段，含终止原因和状态文案 contractStatusText，"
|
||||
+ "以及增量同步对齐字段 contractId/status/startAt/endAt/customerId/updatedAt/version）、"
|
||||
+ "files（合同附件列表，含 fileName/fileUrl）、"
|
||||
+ "processes（审批流程/待办/操作记录，按 id 正序；nodeType 含义：0=创建 1=部门内审核 2=层级审核 3=终止 4=抄送；"
|
||||
+ "operationType 含义：0=创建 1=审核通过 2=审核驳回 3=终止 4=修改再提交 5=抄送；"
|
||||
@@ -226,10 +236,11 @@ public class ContractTools implements ToolObject {
|
||||
User creator = userMapper.selectById(contract.getCreateId());
|
||||
createName = creator == null ? null : creator.getName();
|
||||
}
|
||||
Map<Long, String> customerIds = customerIdsOf(List.of(contract));
|
||||
result.put("record", toRecord(contract, agentNames,
|
||||
Map.of(contract.getCreateId() == null ? 0L : contract.getCreateId(),
|
||||
createName == null ? "未查询到" : createName),
|
||||
sdf, df));
|
||||
customerIds, sdf, df));
|
||||
|
||||
List<ContractFile> files = contractFileMapper.selectList(new LambdaQueryWrapper<ContractFile>()
|
||||
.eq(ContractFile::getSoftDelete, 0)
|
||||
@@ -273,11 +284,22 @@ public class ContractTools implements ToolObject {
|
||||
|
||||
private Map<String, Object> toRecord(ContractNew contract, Map<Long, String> agentNames,
|
||||
Map<Long, String> creatorNames,
|
||||
Map<Long, String> customerIds,
|
||||
SimpleDateFormat sdf, SimpleDateFormat df) {
|
||||
Map<String, Object> record = new LinkedHashMap<>();
|
||||
record.put("id", contract.getId());
|
||||
// REQ-20260911-0013 补齐字段：contractId/status/startAt/endAt/customerId/updatedAt/version
|
||||
//（version 无对应列，按决策 1 由 modify_time 派生 epoch 秒；startAt/endAt 带时区）
|
||||
record.put("contractId", contract.getId());
|
||||
record.put("contractNo", contract.getContractNo());
|
||||
record.put("contractName", contract.getContractName());
|
||||
record.put("status", contract.getContractStatus());
|
||||
record.put("startAt", ChangesEngine.iso(contract.getStartDate()));
|
||||
record.put("endAt", ChangesEngine.iso(contract.getEndDate()));
|
||||
record.put("customerId", contract.getPropertyId() == null
|
||||
? null : customerIds.get(contract.getPropertyId()));
|
||||
record.put("updatedAt", ChangesEngine.iso(contract.getModifyTime()));
|
||||
record.put("version", ChangesEngine.effectiveTs(contract.getModifyTime(), contract.getCreateTime()));
|
||||
record.put("contractAgentId", contract.getContractAgentId());
|
||||
record.put("contractAgentName", contract.getContractAgentId() == null
|
||||
? "未查询到" : agentNames.getOrDefault(contract.getContractAgentId(), "未查询到"));
|
||||
@@ -309,6 +331,21 @@ public class ContractTools implements ToolObject {
|
||||
return record;
|
||||
}
|
||||
|
||||
/** property_id → properyt.customerId（金蝶客户 id）批量富化（一次 IN 查询，消除 N+1）。 */
|
||||
private Map<Long, String> customerIdsOf(List<ContractNew> contracts) {
|
||||
List<Long> propertyIds = contracts.stream().map(ContractNew::getPropertyId)
|
||||
.filter(Objects::nonNull).distinct().collect(Collectors.toList());
|
||||
if (propertyIds.isEmpty()) {
|
||||
return Map.of();
|
||||
}
|
||||
// properyt.customerId 存在 NULL 行（2026-09-14 dev 冒烟命中），Collectors.toMap 遇 null 值抛 NPE
|
||||
Map<Long, String> map = new LinkedHashMap<>();
|
||||
for (Properyt property : properytMapper.selectBatchIds(propertyIds)) {
|
||||
map.put(property.getId(), property.getCustomerId());
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
private Map<Long, String> agentNamesOf(ContractNew contract) {
|
||||
List<Long> agentIds = Stream.of(contract.getContractAgentId(), contract.getCustomerAgentId())
|
||||
.filter(Objects::nonNull).distinct().collect(Collectors.toList());
|
||||
|
||||
@@ -20,6 +20,18 @@ app:
|
||||
# 签发 service 类型 Token 需在 app.open.clients 中配置绑定了作用域的密钥
|
||||
#（prod 用环境变量宽松绑定注入，如 APP_OPEN_CLIENTS_0_KEY / _0_NAME / _0_SCOPES_0，勿写明文）
|
||||
api-key: ${OPEN_API_KEY:}
|
||||
# 增量同步（REQ-20260911-0013）：sync_token/cursor 签名密钥（为空时从 token-cipher-key 派生）、
|
||||
# 快照有效期、单页上限与慢查询超时（秒）
|
||||
sync:
|
||||
token-key: ${SYNC_TOKEN_KEY:}
|
||||
cursor-ttl-hours: ${SYNC_CURSOR_TTL_HOURS:24}
|
||||
query-timeout-seconds: ${SYNC_QUERY_TIMEOUT_SECONDS:10}
|
||||
default-limit: ${SYNC_DEFAULT_LIMIT:100}
|
||||
max-limit: ${SYNC_MAX_LIMIT:200}
|
||||
# 工具调用限流（每 Token 每分钟）：SYNC 与 QUERY 分档互不占配额；0=关闭该档位
|
||||
rate-limit:
|
||||
query-per-minute: ${QUERY_RATE_LIMIT_PER_MINUTE:600}
|
||||
sync-per-minute: ${SYNC_RATE_LIMIT_PER_MINUTE:120}
|
||||
|
||||
spring:
|
||||
application:
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
-- tool.description 512 → 1024（REQ-20260911-0013）：changes 工具的协议说明较长，
|
||||
-- ToolSyncService 启动同步时插入超 512 字触发 Data truncation 导致应用启动失败
|
||||
ALTER TABLE tool
|
||||
MODIFY COLUMN description VARCHAR(1024) NULL COMMENT '工具描述';
|
||||
@@ -2,3 +2,4 @@ f8ec43a4fda81382d5675bc59fe7f4339f2e38a5b4241c260c3569e33b43a26d V1__init.sql
|
||||
0ad892601e891a18712b1b6a613c392c94e63492dc14f4f779375c092a3acab6 V2__p4_admin.sql
|
||||
0b6e407106156911f5fe83a13b342eeec5986e2e1e5b37b6797431a8720e1332 V3__token_cipher.sql
|
||||
5704691fa88406931c6aaf72bceaffd37cf6cc6982ce938c824ad8ecb94db1b5 V4__token_principal_type.sql
|
||||
17b270891eea9c831809de16b9c30a8dbdb01fb5c532f9e372e54513b92a65f9 V5__tool_description_length.sql
|
||||
|
||||
@@ -19,7 +19,7 @@ class FlywayMigrationTest {
|
||||
|
||||
flyway.migrate();
|
||||
|
||||
assertEquals("4", flyway.info().current().getVersion().getVersion());
|
||||
assertEquals("5", flyway.info().current().getVersion().getVersion());
|
||||
assertEquals(255, passwordHashLength(dataSource));
|
||||
}
|
||||
|
||||
@@ -30,12 +30,13 @@ class FlywayMigrationTest {
|
||||
statement.execute("CREATE TABLE admin_user (id BIGINT AUTO_INCREMENT PRIMARY KEY, password_hash CHAR(64) NOT NULL)");
|
||||
statement.execute("CREATE TABLE agent_api_key (id BIGINT AUTO_INCREMENT PRIMARY KEY,"
|
||||
+ " user_id BIGINT NOT NULL, key_hash CHAR(64) NOT NULL)");
|
||||
statement.execute("CREATE TABLE tool (id BIGINT AUTO_INCREMENT PRIMARY KEY, description VARCHAR(512))");
|
||||
}
|
||||
Flyway flyway = flyway(dataSource, true);
|
||||
|
||||
flyway.migrate();
|
||||
|
||||
assertEquals("4", flyway.info().current().getVersion().getVersion());
|
||||
assertEquals("5", flyway.info().current().getVersion().getVersion());
|
||||
assertEquals(255, passwordHashLength(dataSource));
|
||||
}
|
||||
|
||||
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
package me.iiv.agent.config;
|
||||
|
||||
import me.iiv.agent.mcp.sync.SyncQueryTimeoutHolder;
|
||||
import org.apache.ibatis.plugin.Invocation;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.sql.PreparedStatement;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
class SyncQueryTimeoutInterceptorTest {
|
||||
|
||||
private final SyncQueryTimeoutInterceptor interceptor = new SyncQueryTimeoutInterceptor();
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
SyncQueryTimeoutHolder.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void appliesQueryTimeoutWhenHolderSet() throws Throwable {
|
||||
PreparedStatement statement = mock(PreparedStatement.class);
|
||||
Invocation invocation = mock(Invocation.class);
|
||||
when(invocation.proceed()).thenReturn(statement);
|
||||
SyncQueryTimeoutHolder.set(10);
|
||||
|
||||
interceptor.intercept(invocation);
|
||||
|
||||
verify(statement).setQueryTimeout(10);
|
||||
}
|
||||
|
||||
@Test
|
||||
void leavesStatementAloneWhenHolderUnset() throws Throwable {
|
||||
PreparedStatement statement = mock(PreparedStatement.class);
|
||||
Invocation invocation = mock(Invocation.class);
|
||||
when(invocation.proceed()).thenReturn(statement);
|
||||
|
||||
interceptor.intercept(invocation);
|
||||
|
||||
verify(statement, never()).setQueryTimeout(org.mockito.ArgumentMatchers.anyInt());
|
||||
}
|
||||
}
|
||||
@@ -42,6 +42,8 @@ class GuardedToolCallbackTest {
|
||||
ToolGuard toolGuard;
|
||||
@Mock
|
||||
AuditService auditService;
|
||||
@Mock
|
||||
RateLimitService rateLimitService;
|
||||
|
||||
GuardedToolCallback callback;
|
||||
ToolDefinition toolDefinition;
|
||||
@@ -51,7 +53,7 @@ class GuardedToolCallbackTest {
|
||||
toolDefinition = DefaultToolDefinition.builder()
|
||||
.name("agent.count").description("d").inputSchema("{}").build();
|
||||
when(delegate.getToolDefinition()).thenReturn(toolDefinition);
|
||||
callback = new GuardedToolCallback(delegate, toolGuard, auditService);
|
||||
callback = new GuardedToolCallback(delegate, toolGuard, auditService, rateLimitService);
|
||||
AgentApiKey token = new AgentApiKey();
|
||||
token.setId(42L);
|
||||
AuthContextHolder.set(new AuthContext(token, null));
|
||||
@@ -153,4 +155,61 @@ class GuardedToolCallbackTest {
|
||||
assertEquals("req-fixed", errorOf(e).path("request_id").asText(),
|
||||
"过滤器生成的 request_id 必须贯穿到错误负载");
|
||||
}
|
||||
|
||||
@Test
|
||||
void rateLimitedMapsToRateLimitedCode() throws Exception {
|
||||
when(toolGuard.isAllowed(42L, "agent.count")).thenReturn(true);
|
||||
org.mockito.Mockito.doThrow(new ToolCallException(McpErrorCode.RATE_LIMITED, "触发限流"))
|
||||
.when(rateLimitService).check(42L, "agent.count");
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class, () -> callback.call("{}"));
|
||||
|
||||
assertEquals(McpErrorCode.RATE_LIMITED, e.getCode());
|
||||
assertEquals("RATE_LIMITED", errorOf(e).path("code").asText());
|
||||
}
|
||||
|
||||
@Test
|
||||
void syncToolRejectsTenantIdParam() throws Exception {
|
||||
toolDefinition = DefaultToolDefinition.builder()
|
||||
.name("users.changes.list").description("d").inputSchema("{}").build();
|
||||
when(delegate.getToolDefinition()).thenReturn(toolDefinition);
|
||||
callback = new GuardedToolCallback(delegate, toolGuard, auditService, rateLimitService);
|
||||
when(toolGuard.isAllowed(42L, "users.changes.list")).thenReturn(true);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> callback.call("{\"tenant_id\":\"foo\",\"limit\":10}"));
|
||||
|
||||
assertEquals(McpErrorCode.PERMISSION_DENIED, e.getCode());
|
||||
assertTrue(errorOf(e).path("message").asText().contains("tenant_id"));
|
||||
verify(delegate, org.mockito.Mockito.never()).call(anyString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void syncToolRejectsScopeParam() throws Exception {
|
||||
toolDefinition = DefaultToolDefinition.builder()
|
||||
.name("contracts.changes.list").description("d").inputSchema("{}").build();
|
||||
when(delegate.getToolDefinition()).thenReturn(toolDefinition);
|
||||
callback = new GuardedToolCallback(delegate, toolGuard, auditService, rateLimitService);
|
||||
when(toolGuard.isAllowed(42L, "contracts.changes.list")).thenReturn(true);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> callback.call("{\"scope\":\"operation\"}"));
|
||||
|
||||
assertEquals(McpErrorCode.PERMISSION_DENIED, e.getCode());
|
||||
verify(delegate, org.mockito.Mockito.never()).call(anyString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void syncToolWithoutForbiddenParamsPassesThrough() {
|
||||
toolDefinition = DefaultToolDefinition.builder()
|
||||
.name("properties.changes.list").description("d").inputSchema("{}").build();
|
||||
when(delegate.getToolDefinition()).thenReturn(toolDefinition);
|
||||
callback = new GuardedToolCallback(delegate, toolGuard, auditService, rateLimitService);
|
||||
when(toolGuard.isAllowed(42L, "properties.changes.list")).thenReturn(true);
|
||||
when(delegate.call("{\"limit\":10}")).thenReturn("{\"items\":[]}");
|
||||
|
||||
String result = callback.call("{\"limit\":10}");
|
||||
|
||||
assertEquals("{\"items\":[]}", result);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package me.iiv.agent.mcp;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.data.redis.RedisSystemException;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.core.ValueOperations;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.lenient;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class RateLimitServiceTest {
|
||||
|
||||
@Mock
|
||||
StringRedisTemplate redisTemplate;
|
||||
@Mock
|
||||
ValueOperations<String, String> valueOperations;
|
||||
|
||||
private RateLimitService newService(int queryPerMinute, int syncPerMinute) {
|
||||
lenient().when(redisTemplate.opsForValue()).thenReturn(valueOperations);
|
||||
return new RateLimitService(redisTemplate, queryPerMinute, syncPerMinute);
|
||||
}
|
||||
|
||||
@Test
|
||||
void withinLimitPassesAndSetsExpiryOnFirstHit() {
|
||||
RateLimitService service = newService(600, 120);
|
||||
when(valueOperations.increment(anyString())).thenReturn(1L);
|
||||
|
||||
assertDoesNotThrow(() -> service.check(42L, "agent.count"));
|
||||
|
||||
verify(redisTemplate).expire(anyString(), any(Duration.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void exceedingLimitThrowsRateLimited() {
|
||||
RateLimitService service = newService(600, 120);
|
||||
when(valueOperations.increment(anyString())).thenReturn(601L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> service.check(42L, "agent.count"));
|
||||
|
||||
assertEquals(McpErrorCode.RATE_LIMITED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void syncAndQueryTiersUseIndependentBuckets() {
|
||||
RateLimitService service = newService(600, 120);
|
||||
when(valueOperations.increment(anyString())).thenReturn(1L);
|
||||
|
||||
service.check(42L, "users.changes.list");
|
||||
service.check(42L, "contract.list.query");
|
||||
|
||||
ArgumentCaptor<String> keys = ArgumentCaptor.forClass(String.class);
|
||||
verify(valueOperations, org.mockito.Mockito.times(2)).increment(keys.capture());
|
||||
assertTrue(keys.getAllValues().get(0).contains(":sync:"), "changes 工具走 SYNC 档位");
|
||||
assertTrue(keys.getAllValues().get(1).contains(":query:"), "查询工具走 QUERY 档位");
|
||||
assertTrue(!keys.getAllValues().get(0).equals(keys.getAllValues().get(1)), "两档位互不占配额");
|
||||
}
|
||||
|
||||
@Test
|
||||
void syncTierHasOwnLimit() {
|
||||
RateLimitService service = newService(600, 120);
|
||||
when(valueOperations.increment(anyString())).thenReturn(121L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> service.check(42L, "contracts.changes.list"));
|
||||
|
||||
assertEquals(McpErrorCode.RATE_LIMITED, e.getCode(), "SYNC 第 121 次超限");
|
||||
}
|
||||
|
||||
@Test
|
||||
void redisFailureFailsOpen() {
|
||||
RateLimitService service = newService(600, 120);
|
||||
when(valueOperations.increment(anyString())).thenThrow(new RedisSystemException("down", null));
|
||||
|
||||
assertDoesNotThrow(() -> service.check(42L, "agent.count"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void zeroLimitDisablesTier() {
|
||||
RateLimitService service = new RateLimitService(redisTemplate, 0, 0);
|
||||
|
||||
assertDoesNotThrow(() -> service.check(42L, "users.changes.list"));
|
||||
|
||||
verify(redisTemplate, never()).opsForValue();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,212 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.entity.svc.AgentApiKey;
|
||||
import me.iiv.agent.mcp.IntegrationContextService;
|
||||
import me.iiv.agent.mcp.McpErrorCode;
|
||||
import me.iiv.agent.mcp.ToolCallException;
|
||||
import me.iiv.agent.security.AuthContext;
|
||||
import me.iiv.agent.security.AuthContextHolder;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ChangesEngineTest {
|
||||
|
||||
@Mock
|
||||
IntegrationContextService integrationContextService;
|
||||
|
||||
SyncTokenCodec codec = new SyncTokenCodec("engine-test-key", "", 24);
|
||||
FakeEntity entity = new FakeEntity();
|
||||
ChangesEngine engine;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
SyncEntityRegistry registry = new SyncEntityRegistry(List.of(entity));
|
||||
engine = new ChangesEngine(registry, codec, integrationContextService, 10, 100, 200);
|
||||
AgentApiKey token = new AgentApiKey();
|
||||
token.setId(1L);
|
||||
AuthContextHolder.set(new AuthContext(token, null));
|
||||
org.mockito.Mockito.lenient().when(integrationContextService.buildContext(any()))
|
||||
.thenReturn(Map.of("data_scope_version", 7L));
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
AuthContextHolder.clear();
|
||||
}
|
||||
|
||||
/** 可编程假实体：记录引擎传入的窗口/游标/条数，按队列返回行。 */
|
||||
static final class FakeEntity implements SyncEntity {
|
||||
final List<Long> fromSeen = new ArrayList<>();
|
||||
final List<Long> cutoffSeen = new ArrayList<>();
|
||||
final List<Long> cursorMtSeen = new ArrayList<>();
|
||||
final List<Long> cursorIdSeen = new ArrayList<>();
|
||||
final List<Integer> limitSeen = new ArrayList<>();
|
||||
List<SyncRow> nextRows = List.of();
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return "users";
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<SyncRow> fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId,
|
||||
int limit, User user) {
|
||||
fromSeen.add(fromTs);
|
||||
cutoffSeen.add(cutoffTs);
|
||||
cursorMtSeen.add(cursorMt);
|
||||
cursorIdSeen.add(cursorId);
|
||||
limitSeen.add(limit);
|
||||
return nextRows;
|
||||
}
|
||||
}
|
||||
|
||||
private static SyncRow row(long mt, long id) {
|
||||
Map<String, Object> item = new LinkedHashMap<>();
|
||||
item.put("user_id", id);
|
||||
item.put("version", mt);
|
||||
item.put("deleted_at", null);
|
||||
return new SyncRow(item, mt, id);
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void firstFullPullIssuesSnapshotAndNextSyncToken() {
|
||||
entity.nextRows = List.of(row(100, 1L), row(200, 2L));
|
||||
|
||||
Map<String, Object> result = engine.execute("users", null, null, null);
|
||||
|
||||
assertEquals("users", result.get("entity"));
|
||||
assertNotNull(result.get("snapshot_id"));
|
||||
assertNotNull(result.get("data_cutoff_at"));
|
||||
assertEquals(2, result.get("count"));
|
||||
assertEquals(false, result.get("has_more"));
|
||||
assertNotNull(result.get("sync_token"));
|
||||
assertNull(result.get("next_cursor"), "最后一页不再发页内游标");
|
||||
assertNotNull(result.get("next_sync_token"), "只有最后一页签发 next_sync_token");
|
||||
assertEquals(101, entity.limitSeen.get(0), "默认 limit=100，引擎多取一行判 has_more");
|
||||
assertNull(entity.fromSeen.get(0), "全量拉取无下界");
|
||||
}
|
||||
|
||||
@Test
|
||||
void intermediatePageKeepsSnapshotAndOnlyLastPageIssuesNextSyncToken() {
|
||||
// 第一页：limit=2，返回 3 行 → has_more
|
||||
entity.nextRows = List.of(row(100, 1L), row(200, 2L), row(300, 3L));
|
||||
Map<String, Object> page1 = engine.execute("users", null, null, 2);
|
||||
|
||||
assertEquals(true, page1.get("has_more"));
|
||||
assertEquals(2, page1.get("count"), "多取的一行不下发");
|
||||
assertNotNull(page1.get("next_cursor"));
|
||||
assertNull(page1.get("next_sync_token"), "中间页不发 next_sync_token");
|
||||
|
||||
// 第二页：带 sync_token + next_cursor → 快照钉住不变
|
||||
entity.nextRows = List.of(row(300, 3L));
|
||||
Map<String, Object> page2 = engine.execute("users",
|
||||
(String) page1.get("sync_token"), (String) page1.get("next_cursor"), 2);
|
||||
|
||||
assertEquals(page1.get("snapshot_id"), page2.get("snapshot_id"), "同一轮分页快照一致");
|
||||
assertEquals(page1.get("data_cutoff_at"), page2.get("data_cutoff_at"), "同一轮 data_cutoff_at 钉住");
|
||||
assertEquals(page1.get("sync_token"), page2.get("sync_token"), "轮内锚点令牌原样回传");
|
||||
assertEquals(200L, entity.cursorMtSeen.get(1), "游标推进到上一页最后一行");
|
||||
assertEquals(2L, entity.cursorIdSeen.get(1));
|
||||
assertEquals(false, page2.get("has_more"));
|
||||
assertNotNull(page2.get("next_sync_token"));
|
||||
|
||||
// 重复读同一页结果一致（重放 page2 入参）
|
||||
entity.nextRows = List.of(row(300, 3L));
|
||||
Map<String, Object> page2Replay = engine.execute("users",
|
||||
(String) page1.get("sync_token"), (String) page1.get("next_cursor"), 2);
|
||||
assertEquals(page2.get("items"), page2Replay.get("items"), "重复读同一页结果一致");
|
||||
}
|
||||
|
||||
@Test
|
||||
void nextSyncTokenStartsIncrementalRoundFromPreviousCutoff() {
|
||||
entity.nextRows = List.of(row(100, 1L));
|
||||
Map<String, Object> page1 = engine.execute("users", null, null, null);
|
||||
String nextToken = (String) page1.get("next_sync_token");
|
||||
long round1Cutoff = codec.parseToken((String) page1.get("sync_token"), "users").cutoffTs();
|
||||
|
||||
entity.nextRows = List.of();
|
||||
Map<String, Object> page2 = engine.execute("users", nextToken, null, null);
|
||||
|
||||
assertEquals(round1Cutoff, entity.fromSeen.get(1), "增量轮下界=上一轮 cutoff，不丢不重");
|
||||
assertTrue(!page2.get("snapshot_id").equals(page1.get("snapshot_id")), "新一轮换快照");
|
||||
assertEquals(0, page2.get("count"));
|
||||
assertNotNull(page2.get("next_sync_token"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void cursorFromOtherSnapshotIsRejected() {
|
||||
entity.nextRows = List.of(row(100, 1L));
|
||||
Map<String, Object> page1 = engine.execute("users", null, null, null);
|
||||
String foreignCursor = codec.encodeCursor("another-snapshot", 100L, 1L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class, () -> engine.execute("users",
|
||||
(String) page1.get("sync_token"), foreignCursor, null));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void expiredSyncTokenIsRejected() {
|
||||
SyncTokenCodec zeroTtl = new SyncTokenCodec("engine-test-key", "", -1);
|
||||
String expiredToken = zeroTtl.encodeHandoff("users", 100L, 7L);
|
||||
ChangesEngine expiredEngine = new ChangesEngine(new SyncEntityRegistry(List.of(entity)),
|
||||
zeroTtl, integrationContextService, 10, 100, 200);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> expiredEngine.execute("users", expiredToken, null, null));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void dataScopeVersionChangeInvalidatesToken() {
|
||||
String token = codec.encodeHandoff("users", 100L, 999L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> engine.execute("users", token, null, null));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void limitIsClampedToMaxWithNotice() {
|
||||
entity.nextRows = List.of();
|
||||
|
||||
Map<String, Object> result = engine.execute("users", null, null, 500);
|
||||
|
||||
assertEquals(200, result.get("limit"));
|
||||
assertEquals(201, entity.limitSeen.get(0));
|
||||
assertEquals("limit 超过上限 200，已截断为 200", result.get("notice"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void unknownEntityIsRejected() {
|
||||
SyncEntityRegistry registry = new SyncEntityRegistry(List.of(entity));
|
||||
ChangesEngine emptyEngine = new ChangesEngine(registry, codec, integrationContextService, 10, 100, 200);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> emptyEngine.execute("orders", null, null, null));
|
||||
|
||||
assertEquals(McpErrorCode.DATA_INTEGRITY_ERROR, e.getCode());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,215 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||
import me.iiv.agent.entity.m0.Properyt;
|
||||
import me.iiv.agent.entity.m0.User;
|
||||
import me.iiv.agent.entity.m2.ContractNew;
|
||||
import me.iiv.agent.mapper.m0.ProperytMapper;
|
||||
import me.iiv.agent.mapper.m0.UserMapper;
|
||||
import me.iiv.agent.mapper.m2.ContractNewMapper;
|
||||
import me.iiv.agent.permission.PermissionService;
|
||||
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.sql.Timestamp;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* 三个同步实体的窗口条件、tombstone 与字段映射测试（REQ-20260911-0013）。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class SyncEntitiesTest {
|
||||
|
||||
@Mock
|
||||
UserMapper userMapper;
|
||||
@Mock
|
||||
ProperytMapper properytMapper;
|
||||
@Mock
|
||||
ContractNewMapper contractNewMapper;
|
||||
@Mock
|
||||
PermissionService permissionService;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), User.class);
|
||||
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), Properyt.class);
|
||||
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), ContractNew.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void usersAppliesCoalesceWindowAndMapsFields() {
|
||||
UsersSyncEntity entity = new UsersSyncEntity(userMapper);
|
||||
User active = new User();
|
||||
active.setId(1L);
|
||||
active.setName("张三");
|
||||
active.setNo("E001");
|
||||
active.setMobile("13800000000");
|
||||
active.setDeptId(88L);
|
||||
active.setSoftDelete(0);
|
||||
active.setModifyTime(Timestamp.valueOf("2026-09-01 10:00:00"));
|
||||
User deleted = new User();
|
||||
deleted.setId(2L);
|
||||
deleted.setName("李四");
|
||||
deleted.setSoftDelete(1);
|
||||
deleted.setModifyTime(Timestamp.valueOf("2026-09-02 10:00:00"));
|
||||
when(userMapper.selectList(any())).thenReturn(List.of(active, deleted));
|
||||
|
||||
List<SyncRow> rows = entity.fetchPage(100L, 2000L, 150L, 5L, 3, null);
|
||||
|
||||
ArgumentCaptor<LambdaQueryWrapper<User>> captor = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
|
||||
verify(userMapper).selectList(captor.capture());
|
||||
String sql = captor.getValue().getSqlSegment();
|
||||
assertTrue(sql.contains("COALESCE(modify_time, create_time)"), "窗口必须按生效修改时间过滤: " + sql);
|
||||
|
||||
assertEquals(2, rows.size());
|
||||
assertEquals(1L, rows.get(0).item().get("user_id"));
|
||||
assertEquals("张三", rows.get(0).item().get("display_name"));
|
||||
assertEquals("active", rows.get(0).item().get("status"));
|
||||
assertEquals("E001", rows.get(0).item().get("employee_no"));
|
||||
assertEquals("13800000000", rows.get(0).item().get("mobile"));
|
||||
assertEquals(List.of(88L), rows.get(0).item().get("organization_ids"));
|
||||
assertEquals("2026-09-01T10:00:00+08:00", rows.get(0).item().get("updated_at"));
|
||||
assertEquals(Timestamp.valueOf("2026-09-01 10:00:00").getTime() / 1000,
|
||||
rows.get(0).item().get("version"));
|
||||
assertNull(rows.get(0).item().get("deleted_at"));
|
||||
|
||||
// tombstone：软删行 deleted_at 非空、status=inactive
|
||||
assertEquals("inactive", rows.get(1).item().get("status"));
|
||||
assertEquals("2026-09-02T10:00:00+08:00", rows.get(1).item().get("deleted_at"));
|
||||
assertEquals(List.of(), rows.get(1).item().get("organization_ids"), "无部门时为空列表");
|
||||
}
|
||||
|
||||
@Test
|
||||
void usersFallsBackToCreateTimeWhenModifyTimeNull() {
|
||||
UsersSyncEntity entity = new UsersSyncEntity(userMapper);
|
||||
User legacy = new User();
|
||||
legacy.setId(3L);
|
||||
legacy.setSoftDelete(0);
|
||||
legacy.setModifyTime(null);
|
||||
legacy.setCreateTime(Timestamp.valueOf("2026-01-01 08:00:00"));
|
||||
when(userMapper.selectList(any())).thenReturn(List.of(legacy));
|
||||
|
||||
List<SyncRow> rows = entity.fetchPage(null, 2000L, null, null, 3, null);
|
||||
|
||||
assertEquals(Timestamp.valueOf("2026-01-01 08:00:00").getTime() / 1000, rows.get(0).modifyTs(),
|
||||
"modify_time 为 NULL 的存量行按 create_time 参与游标，不能丢");
|
||||
}
|
||||
|
||||
@Test
|
||||
void propertiesAppliesRowFilterForUserAndMapsFields() {
|
||||
PropertiesSyncEntity entity = new PropertiesSyncEntity(properytMapper, permissionService);
|
||||
Properyt property = new Properyt();
|
||||
property.setId(9L);
|
||||
property.setName("望京SOHO");
|
||||
property.setCustomerId("KD10086");
|
||||
property.setJinideName("金蝶客户名称");
|
||||
property.setIdentificationNumber("91110105XXXXXXXX");
|
||||
property.setSerialnumber("452542");
|
||||
property.setSoftDelete(0);
|
||||
property.setModifyTime(Timestamp.valueOf("2026-09-01 10:00:00"));
|
||||
when(properytMapper.selectList(any())).thenReturn(List.of(property));
|
||||
User user = new User();
|
||||
user.setId(7L);
|
||||
|
||||
List<SyncRow> rows = entity.fetchPage(null, 2000L, null, null, 3, user);
|
||||
|
||||
verify(permissionService).applyPropertyRowFilter(any(), any());
|
||||
assertEquals(9L, rows.get(0).item().get("property_id"));
|
||||
assertEquals("KD10086", rows.get(0).item().get("customer_id"));
|
||||
assertEquals("望京SOHO", rows.get(0).item().get("property_name"));
|
||||
assertEquals("金蝶客户名称", rows.get(0).item().get("customer_name"));
|
||||
assertEquals("91110105XXXXXXXX", rows.get(0).item().get("unified_social_credit_code"));
|
||||
assertEquals("452542", rows.get(0).item().get("external_customer_no"));
|
||||
assertEquals("active", rows.get(0).item().get("status"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void propertiesSkipsRowFilterForServiceToken() {
|
||||
PropertiesSyncEntity entity = new PropertiesSyncEntity(properytMapper, permissionService);
|
||||
when(properytMapper.selectList(any())).thenReturn(List.of());
|
||||
|
||||
entity.fetchPage(null, 2000L, null, null, 3, null);
|
||||
|
||||
verify(permissionService, never()).applyPropertyRowFilter(any(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void contractsToleratesNullCustomerIdOnProperty() {
|
||||
// 回归：properyt.customerId 存在 NULL 行（2026-09-14 dev 冒烟命中 NPE），富化不得炸
|
||||
ContractsSyncEntity entity = new ContractsSyncEntity(contractNewMapper, properytMapper, permissionService);
|
||||
ContractNew contract = new ContractNew();
|
||||
contract.setId(10L);
|
||||
contract.setPropertyId(2L);
|
||||
contract.setSoftDelete(0);
|
||||
contract.setModifyTime(Timestamp.valueOf("2026-09-01 13:54:21"));
|
||||
when(contractNewMapper.selectList(any())).thenReturn(List.of(contract));
|
||||
Properyt property = new Properyt();
|
||||
property.setId(2L);
|
||||
property.setCustomerId(null);
|
||||
when(properytMapper.selectBatchIds(anyList())).thenReturn(List.of(property));
|
||||
|
||||
List<SyncRow> rows = entity.fetchPage(null, 2000L, null, null, 3, null);
|
||||
|
||||
assertEquals(1, rows.size());
|
||||
assertNull(rows.get(0).item().get("customer_id"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void contractsMapsOwnerAmountAndTombstone() {
|
||||
ContractsSyncEntity entity = new ContractsSyncEntity(contractNewMapper, properytMapper, permissionService);
|
||||
ContractNew contract = new ContractNew();
|
||||
contract.setId(9L);
|
||||
contract.setContractNo("452542C001");
|
||||
contract.setContractName("望京SOHO厕纸服务合同");
|
||||
contract.setContractStatus(2);
|
||||
contract.setPropertyId(1L);
|
||||
contract.setSaleId(1001L);
|
||||
contract.setExpireRemind(1);
|
||||
contract.setAmount(new BigDecimal("120000.00"));
|
||||
contract.setStartDate(java.sql.Date.valueOf("2026-01-01"));
|
||||
contract.setEndDate(java.sql.Date.valueOf("2026-12-31"));
|
||||
contract.setSoftDelete(1);
|
||||
contract.setModifyTime(Timestamp.valueOf("2026-09-01 13:54:21"));
|
||||
when(contractNewMapper.selectList(any())).thenReturn(List.of(contract));
|
||||
Properyt property = new Properyt();
|
||||
property.setId(1L);
|
||||
property.setCustomerId("KD10086");
|
||||
when(properytMapper.selectBatchIds(anyList())).thenReturn(List.of(property));
|
||||
User user = new User();
|
||||
user.setId(7L);
|
||||
|
||||
List<SyncRow> rows = entity.fetchPage(null, 2000L, null, null, 3, user);
|
||||
|
||||
verify(permissionService).applyContractRowFilter(any(), any());
|
||||
assertEquals(9L, rows.get(0).item().get("contract_id"));
|
||||
assertEquals("452542C001", rows.get(0).item().get("contract_no"));
|
||||
assertEquals(2, rows.get(0).item().get("status"));
|
||||
assertEquals("2026-01-01T00:00:00+08:00", rows.get(0).item().get("start_at"));
|
||||
assertEquals("2026-12-31T00:00:00+08:00", rows.get(0).item().get("end_at"));
|
||||
assertEquals(12000000L, rows.get(0).item().get("amount_minor"), "金额按分输出");
|
||||
assertEquals("KD10086", rows.get(0).item().get("customer_id"));
|
||||
assertEquals(1L, rows.get(0).item().get("property_id"));
|
||||
assertEquals(1001L, rows.get(0).item().get("owner_user_id"), "负责人=稳定 user.id");
|
||||
assertEquals(1, rows.get(0).item().get("expire_remind"));
|
||||
assertEquals(Timestamp.valueOf("2026-09-01 13:54:21").getTime() / 1000,
|
||||
rows.get(0).item().get("version"));
|
||||
assertEquals("2026-09-01T13:54:21+08:00", rows.get(0).item().get("deleted_at"), "软删行返回 tombstone");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package me.iiv.agent.mcp.sync;
|
||||
|
||||
import me.iiv.agent.mcp.McpErrorCode;
|
||||
import me.iiv.agent.mcp.ToolCallException;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
class SyncTokenCodecTest {
|
||||
|
||||
private final SyncTokenCodec codec = new SyncTokenCodec("test-sync-key", "", 24);
|
||||
|
||||
@Test
|
||||
void roundTokenRoundTrips() {
|
||||
String token = codec.encodeRound("users", 1000L, 2000L, "snap-1", 7L);
|
||||
|
||||
SyncTokenCodec.SyncToken parsed = codec.parseToken(token, "users");
|
||||
|
||||
assertEquals(SyncTokenCodec.TYPE_ROUND, parsed.type());
|
||||
assertEquals("users", parsed.entity());
|
||||
assertEquals(1000L, parsed.fromTs());
|
||||
assertEquals(2000L, parsed.cutoffTs());
|
||||
assertEquals("snap-1", parsed.snapshotId());
|
||||
assertEquals(7L, parsed.scopeVersion());
|
||||
}
|
||||
|
||||
@Test
|
||||
void roundTokenWithNullFromMeansFullPull() {
|
||||
String token = codec.encodeRound("contracts", null, 2000L, "snap-2", 0L);
|
||||
|
||||
SyncTokenCodec.SyncToken parsed = codec.parseToken(token, "contracts");
|
||||
|
||||
assertNull(parsed.fromTs());
|
||||
}
|
||||
|
||||
@Test
|
||||
void handoffTokenCarriesOnlyFrom() {
|
||||
String token = codec.encodeHandoff("properties", 3000L, 9L);
|
||||
|
||||
SyncTokenCodec.SyncToken parsed = codec.parseToken(token, "properties");
|
||||
|
||||
assertEquals(SyncTokenCodec.TYPE_HANDOFF, parsed.type());
|
||||
assertEquals(3000L, parsed.fromTs());
|
||||
assertNull(parsed.cutoffTs());
|
||||
assertNull(parsed.snapshotId());
|
||||
assertEquals(9L, parsed.scopeVersion());
|
||||
}
|
||||
|
||||
@Test
|
||||
void cursorRoundTrips() {
|
||||
String cursor = codec.encodeCursor("snap-9", 1757L, 42L);
|
||||
|
||||
SyncTokenCodec.Cursor parsed = codec.parseCursor(cursor);
|
||||
|
||||
assertEquals("snap-9", parsed.snapshotId());
|
||||
assertEquals(1757L, parsed.modifyTs());
|
||||
assertEquals(42L, parsed.id());
|
||||
}
|
||||
|
||||
@Test
|
||||
void tamperedTokenIsRejectedAsExpired() {
|
||||
String token = codec.encodeRound("users", null, 2000L, "snap-1", 0L);
|
||||
String tampered = token.substring(0, token.length() - 2)
|
||||
+ (token.endsWith("AA") ? "BB" : "AA");
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> codec.parseToken(tampered, "users"));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void entityMismatchIsRejected() {
|
||||
String token = codec.encodeRound("users", null, 2000L, "snap-1", 0L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> codec.parseToken(token, "contracts"));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void expiredTokenIsRejected() {
|
||||
SyncTokenCodec zeroTtl = new SyncTokenCodec("test-sync-key", "", -1);
|
||||
String token = zeroTtl.encodeHandoff("users", 100L, 0L);
|
||||
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> zeroTtl.parseToken(token, "users"));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void garbageInputIsRejected() {
|
||||
ToolCallException e = assertThrows(ToolCallException.class,
|
||||
() -> codec.parseToken("not-a-token", "users"));
|
||||
|
||||
assertEquals(McpErrorCode.SYNC_CURSOR_EXPIRED, e.getCode());
|
||||
}
|
||||
|
||||
@Test
|
||||
void keyFallsBackToCipherKeyDerivation() {
|
||||
SyncTokenCodec derived = new SyncTokenCodec("", "cipher-key", 24);
|
||||
SyncTokenCodec explicit = new SyncTokenCodec("", "cipher-key", 24);
|
||||
|
||||
String token = derived.encodeHandoff("users", 1L, 0L);
|
||||
|
||||
// 相同材料派生的密钥可互相校验
|
||||
explicit.parseToken(token, "users");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package me.iiv.agent.mcp.tool;
|
||||
|
||||
import me.iiv.agent.mcp.sync.ChangesEngine;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.ai.tool.method.MethodToolCallbackProvider;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
* 三个 changes 工具的注册与路由（REQ-20260911-0013）。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ChangesToolsTest {
|
||||
|
||||
@Mock
|
||||
ChangesEngine changesEngine;
|
||||
|
||||
@Test
|
||||
void springAiPublishesThreeChangesToolSchemas() {
|
||||
ChangesTools tools = new ChangesTools(changesEngine);
|
||||
var callbacks = MethodToolCallbackProvider.builder().toolObjects(tools).build().getToolCallbacks();
|
||||
Map<String, String> schemas = Arrays.stream(callbacks).collect(Collectors.toMap(
|
||||
callback -> callback.getToolDefinition().name(),
|
||||
callback -> callback.getToolDefinition().inputSchema()));
|
||||
|
||||
assertEquals(3, schemas.size());
|
||||
for (String name : new String[]{"users.changes.list", "properties.changes.list", "contracts.changes.list"}) {
|
||||
assertTrue(schemas.containsKey(name), "缺少工具 " + name);
|
||||
String schema = schemas.get(name);
|
||||
assertTrue(schema.contains("syncToken"));
|
||||
assertTrue(schema.contains("cursor"));
|
||||
assertTrue(schema.contains("limit"));
|
||||
assertFalse(schema.contains("tenant_id"), "SYNC 工具禁止 tenant_id 入参: " + name);
|
||||
assertFalse(schema.contains("\"scope\""), "SYNC 工具禁止 scope 入参: " + name);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void toolsRouteToEngineWithEntityName() {
|
||||
ChangesTools tools = new ChangesTools(changesEngine);
|
||||
|
||||
tools.usersChangesList("t1", "c1", 50);
|
||||
verify(changesEngine).execute("users", "t1", "c1", 50);
|
||||
|
||||
tools.propertiesChangesList(null, null, null);
|
||||
verify(changesEngine).execute("properties", null, null, null);
|
||||
|
||||
tools.contractsChangesList("t2", null, 200);
|
||||
verify(changesEngine).execute("contracts", "t2", null, 200);
|
||||
}
|
||||
}
|
||||
@@ -54,6 +54,8 @@ class ContractToolsTest {
|
||||
@Mock
|
||||
UserMapper userMapper;
|
||||
@Mock
|
||||
me.iiv.agent.mapper.m0.ProperytMapper properytMapper;
|
||||
@Mock
|
||||
PermissionService permissionService;
|
||||
|
||||
ContractTools tools;
|
||||
@@ -68,7 +70,7 @@ class ContractToolsTest {
|
||||
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""),
|
||||
ContractProcess.class);
|
||||
tools = new ContractTools(contractNewMapper, contractFileMapper, contractProcessMapper,
|
||||
agentMapper, userMapper, permissionService);
|
||||
agentMapper, userMapper, properytMapper, permissionService);
|
||||
user = new User();
|
||||
user.setId(7L);
|
||||
AuthContextHolder.set(new AuthContext(null, user));
|
||||
@@ -96,6 +98,10 @@ class ContractToolsTest {
|
||||
creator.setId(1000L);
|
||||
creator.setName("张三");
|
||||
when(userMapper.selectBatchIds(anyList())).thenReturn(List.of(creator));
|
||||
me.iiv.agent.entity.m0.Properyt property = new me.iiv.agent.entity.m0.Properyt();
|
||||
property.setId(1L);
|
||||
property.setCustomerId("KD10086");
|
||||
when(properytMapper.selectBatchIds(anyList())).thenReturn(List.of(property));
|
||||
|
||||
Map<String, Object> result = tools.contractListQuery(
|
||||
"452542C", "服务合同", "望京", "金蝶客户", "452542",
|
||||
@@ -134,6 +140,17 @@ class ContractToolsTest {
|
||||
assertEquals(new BigDecimal("120000.00"), record.get("amount"));
|
||||
assertEquals("2026-01-01", record.get("startDate"));
|
||||
assertEquals("2026-12-31", record.get("endDate"));
|
||||
// REQ-20260911-0013 补齐字段
|
||||
assertEquals(9L, record.get("contractId"));
|
||||
assertEquals(2, record.get("status"));
|
||||
assertEquals("2026-01-01T00:00:00+08:00", record.get("startAt"));
|
||||
assertEquals("2026-12-31T00:00:00+08:00", record.get("endAt"));
|
||||
assertEquals("KD10086", record.get("customerId"));
|
||||
assertEquals(1001L, record.get("saleId"));
|
||||
assertEquals("销售甲", record.get("saleName"));
|
||||
assertEquals(1L, record.get("propertyId"));
|
||||
assertEquals("2026-08-15T10:20:30+08:00", record.get("updatedAt"));
|
||||
assertEquals(java.sql.Timestamp.valueOf("2026-08-15 10:20:30").getTime() / 1000, record.get("version"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -169,6 +186,10 @@ class ContractToolsTest {
|
||||
creator.setId(1000L);
|
||||
creator.setName("张三");
|
||||
when(userMapper.selectById(1000L)).thenReturn(creator);
|
||||
me.iiv.agent.entity.m0.Properyt property = new me.iiv.agent.entity.m0.Properyt();
|
||||
property.setId(1L);
|
||||
property.setCustomerId("KD10086");
|
||||
when(properytMapper.selectBatchIds(anyList())).thenReturn(List.of(property));
|
||||
|
||||
ContractFile file = new ContractFile();
|
||||
file.setId(1L);
|
||||
@@ -202,6 +223,9 @@ class ContractToolsTest {
|
||||
assertEquals("未查询到", record.get("customerAgentName"));
|
||||
assertEquals("张三", record.get("createName"));
|
||||
assertEquals("正常", record.get("contractStatusText"));
|
||||
assertEquals(9L, record.get("contractId"));
|
||||
assertEquals("KD10086", record.get("customerId"));
|
||||
assertEquals(java.sql.Timestamp.valueOf("2026-08-15 10:20:30").getTime() / 1000, record.get("version"));
|
||||
|
||||
List<Map<String, Object>> files = (List<Map<String, Object>>) result.get("files");
|
||||
assertEquals(1, files.size());
|
||||
@@ -253,6 +277,7 @@ class ContractToolsTest {
|
||||
contract.setSaleName("销售甲");
|
||||
contract.setContractStatus(2);
|
||||
contract.setCreateId(1000L);
|
||||
contract.setModifyTime(java.sql.Timestamp.valueOf("2026-08-15 10:20:30"));
|
||||
return contract;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user