From 6639b93c45d2d54474ef4b2bb2db69aed4227703 Mon Sep 17 00:00:00 2001 From: wangheng Date: Mon, 14 Sep 2026 12:19:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20=E5=A2=9E=E9=87=8F=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=A1=86=E6=9E=B6=E4=B8=8E=20users/properties/contrac?= =?UTF-8?q?ts=20changes=20=E5=B7=A5=E5=85=B7=20(REQ-20260911-0013)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 通用 ChangesEngine：snapshot 钉住 data_cutoff_at、(modify_time,id) 复合游标、 仅最后一页签发 next_sync_token、软删 tombstone；sync_token/cursor 为 HMAC-SHA256 无状态令牌（SyncTokenCodec），过期/篡改/跨快照/数据域变更 统一 SYNC_CURSOR_EXPIRED - 注册式实体扩展：SyncEntity/Registry + users(m0.user)/properties(m0.properyt)/ contracts(m2.contract_new) 三个实现；生效修改时间 COALESCE(modify_time, create_time) - 硬约束：GuardedToolCallback 对 *.changes.list 拒绝 tenant_id/scope 入参， 数据域只从 Token 推导；行级权限复用 PermissionService 与 QUERY 工具同一 ACL - 资源隔离：RateLimitService SYNC/QUERY 分档限流（Redis 固定分钟窗）、 limit≤200 截断、SyncQueryTimeoutInterceptor 慢查询超时（三套 SqlSessionFactory 接入） - contract.list.query 补齐 12 字段、contract.detail.query 补 version（ACL 不变） - V5 migration：tool.description 512→1024（含 sql/rollback 回滚文件） - seed-dev.sql：三个 changes 工具绑定进 sync 作用域 - 修复：properyt.customerId 为 NULL 时 Collectors.toMap NPE（冒烟命中）， 改为 put 逐条装填（ContractsSyncEntity/ContractTools） --- .../V5__tool_description_length.down.sql | 6 + xiaoqu-agent-service/sql/seed-dev.sql | 9 +- .../iiv/agent/config/M0DataSourceConfig.java | 5 +- .../iiv/agent/config/M1DataSourceConfig.java | 5 +- .../iiv/agent/config/M2DataSourceConfig.java | 5 +- .../me/iiv/agent/config/McpToolConfig.java | 6 +- .../config/SyncQueryTimeoutInterceptor.java | 33 +++ .../me/iiv/agent/mcp/GuardedToolCallback.java | 42 +++- .../mcp/GuardedToolCallbackProvider.java | 7 +- .../me/iiv/agent/mcp/RateLimitService.java | 63 +++++ .../agent/mcp/sync/AbstractSyncEntity.java | 33 +++ .../me/iiv/agent/mcp/sync/ChangesEngine.java | 180 ++++++++++++++ .../agent/mcp/sync/ContractsSyncEntity.java | 114 +++++++++ .../agent/mcp/sync/PropertiesSyncEntity.java | 75 ++++++ .../me/iiv/agent/mcp/sync/SyncEntity.java | 26 +++ .../agent/mcp/sync/SyncEntityRegistry.java | 28 +++ .../mcp/sync/SyncQueryTimeoutHolder.java | 26 +++ .../java/me/iiv/agent/mcp/sync/SyncRow.java | 9 + .../me/iiv/agent/mcp/sync/SyncTokenCodec.java | 219 ++++++++++++++++++ .../iiv/agent/mcp/sync/UsersSyncEntity.java | 65 ++++++ .../me/iiv/agent/mcp/tool/ChangesTools.java | 78 +++++++ .../me/iiv/agent/mcp/tool/ContractTools.java | 47 +++- .../src/main/resources/application.yml | 12 + .../migration/V5__tool_description_length.sql | 4 + .../resources/db/migration/checksums.sha256 | 1 + .../iiv/agent/config/FlywayMigrationTest.java | 5 +- .../SyncQueryTimeoutInterceptorTest.java | 46 ++++ .../agent/mcp/GuardedToolCallbackTest.java | 61 ++++- .../iiv/agent/mcp/RateLimitServiceTest.java | 101 ++++++++ .../iiv/agent/mcp/sync/ChangesEngineTest.java | 212 +++++++++++++++++ .../iiv/agent/mcp/sync/SyncEntitiesTest.java | 215 +++++++++++++++++ .../agent/mcp/sync/SyncTokenCodecTest.java | 113 +++++++++ .../iiv/agent/mcp/tool/ChangesToolsTest.java | 61 +++++ .../iiv/agent/mcp/tool/ContractToolsTest.java | 27 ++- 34 files changed, 1917 insertions(+), 22 deletions(-) create mode 100644 xiaoqu-agent-service/sql/rollback/V5__tool_description_length.down.sql create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/config/SyncQueryTimeoutInterceptor.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/RateLimitService.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/AbstractSyncEntity.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ChangesEngine.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ContractsSyncEntity.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/PropertiesSyncEntity.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntity.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntityRegistry.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncQueryTimeoutHolder.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncRow.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncTokenCodec.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/UsersSyncEntity.java create mode 100644 xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ChangesTools.java create mode 100644 xiaoqu-agent-service/src/main/resources/db/migration/V5__tool_description_length.sql create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/config/SyncQueryTimeoutInterceptorTest.java create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/RateLimitServiceTest.java create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/ChangesEngineTest.java create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncEntitiesTest.java create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncTokenCodecTest.java create mode 100644 xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ChangesToolsTest.java diff --git a/xiaoqu-agent-service/sql/rollback/V5__tool_description_length.down.sql b/xiaoqu-agent-service/sql/rollback/V5__tool_description_length.down.sql new file mode 100644 index 0000000..e7ace3f --- /dev/null +++ b/xiaoqu-agent-service/sql/rollback/V5__tool_description_length.down.sql @@ -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 '工具描述'; diff --git a/xiaoqu-agent-service/sql/seed-dev.sql b/xiaoqu-agent-service/sql/seed-dev.sql index a0a3e42..4585a79 100644 --- a/xiaoqu-agent-service/sql/seed-dev.sql +++ b/xiaoqu-agent-service/sql/seed-dev.sql @@ -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; diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M0DataSourceConfig.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M0DataSourceConfig.java index 438845c..d885e7e 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M0DataSourceConfig.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M0DataSourceConfig.java @@ -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(); } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M1DataSourceConfig.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M1DataSourceConfig.java index 8acdc5a..c9561b1 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M1DataSourceConfig.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M1DataSourceConfig.java @@ -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(); } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M2DataSourceConfig.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M2DataSourceConfig.java index caba3e1..c6ebe1e 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M2DataSourceConfig.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/M2DataSourceConfig.java @@ -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(); } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/McpToolConfig.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/McpToolConfig.java index c51f083..4a66ba2 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/McpToolConfig.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/McpToolConfig.java @@ -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 toolObjects, ToolGuard toolGuard, AuditService auditService) { + public ToolCallbackProvider xiaoquTools(List 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); } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/SyncQueryTimeoutInterceptor.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/SyncQueryTimeoutInterceptor.java new file mode 100644 index 0000000..0c502ed --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/config/SyncQueryTimeoutInterceptor.java @@ -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; + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallback.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallback.java index 4c3d55c..9496075 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallback.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallback.java @@ -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 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），先拆包取根因再分类。 diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallbackProvider.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallbackProvider.java index d5d271f..7e6d360 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallbackProvider.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/GuardedToolCallbackProvider.java @@ -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); } } diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/RateLimitService.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/RateLimitService.java new file mode 100644 index 0000000..3773aff --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/RateLimitService.java @@ -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()); + } + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/AbstractSyncEntity.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/AbstractSyncEntity.java new file mode 100644 index 0000000..7fab7d5 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/AbstractSyncEntity.java @@ -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)"; + + void applyWindow(LambdaQueryWrapper 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 直拼，无注入面）。 */ + void applyOrderAndLimit(LambdaQueryWrapper wrapper, int limit) { + wrapper.last("ORDER BY " + EFFECTIVE_TIME + " ASC, id ASC LIMIT " + limit); + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ChangesEngine.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ChangesEngine.java new file mode 100644 index 0000000..09862f1 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ChangesEngine.java @@ -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，协议只实现一次）： + *
    + *
  • 每轮同步签发 snapshot_id 钉住 data_cutoff_at，页间按 (modify_time, id) 复合游标推进， + * 不依赖页码；
    • + *
  • 中间页原样回传 sync_token + next_cursor，只有最后一页（has_more=false）签发 + * next_sync_token（from=本轮 cutoff）；
    • + *
  • 软删行由实体输出 tombstone（deleted_at 非空），游标过期/篡改/跨快照/数据域变更 + * 统一 SYNC_CURSOR_EXPIRED；
    • + *
  • 数据域只能从 Token 推导：本引擎无任何 tenant_id/scope 入参，行级权限跟随 + * Token 绑定的人（服务 Token 无人员绑定，不加行过滤）。
    • + *
+ */ +@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 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 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 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> 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; + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ContractsSyncEntity.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ContractsSyncEntity.java new file mode 100644 index 0000000..597934e --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/ContractsSyncEntity.java @@ -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 fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId, + int limit, User user) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + if (user != null) { + permissionService.applyContractRowFilter(wrapper, user); + } + applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId); + applyOrderAndLimit(wrapper, limit); + List contracts = contractNewMapper.selectList(wrapper); + + // customer_id 批量富化（property_id → properyt.customerId，一次 IN 查询，消除 N+1） + List propertyIds = contracts.stream().map(ContractNew::getPropertyId) + .filter(Objects::nonNull).distinct().collect(Collectors.toList()); + Map customerIds = propertyIds.isEmpty() + ? Map.of() + : customerIdsByProperty(propertyIds); + + List 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 toItem(ContractNew row, long mt, Map customerIds) { + boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1; + Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime(); + Map 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 customerIdsByProperty(List propertyIds) { + Map map = new LinkedHashMap<>(); + for (Properyt property : properytMapper.selectBatchIds(propertyIds)) { + map.put(property.getId(), property.getCustomerId()); + } + return map; + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/PropertiesSyncEntity.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/PropertiesSyncEntity.java new file mode 100644 index 0000000..275b547 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/PropertiesSyncEntity.java @@ -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 fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId, + int limit, User user) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + if (user != null) { + permissionService.applyPropertyRowFilter(wrapper, user); + } + applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId); + applyOrderAndLimit(wrapper, limit); + List 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 toItem(Properyt row, long mt) { + boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1; + Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime(); + Map 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; + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntity.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntity.java new file mode 100644 index 0000000..2de0c17 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntity.java @@ -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 fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId, + int limit, User user); +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntityRegistry.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntityRegistry.java new file mode 100644 index 0000000..63b949b --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncEntityRegistry.java @@ -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 byName; + + public SyncEntityRegistry(List entities) { + Map 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); + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncQueryTimeoutHolder.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncQueryTimeoutHolder.java new file mode 100644 index 0000000..3a84666 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncQueryTimeoutHolder.java @@ -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 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(); + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncRow.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncRow.java new file mode 100644 index 0000000..9e2b443 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncRow.java @@ -0,0 +1,9 @@ +package me.iiv.agent.mcp.sync; + +import java.util.Map; + +/** + * 一页中的一行：已映射的对外字段 + 引擎推进游标所需的 (生效修改时间 epoch 秒, id)。 + */ +public record SyncRow(Map item, long modifyTs, long id) { +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncTokenCodec.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncTokenCodec.java new file mode 100644 index 0000000..12c99e1 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/SyncTokenCodec.java @@ -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）由令牌自身钉住。 + *
    + *
  • anchor 令牌（typ=r）：一轮同步的锚点，含 from/cutoff/snap，页间原样回传；
    • + *
  • handoff 令牌（typ=n）：一轮最后一页签发的 next_sync_token，只含 from（=上轮 cutoff）， + * 下一轮首次调用时由服务端钉新的 cutoff 与 snapshot；
    • + *
  • cursor（typ=c）：页内位置 (modify_time, id)，绑定 snapshot，跨快照使用判过期。
    • + *
+ * 签名/结构/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) { + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/UsersSyncEntity.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/UsersSyncEntity.java new file mode 100644 index 0000000..3cf4943 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/sync/UsersSyncEntity.java @@ -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 fetchPage(Long fromTs, long cutoffTs, Long cursorMt, Long cursorId, + int limit, User user) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + applyWindow(wrapper, fromTs, cutoffTs, cursorMt, cursorId); + applyOrderAndLimit(wrapper, limit); + List 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 toItem(User row, long mt) { + boolean deleted = row.getSoftDelete() != null && row.getSoftDelete() == 1; + Date effective = row.getModifyTime() != null ? row.getModifyTime() : row.getCreateTime(); + Map 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; + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ChangesTools.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ChangesTools.java new file mode 100644 index 0000000..e14db26 --- /dev/null +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ChangesTools.java @@ -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 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 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 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); + } +} diff --git a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ContractTools.java b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ContractTools.java index 2b93e42..3d8b838 100644 --- a/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ContractTools.java +++ b/xiaoqu-agent-service/src/main/java/me/iiv/agent/mcp/tool/ContractTools.java @@ -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 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 customerIds = customerIdsOf(contracts); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd"); List> 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 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 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 files = contractFileMapper.selectList(new LambdaQueryWrapper() .eq(ContractFile::getSoftDelete, 0) @@ -273,11 +284,22 @@ public class ContractTools implements ToolObject { private Map toRecord(ContractNew contract, Map agentNames, Map creatorNames, + Map customerIds, SimpleDateFormat sdf, SimpleDateFormat df) { Map 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 customerIdsOf(List contracts) { + List 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 map = new LinkedHashMap<>(); + for (Properyt property : properytMapper.selectBatchIds(propertyIds)) { + map.put(property.getId(), property.getCustomerId()); + } + return map; + } + private Map agentNamesOf(ContractNew contract) { List agentIds = Stream.of(contract.getContractAgentId(), contract.getCustomerAgentId()) .filter(Objects::nonNull).distinct().collect(Collectors.toList()); diff --git a/xiaoqu-agent-service/src/main/resources/application.yml b/xiaoqu-agent-service/src/main/resources/application.yml index 17a69e6..077e7ac 100644 --- a/xiaoqu-agent-service/src/main/resources/application.yml +++ b/xiaoqu-agent-service/src/main/resources/application.yml @@ -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: diff --git a/xiaoqu-agent-service/src/main/resources/db/migration/V5__tool_description_length.sql b/xiaoqu-agent-service/src/main/resources/db/migration/V5__tool_description_length.sql new file mode 100644 index 0000000..846950e --- /dev/null +++ b/xiaoqu-agent-service/src/main/resources/db/migration/V5__tool_description_length.sql @@ -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 '工具描述'; diff --git a/xiaoqu-agent-service/src/main/resources/db/migration/checksums.sha256 b/xiaoqu-agent-service/src/main/resources/db/migration/checksums.sha256 index 7a93e1c..8668fd0 100644 --- a/xiaoqu-agent-service/src/main/resources/db/migration/checksums.sha256 +++ b/xiaoqu-agent-service/src/main/resources/db/migration/checksums.sha256 @@ -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 diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/FlywayMigrationTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/FlywayMigrationTest.java index 6cb7471..0ad271c 100644 --- a/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/FlywayMigrationTest.java +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/FlywayMigrationTest.java @@ -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)); } diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/SyncQueryTimeoutInterceptorTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/SyncQueryTimeoutInterceptorTest.java new file mode 100644 index 0000000..ee8fec9 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/config/SyncQueryTimeoutInterceptorTest.java @@ -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()); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/GuardedToolCallbackTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/GuardedToolCallbackTest.java index 02e9414..715dba5 100644 --- a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/GuardedToolCallbackTest.java +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/GuardedToolCallbackTest.java @@ -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); + } } diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/RateLimitServiceTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/RateLimitServiceTest.java new file mode 100644 index 0000000..9e178a5 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/RateLimitServiceTest.java @@ -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 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 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(); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/ChangesEngineTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/ChangesEngineTest.java new file mode 100644 index 0000000..f403727 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/ChangesEngineTest.java @@ -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 fromSeen = new ArrayList<>(); + final List cutoffSeen = new ArrayList<>(); + final List cursorMtSeen = new ArrayList<>(); + final List cursorIdSeen = new ArrayList<>(); + final List limitSeen = new ArrayList<>(); + List nextRows = List.of(); + + @Override + public String name() { + return "users"; + } + + @Override + public List 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 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 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 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 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 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 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 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 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 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()); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncEntitiesTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncEntitiesTest.java new file mode 100644 index 0000000..ab67251 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncEntitiesTest.java @@ -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 rows = entity.fetchPage(100L, 2000L, 150L, 5L, 3, null); + + ArgumentCaptor> 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 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 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 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 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"); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncTokenCodecTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncTokenCodecTest.java new file mode 100644 index 0000000..7363b24 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/sync/SyncTokenCodecTest.java @@ -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"); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ChangesToolsTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ChangesToolsTest.java new file mode 100644 index 0000000..3ec4783 --- /dev/null +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ChangesToolsTest.java @@ -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 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); + } +} diff --git a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ContractToolsTest.java b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ContractToolsTest.java index 19cf6d6..8c6cc9d 100644 --- a/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ContractToolsTest.java +++ b/xiaoqu-agent-service/src/test/java/me/iiv/agent/mcp/tool/ContractToolsTest.java @@ -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 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> files = (List>) 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; } }